feat(relay-loop-rewrite): 依 DESIGN-03「relay 與迴圈」改寫 outbox relay（搶列→交易外整批送出→小交易條件式回寫）、迴圈小交易與滿批續做、broker 逾時與關機設定

api/src/main/java/com/cdp/common/fault/FaultPoint.java

@@ -6,6 +6,8 @@ public enum FaultPoint {
     RELAY_BEFORE_PUBLISH,
     /** broker 已 confirm、標 PUBLISHED 之前：模擬重複推送。 */
     RELAY_AFTER_CONFIRM,
+    /** 整批 confirm 等完、條件式回寫之前：測試用來製造「遲到的 confirm」（lease 已被另一輪接手）。 */
+    RELAY_BEFORE_WRITEBACK,
     /** consumer 交易已提交、ack 之前：模擬「已推送、consumer 未 ack」。 */
     CONSUMER_AFTER_COMMIT
 }

api/src/main/java/com/cdp/delivery/DeliveryConsumer.java

@@ -24,7 +24,7 @@
 import com.cdp.engine.node.Effect;
 
 /**
- * 冪等 consumer（INIT-01 紅線 3）。單一交易：insertIfAbsentAndLock → 已 SENT/CANCELLED 即結束 → KillSwitch →
+ * 冪等 consumer（INIT-01 紅線 3）。單一交易：insertIfAbsentAndLock → 已 SENT/CANCELLED 即結束 → outbox 列 CANCELLED 或 KillSwitch → CANCELLED →
  * EffectExecutor → 提交 → FaultPoint.CONSUMER_AFTER_COMMIT → ack。無法解析 → DLQ；DB 不可用 →
  * ImmediateRequeueAmqpException。
  */
@@ -125,7 +125,8 @@ private void handle(String key) {
         if (task.status() != DeliveryTaskStatus.PENDING) {
             return; // 重複投遞：已處理過
         }
-        if (killSwitch.isStopped(entry.journeyId(), entry.journeyVersion())) {
+        if (entry.status() == OutboxStatus.CANCELLED
+                || killSwitch.isStopped(entry.journeyId(), entry.journeyVersion())) {
             tasks.update(
                     new DeliveryTask(
                             task.idempotencyKey(),

api/src/main/java/com/cdp/delivery/DeliveryListenerConfig.java

@@ -0,0 +1,17 @@
+package com.cdp.delivery;
+
+import org.springframework.amqp.rabbit.config.ContainerCustomizer;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/** listener container 關機等待（DELIVERY_SHUTDOWN_TIMEOUT）；Boot 沒有對應的 spring.rabbitmq 屬性。 */
+@Configuration(proxyBeanMethods = false)
+public class DeliveryListenerConfig {
+
+    @Bean
+    ContainerCustomizer<SimpleMessageListenerContainer> deliveryShutdownCustomizer(
+            DeliveryProperties properties) {
+        return container -> container.setShutdownTimeout(properties.shutdownTimeout().toMillis());
+    }
+}

api/src/main/java/com/cdp/delivery/DeliveryProperties.java

@@ -4,6 +4,9 @@
 
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
-/** cdp.delivery.*：consumer 暫時性 DB 例外重排前的等待（DELIVERY_REQUEUE_DELAY）。 */
+/**
+ * cdp.delivery.*：requeueDelay = consumer 暫時性 DB 例外重排前的等待（DELIVERY_REQUEUE_DELAY）；shutdownTimeout =
+ * listener container 關機時等在途訊息處理完的上限（DELIVERY_SHUTDOWN_TIMEOUT）。
+ */
 @ConfigurationProperties("cdp.delivery")
-public record DeliveryProperties(Duration requeueDelay) {}
+public record DeliveryProperties(Duration requeueDelay, Duration shutdownTimeout) {}

api/src/main/java/com/cdp/delivery/LoopProperties.java

@@ -0,0 +1,9 @@
+package com.cdp.delivery;
+
+import java.time.Duration;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/** cdp.loop.*：三個輪詢迴圈共用，滿批續做的時間上限（CDP_LOOP_MAX_DRAIN）。 */
+@ConfigurationProperties("cdp.loop")
+public record LoopProperties(Duration maxDrain) {}

api/src/main/java/com/cdp/delivery/LoopStartupValidator.java

@@ -0,0 +1,42 @@
+package com.cdp.delivery;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.core.env.Environment;
+import org.springframework.stereotype.Component;
+
+import com.cdp.delivery.relay.RelayProperties;
+
+/** 啟動驗證（DESIGN-03「設定」）：lease 必須大於 confirm 逾時；排程執行緒池不得小於輪詢迴圈數。不符即拒絕啟動。 */
+@Component
+public class LoopStartupValidator implements InitializingBean {
+
+    /** 輪詢迴圈數：EventDispatcher、AdvanceLoop、OutboxRelay。加迴圈時（v0.3.0 JourneyStopSweeper）一併調整。 */
+    static final int LOOP_COUNT = 3;
+
+    private final RelayProperties relay;
+    private final Environment env;
+
+    public LoopStartupValidator(RelayProperties relay, Environment env) {
+        this.relay = relay;
+        this.env = env;
+    }
+
+    @Override
+    public void afterPropertiesSet() {
+        if (relay.lease() == null
+                || relay.confirmTimeout() == null
+                || relay.lease().compareTo(relay.confirmTimeout()) <= 0) {
+            throw new IllegalStateException(
+                    "cdp.relay.lease (OUTBOX_RELAY_LEASE) must be greater than"
+                            + " cdp.relay.confirm-timeout (OUTBOX_RELAY_CONFIRM_TIMEOUT): lease="
+                            + relay.lease() + ", confirm-timeout=" + relay.confirmTimeout());
+        }
+        int pool = env.getProperty("spring.task.scheduling.pool.size", Integer.class, 1);
+        if (pool < LOOP_COUNT) {
+            throw new IllegalStateException(
+                    "spring.task.scheduling.pool.size (CDP_SCHEDULER_POOL_SIZE) = " + pool
+                            + " is smaller than the number of polling loops (" + LOOP_COUNT
+                            + "); scheduler pool must be >= loop count");
+        }
+    }
+}

api/src/main/java/com/cdp/delivery/OutboxStore.java

@@ -4,19 +4,40 @@
 import java.util.List;
 import java.util.Optional;
 
-/** 表 effect_outbox。insert 由 StateStore.save 在同一交易內呼叫。 */
+/**
+ * 表 effect_outbox。insert 由 StateStore.save 在同一交易內呼叫。relay 的 lease 與 fencing 見 spec 07 §12「relay」：
+ * available_at 兼作 lease 與 fencing 值，回寫只在 {@code status='PENDING' AND available_at = leaseUntil} 時生效。
+ */
 public interface OutboxStore {
 
+    /** 回寫用的 fencing 值：搶列時取得的 lease_until。 */
+    record Lease(String idempotencyKey, Instant leaseUntil) {}
+
     /** idempotency_key 已存在則略過（重放）。 */
     void insertAll(List<OutboxEntry> entries);
 
-    /** status=PENDING 且 available_at <= now，FOR UPDATE SKIP LOCKED；須在交易內呼叫。 */
-    List<OutboxEntry> lockDue(Instant now, int limit);
+    /**
+     * 搶列（單一敘述，不需外層交易）：status=PENDING 且 available_at <= now，FOR UPDATE SKIP LOCKED，同時把
+     * available_at 設為 leaseUntil 並回傳列（回傳列的 availableAt 即 fencing 值）。
+     */
+    List<OutboxEntry> claimDue(Instant now, Instant leaseUntil, int limit);
+
+    /** 標 PUBLISHED、publish_count+1；僅在 status=PENDING 且 available_at=leaseUntil 時生效，回傳是否生效。 */
+    boolean markPublished(String idempotencyKey, Instant leaseUntil, Instant publishedAt);
+
+    /** 退回 PENDING 並延後到 availableAt；同樣的條件式回寫，回傳是否生效。 */
+    boolean reschedule(String idempotencyKey, Instant leaseUntil, Instant availableAt);
+
+    /** {@link #markPublished} 的批次版（一次 batchUpdate），回傳與輸入同序的各列影響筆數。 */
+    int[] markPublishedAll(List<Lease> leases, Instant publishedAt);
+
+    /** {@link #reschedule} 的批次版，回傳與輸入同序的各列影響筆數。 */
+    int[] rescheduleAll(List<Lease> leases, Instant availableAt);
 
-    void markPublished(String idempotencyKey, Instant publishedAt);
+    /** PENDING → CANCELLED（不論是否被 relay 搶走）；回傳取消筆數。 */
+    int cancelPendingByEnrollment(String enrollmentId);
 
-    /** 退回 PENDING 並延後（重試退避，INIT-01 紅線 1）；v0.1.0 Mock 不會用到。 */
-    void reschedule(String idempotencyKey, Instant availableAt);
+    int cancelPendingByJourney(String journeyId);
 
     Optional<OutboxEntry> find(String idempotencyKey);
 }