From 111f981b36e08cd7f82f2811b998c174d8a1fa88 Mon Sep 17 00:00:00 2001 From: Marcus Date: Thu, 22 Oct 2026 01:36:05 +0800 Subject: [PATCH 1/3] =?UTF-8?q?test(relay-loop-rewrite):=20relay=20?= =?UTF-8?q?=E8=88=87=E8=BF=B4=E5=9C=88=E6=94=B9=E5=AF=AB=E9=A9=97=E6=94=B6?= =?UTF-8?q?=E6=B8=AC=E8=A9=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 5.5 --- .../relay-loop-rewrite/RelayLoopBase.java | 159 +++++++++ .../RelayLoopTestSupport.java | 149 +++++++++ .../RelayRewriteAcceptanceTest.java | 311 ++++++++++++++++++ .../RelayStartupValidationAcceptanceTest.java | 104 ++++++ .../RelayStuckTimersAcceptanceTest.java | 87 +++++ 5 files changed, 810 insertions(+) create mode 100644 api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayLoopBase.java create mode 100644 api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayLoopTestSupport.java create mode 100644 api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayRewriteAcceptanceTest.java create mode 100644 api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayStartupValidationAcceptanceTest.java create mode 100644 api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayStuckTimersAcceptanceTest.java diff --git a/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayLoopBase.java b/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayLoopBase.java new file mode 100644 index 0000000..254bf97 --- /dev/null +++ b/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayLoopBase.java @@ -0,0 +1,159 @@ +package com.cdp.acceptance.relayloop; + +import static org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.csrf; +import static org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.user; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +import java.sql.Timestamp; +import java.time.Duration; +import java.time.Instant; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.http.MediaType; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.test.web.servlet.ResultActions; + +import com.cdp.acceptance.engine.EngineTestSupport; +import com.cdp.acceptance.relayloop.RelayLoopTestSupport.MutableClock; + +import tools.jackson.databind.json.JsonMapper; + +/** 共用 helper：建旅程（等待 waitIso 後發訊息）、送事件、查 outbox／delivery_task／Mock。 */ +abstract class RelayLoopBase { + + static final Duration HOUR_PLUS = Duration.ofHours(1).plusSeconds(1); + + @Autowired MockMvc mvc; + @Autowired JsonMapper json; + @Autowired JdbcTemplate jdbc; + @Autowired MutableClock clock; + + static String ulid() { + return EngineTestSupport.ulid(); + } + + ResultActions postJson(String path, String key, Object body) throws Exception { + return mvc.perform( + post(path) + .header("Authorization", "Bearer " + key) + .with(user("qa").roles("ADMIN", "INGEST")) + .with(csrf()) + .contentType(MediaType.APPLICATION_JSON) + .content(json.writeValueAsString(body))); + } + + private static Map node( + String id, String type, Map config, Map outs) { + Map n = new LinkedHashMap<>(); + n.put("id", id); + n.put("type", type); + n.put("config", config); + n.put("outputs", outs); + return n; + } + + /** 建立「事件 → 等待 waitIso → 發訊息」旅程，回傳事件名稱。 */ + String createJourney(String waitIso) throws Exception { + String eventName = "evt_" + ulid().toLowerCase(); + Map j = new LinkedHashMap<>(); + j.put("name", "relay-loop"); + j.put("settings", Map.of()); + j.put("entry_node", "start_1"); + j.put( + "nodes", + List.of( + node("start_1", "StartNode", Map.of(), Map.of("next", "src_1")), + node( + "src_1", + "EventSourceNode", + Map.of("event_name", eventName), + Map.of("next", "wait_1")), + node( + "wait_1", + "WaitDelayNode", + Map.of("duration", waitIso), + Map.of("next", "send_1")), + node( + "send_1", + "SendMessageNode", + Map.of("template_id", "tpl_a", "channel_id", "ch_a"), + Map.of("sent", "end_1")), + node("end_1", "EndNode", Map.of(), Map.of()))); + postJson("/api/v1/journeys", "test-admin-key", j).andExpect(status().isCreated()); + return eventName; + } + + /** 送一個事件，回傳 profile id。 */ + String postEvent(String eventName) throws Exception { + String profile = ulid(); + Map e = new LinkedHashMap<>(); + e.put("event_id", ulid()); + e.put("source_type", "CUSTOMER_BEHAVIOR"); + e.put("source_id", "web-sdk@2.1.0"); + e.put("name", eventName); + e.put("occurred_at", clock.instant().minusSeconds(60).toString()); + e.put("identity", Map.of("profile_id", profile)); + e.put("properties", Map.of("cart_id", "CART-1")); + postJson("/api/v1/events", "test-ingest-key", e).andExpect(status().isAccepted()); + return profile; + } + + String enrollmentOf(String profile) { + List r = + jdbc.queryForList( + "select enrollment_id from enrollment_record where profile_id = ?", + String.class, + profile); + return r.isEmpty() ? null : r.get(0); + } + + String outboxKeyOf(String enrollmentId) { + List r = + jdbc.queryForList( + "select idempotency_key from effect_outbox where enrollment_id = ?", + String.class, + enrollmentId); + return r.isEmpty() ? null : r.get(0); + } + + String outboxStatus(String key) { + return jdbc.queryForObject( + "select status from effect_outbox where idempotency_key = ?", String.class, key); + } + + int publishCount(String key) { + return jdbc.queryForObject( + "select publish_count from effect_outbox where idempotency_key = ?", + Integer.class, + key); + } + + Instant availableAt(String key) { + return jdbc.queryForObject( + "select available_at from effect_outbox where idempotency_key = ?", + Timestamp.class, + key) + .toInstant(); + } + + String taskStatus(String key) { + List r = + jdbc.queryForList( + "select status from delivery_task where idempotency_key = ?", + String.class, + key); + return r.isEmpty() ? null : r.get(0); + } + + int mockCount(String key) { + return jdbc.queryForObject( + "select count(*) from mock_channel_message where idempotency_key = ?", + Integer.class, + key); + } +} diff --git a/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayLoopTestSupport.java b/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayLoopTestSupport.java new file mode 100644 index 0000000..756bca5 --- /dev/null +++ b/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayLoopTestSupport.java @@ -0,0 +1,149 @@ +package com.cdp.acceptance.relayloop; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.boot.testcontainers.service.connection.ServiceConnection; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Primary; +import org.testcontainers.postgresql.PostgreSQLContainer; +import org.testcontainers.rabbitmq.RabbitMQContainer; + +import com.cdp.common.fault.FaultInjector; +import com.cdp.common.fault.FaultPoint; + +/** + * relay-loop-rewrite 驗收接縫：Testcontainers（真實 RabbitTopology）、可調 Clock、 + * 可編程 FaultInjector（丟例外，或一次性阻塞到測試放行，用來製造「遲到的 confirm」）。 + */ +@TestConfiguration(proxyBeanMethods = false) +public class RelayLoopTestSupport { + + public static final Instant BASE = Instant.parse("2026-10-18T02:00:00Z"); + + @Bean + @ServiceConnection + PostgreSQLContainer postgres() { + return new PostgreSQLContainer("postgres:18-alpine"); + } + + @Bean + @ServiceConnection + RabbitMQContainer rabbit() { + return new RabbitMQContainer("rabbitmq:4.3-management-alpine"); + } + + @Bean + @Primary + MutableClock testClock() { + return new MutableClock(BASE); + } + + @Bean + @Primary + ProgrammableFaultInjector testFaults() { + return new ProgrammableFaultInjector(); + } + + public static final class Gate { + final CountDownLatch arrived = new CountDownLatch(1); + final CountDownLatch release = new CountDownLatch(1); + + public boolean awaitArrived(Duration d) throws InterruptedException { + return arrived.await(d.toMillis(), TimeUnit.MILLISECONDS); + } + + public void release() { + release.countDown(); + } + } + + public static final class ProgrammableFaultInjector implements FaultInjector { + private final Map armed = new ConcurrentHashMap<>(); + private final Map gates = new ConcurrentHashMap<>(); + + /** 對指定 (點, key) 丟例外，直到 disarm。 */ + public void arm(FaultPoint point, String key) { + armed.put(point + "|" + key, Boolean.TRUE); + } + + public void disarm(FaultPoint point, String key) { + armed.remove(point + "|" + key); + } + + /** 第一個到達 (點, key) 的執行緒阻塞到 {@link Gate#release()}；之後的到達者不受影響（一次性）。 */ + public Gate blockOnce(FaultPoint point, String key) { + Gate g = new Gate(); + gates.put(point + "|" + key, g); + return g; + } + + public void reset() { + armed.clear(); + gates.values().forEach(Gate::release); + gates.clear(); + } + + @Override + public void at(FaultPoint point, String key) { + String id = point + "|" + key; + Gate g = gates.remove(id); + if (g != null) { + g.arrived.countDown(); + try { + g.release.await(60, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + if (armed.containsKey(id)) { + throw new InjectedFault(id); + } + } + } + + public static final class InjectedFault extends RuntimeException { + InjectedFault(String message) { + super(message); + } + } + + public static final class MutableClock extends Clock { + private volatile Instant now; + + MutableClock(Instant start) { + this.now = start; + } + + public void reset() { + now = BASE; + } + + public void advance(Duration d) { + now = now.plus(d); + } + + @Override + public ZoneId getZone() { + return ZoneOffset.UTC; + } + + @Override + public Clock withZone(ZoneId zone) { + return this; + } + + @Override + public Instant instant() { + return now; + } + } +} diff --git a/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayRewriteAcceptanceTest.java b/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayRewriteAcceptanceTest.java new file mode 100644 index 0000000..4e990b9 --- /dev/null +++ b/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayRewriteAcceptanceTest.java @@ -0,0 +1,311 @@ +package com.cdp.acceptance.relayloop; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.amqp.core.AmqpAdmin; +import org.springframework.amqp.core.Binding; +import org.springframework.amqp.core.Binding.DestinationType; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.webmvc.test.autoconfigure.AutoConfigureMockMvc; +import org.springframework.context.annotation.Import; +import org.springframework.test.context.ActiveProfiles; + +import com.cdp.acceptance.relayloop.RelayLoopTestSupport.Gate; +import com.cdp.acceptance.relayloop.RelayLoopTestSupport.ProgrammableFaultInjector; +import com.cdp.common.fault.FaultPoint; +import com.cdp.delivery.RabbitTopology; +import com.cdp.delivery.relay.OutboxRelay; +import com.cdp.engine.runtime.AdvanceLoop; +import com.cdp.ingest.EventDispatcher; + +/** + * relay-loop-rewrite 驗收（DESIGN-03「relay 與迴圈」、spec 07 §12「relay」）： + * 兩處中斷恰好一則、lease 過期重發恰好一則、遲到 confirm 不覆寫、3 倍批量一個 tick 內清空、 + * consumer 遇 outbox CANCELLED → delivery_task CANCELLED。 + * + *

設定名稱（依既有 OUTBOX_RELAY_RETRY_DELAY → cdp.relay.retry-delay 的對應慣例）： + * cdp.relay.confirm-timeout、cdp.relay.lease。時間用可控 Clock，lease 相對於 Clock。 + */ +@SpringBootTest( + properties = { + "cdp.engine.poll-interval=PT1H", + "cdp.relay.poll-interval=PT1H", + "cdp.engine.batch-size=5", + "cdp.relay.batch-size=5", + "cdp.relay.confirm-timeout=PT2S", + "cdp.relay.lease=PT10S", + "cdp.relay.retry-delay=PT30S", + "spring.task.scheduling.pool.size=4" + }) +@AutoConfigureMockMvc +@ActiveProfiles("test") +@Import(RelayLoopTestSupport.class) +class RelayRewriteAcceptanceTest extends RelayLoopBase { + + private static final Duration WAIT = Duration.ofSeconds(30); + private static final Duration LEASE = Duration.ofSeconds(10); + private static final Duration RETRY_DELAY = Duration.ofSeconds(30); + + @Autowired EventDispatcher dispatcher; + @Autowired AdvanceLoop advanceLoop; + @Autowired OutboxRelay relay; + @Autowired RabbitTemplate rabbit; + @Autowired AmqpAdmin amqpAdmin; + @Autowired ProgrammableFaultInjector faults; + + private final ExecutorService pool = Executors.newCachedThreadPool(); + + @BeforeEach + void reset() { + faults.reset(); + clock.reset(); + // 前一個測試遺留的待推列不得影響本測試 + jdbc.update("update effect_outbox set status = 'PUBLISHED' where status = 'PENDING'"); + } + + @AfterEach + void cleanup() { + faults.reset(); + pool.shutdownNow(); + } + + // ---------- helpers ---------- + + private record Scheduled(String enrollmentId, String key) {} + + /** 建旅程 → 事件 → 派發 → 快轉 → 喚醒；回傳 outbox 中等待推送的那一列。 */ + private Scheduled scheduleSend() throws Exception { + String eventName = createJourney("PT1H"); + String profile = postEvent(eventName); + dispatcher.runOnce(); + String eid = enrollmentOf(profile); + assertThat(eid).as("事件應已派發建出 Enrollment").isNotNull(); + clock.advance(HOUR_PLUS); + advanceLoop.runOnce(); + String key = outboxKeyOf(eid); + assertThat(key).as("到期後應寫入 outbox").isNotNull(); + return new Scheduled(eid, key); + } + + /** relay 被注入的故障打斷時，不論例外是否傳出 runOnce，測試都只看資料庫結果。 */ + private void runRelayIgnoringFaults() { + try { + relay.runOnce(); + } catch (RuntimeException ignored) { + // 注入的中斷 + } + } + + private long queueDepth() { + var info = amqpAdmin.getQueueInfo(RabbitTopology.SEND_MESSAGE_QUEUE); + return info == null ? -1L : info.getMessageCount(); + } + + private void assertSentExactlyOnce(Scheduled s) throws Exception { + await().atMost(WAIT).until(() -> "SENT".equals(taskStatus(s.key()))); + await().atMost(WAIT).until(() -> queueDepth() == 0); + Thread.sleep(1500); + assertThat(mockCount(s.key())).as("恰好一則").isEqualTo(1); + assertThat( + jdbc.queryForObject( + "select count(*) from delivery_task where idempotency_key = ?", + Integer.class, + s.key())) + .isEqualTo(1); + } + + private static Binding sendMessageBinding() { + return new Binding( + RabbitTopology.SEND_MESSAGE_QUEUE, + DestinationType.QUEUE, + RabbitTopology.EXCHANGE, + RabbitTopology.SEND_MESSAGE_ROUTING_KEY, + null); + } + + // ---------- 沿用的兩處中斷：恰好一則 ---------- + + @Test + void interruptBeforePublish_thenRestart_exactlyOnce() throws Exception { + Scheduled s = scheduleSend(); + faults.arm(FaultPoint.RELAY_BEFORE_PUBLISH, s.key()); + + runRelayIgnoringFaults(); + + assertThat(outboxStatus(s.key())).as("未推送，維持 PENDING").isEqualTo("PENDING"); + assertThat(publishCount(s.key())).isZero(); + assertThat(taskStatus(s.key())).isNull(); + assertThat(mockCount(s.key())).isZero(); + + // 重啟後（lease 未過期前不被搶；過期後重搶重推） + faults.disarm(FaultPoint.RELAY_BEFORE_PUBLISH, s.key()); + clock.advance(LEASE.plusSeconds(1)); + relay.runOnce(); + + assertThat(outboxStatus(s.key())).isEqualTo("PUBLISHED"); + assertThat(publishCount(s.key())).isEqualTo(1); + assertSentExactlyOnce(s); + } + + @Test + void interruptAfterConfirm_thenRestart_exactlyOnce() throws Exception { + Scheduled s = scheduleSend(); + faults.arm(FaultPoint.RELAY_AFTER_CONFIRM, s.key()); + + runRelayIgnoringFaults(); + + // 訊息已進 broker 並被 consumer 處理，但列尚未標 PUBLISHED + await().atMost(WAIT).until(() -> "SENT".equals(taskStatus(s.key()))); + assertThat(outboxStatus(s.key())).isEqualTo("PENDING"); + assertThat(publishCount(s.key())).isZero(); + + faults.disarm(FaultPoint.RELAY_AFTER_CONFIRM, s.key()); + clock.advance(LEASE.plusSeconds(1)); + relay.runOnce(); + + assertThat(outboxStatus(s.key())).isEqualTo("PUBLISHED"); + assertThat(publishCount(s.key())).isEqualTo(1); + assertSentExactlyOnce(s); + } + + // ---------- lease 過期重發恰好一則 ---------- + + @Test + void leaseExpiredResend_isAbsorbedByConsumer_exactlyOne_publishCountOne() throws Exception { + Scheduled s = scheduleSend(); + faults.arm(FaultPoint.RELAY_AFTER_CONFIRM, s.key()); + runRelayIgnoringFaults(); + await().atMost(WAIT).until(() -> "SENT".equals(taskStatus(s.key()))); + + // lease 未過期：另一輪 relay 不得重搶 + faults.disarm(FaultPoint.RELAY_AFTER_CONFIRM, s.key()); + relay.runOnce(); + assertThat(outboxStatus(s.key())).as("lease 內不重搶").isEqualTo("PENDING"); + assertThat(publishCount(s.key())).isZero(); + + // 快轉超過 lease：重搶並重發第二則訊息，由 consumer 冪等吸收 + clock.advance(LEASE.plusSeconds(1)); + relay.runOnce(); + + assertThat(outboxStatus(s.key())).isEqualTo("PUBLISHED"); + assertThat(publishCount(s.key())).as("只在標 PUBLISHED 時遞增").isEqualTo(1); + assertSentExactlyOnce(s); + } + + // ---------- 遲到 confirm 不覆寫 ---------- + + @Test + void lateConfirm_afterOtherRoundPublished_doesNotOverwrite_publishCountOne() throws Exception { + Scheduled s = scheduleSend(); + Gate gate = faults.blockOnce(FaultPoint.RELAY_BEFORE_WRITEBACK, s.key()); + + Future slow = pool.submit(this::runRelayIgnoringFaults); + assertThat(gate.awaitArrived(WAIT)).as("第一輪已 confirm、卡在回寫前").isTrue(); + + // lease 過期後另一輪重搶、發送、標 PUBLISHED + clock.advance(LEASE.plusSeconds(1)); + relay.runOnce(); + assertThat(outboxStatus(s.key())).isEqualTo("PUBLISHED"); + assertThat(publishCount(s.key())).isEqualTo(1); + + // 第一輪的遲到回寫：條件（status=PENDING 且 available_at = 自己的 lease）不成立，不得覆寫 + gate.release(); + slow.get(WAIT.toSeconds(), TimeUnit.SECONDS); + + assertThat(outboxStatus(s.key())).isEqualTo("PUBLISHED"); + assertThat(publishCount(s.key())).as("遲到 confirm 不得再加一").isEqualTo(1); + assertSentExactlyOnce(s); + } + + @Test + void lateConfirm_afterOtherRoundRescheduled_isFenced_doesNotMarkPublished() throws Exception { + Scheduled s = scheduleSend(); + Gate gate = faults.blockOnce(FaultPoint.RELAY_BEFORE_WRITEBACK, s.key()); + + Future slow = pool.submit(this::runRelayIgnoringFaults); + assertThat(gate.awaitArrived(WAIT)).isTrue(); + + // 第二輪重搶後發佈失敗（沒有 binding → returned）→ 改排 now + retry-delay + clock.advance(LEASE.plusSeconds(1)); + var expectedNext = clock.instant().plus(RETRY_DELAY); + amqpAdmin.removeBinding(sendMessageBinding()); + try { + relay.runOnce(); + } finally { + amqpAdmin.declareBinding(sendMessageBinding()); + } + assertThat(outboxStatus(s.key())).isEqualTo("PENDING"); + assertThat(availableAt(s.key())).isEqualTo(expectedNext); + + // 第一輪的遲到成功回寫：available_at 已不是它的 lease，必須被擋下 + gate.release(); + slow.get(WAIT.toSeconds(), TimeUnit.SECONDS); + + assertThat(outboxStatus(s.key())).as("fencing：不得被舊 lease 標成 PUBLISHED").isEqualTo("PENDING"); + assertThat(publishCount(s.key())).isZero(); + assertThat(availableAt(s.key())).as("也不得被舊結果改寫排程").isEqualTo(expectedNext); + } + + // ---------- 3 倍批量到期一個 tick 內清空 ---------- + + @Test + void threeTimesBatchDue_drainedWithinOneTickPerLoop() throws Exception { + String eventName = createJourney("PT1H"); + List profiles = new ArrayList<>(); + for (int i = 0; i < 15; i++) { // batch-size = 5 的 3 倍 + profiles.add(postEvent(eventName)); + } + + dispatcher.tick(); + for (String p : profiles) { + assertThat(enrollmentOf(p)).as("EventDispatcher 一個 tick 內滿批續做，派發完 15 個事件").isNotNull(); + } + + clock.advance(HOUR_PLUS); + advanceLoop.tick(); + List keys = new ArrayList<>(); + for (String p : profiles) { + String key = outboxKeyOf(enrollmentOf(p)); + assertThat(key).as("AdvanceLoop 一個 tick 內清空 15 筆到期").isNotNull(); + keys.add(key); + } + + relay.tick(); + for (String k : keys) { + assertThat(outboxStatus(k)).as("relay 一個 tick 內清空 15 列").isEqualTo("PUBLISHED"); + assertThat(publishCount(k)).isEqualTo(1); + } + } + + // ---------- consumer 遇 outbox CANCELLED ---------- + + @Test + void consumer_outboxCancelled_marksDeliveryTaskCancelled_noMessageSent() throws Exception { + Scheduled s = scheduleSend(); + jdbc.update("update effect_outbox set status = 'CANCELLED' where idempotency_key = ?", s.key()); + + // 模擬取消前已在 MQ 的訊息 + rabbit.convertAndSend( + RabbitTopology.EXCHANGE, RabbitTopology.SEND_MESSAGE_ROUTING_KEY, s.key()); + + await().atMost(WAIT).until(() -> "CANCELLED".equals(taskStatus(s.key()))); + await().atMost(WAIT).until(() -> queueDepth() == 0); + Thread.sleep(500); + assertThat(mockCount(s.key())).as("已取消不得發送").isZero(); + assertThat(outboxStatus(s.key())).isEqualTo("CANCELLED"); + } +} diff --git a/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayStartupValidationAcceptanceTest.java b/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayStartupValidationAcceptanceTest.java new file mode 100644 index 0000000..a6cb743 --- /dev/null +++ b/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayStartupValidationAcceptanceTest.java @@ -0,0 +1,104 @@ +package com.cdp.acceptance.relayloop; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.util.LinkedHashMap; +import java.util.Map; + +import org.junit.jupiter.api.Test; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.context.ConfigurableApplicationContext; + +import com.cdp.CdpApplication; +import com.cdp.acceptance.security.SecurityTestSupport; + +/** + * 啟動驗證：OUTBOX_RELAY_LEASE（cdp.relay.lease）必須大於 OUTBOX_RELAY_CONFIRM_TIMEOUT + * （cdp.relay.confirm-timeout），否則拒絕啟動；排程執行緒池（spring.task.scheduling.pool.size）不得小於迴圈數。 + * 合法設定可啟動。以命令列參數覆蓋 application-test.yml。 + */ +class RelayStartupValidationAcceptanceTest { + + private static String[] args(String... overrides) { + Map merged = new LinkedHashMap<>(); + for (String a : + new String[] { + "--spring.rabbitmq.listener.simple.auto-startup=false", + "--cdp.engine.poll-interval=PT1H", + "--cdp.relay.poll-interval=PT1H", + "--cdp.relay.confirm-timeout=PT5S", + "--cdp.relay.lease=PT30S", + "--spring.task.scheduling.pool.size=4", + "--cdp.security.ingest-api-key=SECRET-aaaa-1111", + "--cdp.security.admin-api-key=SECRET-bbbb-2222", + "--cdp.security.support-username=agent", + "--cdp.security.support-password=SECRET-pppp-3333" + }) { + merged.put(a.substring(0, a.indexOf('=')), a); + } + for (String a : overrides) { + merged.put(a.substring(0, a.indexOf('=')), a); + } + return merged.values().toArray(String[]::new); + } + + private static SpringApplicationBuilder app() { + return new SpringApplicationBuilder(CdpApplication.class, SecurityTestSupport.class) + .profiles("test"); + } + + private void assertStartFails(String mustMention, String... overrides) { + String[] args = args(overrides); + ConfigurableApplicationContext[] started = new ConfigurableApplicationContext[1]; + try { + assertThatThrownBy(() -> started[0] = app().run(args)) + .satisfies( + t -> { + StringBuilder all = new StringBuilder(); + for (Throwable c = t; c != null; c = c.getCause()) { + all.append(c).append("\n").append(c.getMessage()).append("\n"); + } + assertThat(all.toString().toLowerCase()) + .as("cause 鏈訊息需點出設定：%s", all) + .contains(mustMention); + }); + } finally { + if (started[0] != null) { + started[0].close(); + } + } + } + + @Test + void leaseEqualToConfirmTimeout_failsStartup() { + assertStartFails("lease", "--cdp.relay.confirm-timeout=PT7S", "--cdp.relay.lease=PT7S"); + } + + @Test + void leaseShorterThanConfirmTimeout_failsStartup() { + assertStartFails("lease", "--cdp.relay.confirm-timeout=PT20S", "--cdp.relay.lease=PT3S"); + } + + @Test + void leaseLongerThanConfirmTimeout_starts() { + try (ConfigurableApplicationContext ctx = + app().run(args("--cdp.relay.confirm-timeout=PT5S", "--cdp.relay.lease=PT6S"))) { + assertThat(ctx.isRunning()).isTrue(); + } + } + + @Test + void schedulerPoolSmallerThanLoopCount_failsStartup() { + // 目前三個迴圈（EventDispatcher、AdvanceLoop、OutboxRelay）；池 = 1 不足 + assertStartFails("pool", "--spring.task.scheduling.pool.size=1"); + } + + @Test + void schedulerPoolAtLeastLoopCount_starts() { + try (ConfigurableApplicationContext ctx = + app().run(args("--spring.task.scheduling.pool.size=4"))) { + assertThat(ctx.isRunning()).isTrue(); + } + } +} diff --git a/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayStuckTimersAcceptanceTest.java b/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayStuckTimersAcceptanceTest.java new file mode 100644 index 0000000..d0747a3 --- /dev/null +++ b/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayStuckTimersAcceptanceTest.java @@ -0,0 +1,87 @@ +package com.cdp.acceptance.relayloop; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +import java.time.Duration; + +import org.junit.jupiter.api.Test; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.webmvc.test.autoconfigure.AutoConfigureMockMvc; +import org.springframework.context.annotation.Import; +import org.springframework.test.context.ActiveProfiles; +import org.testcontainers.DockerClientFactory; +import org.testcontainers.rabbitmq.RabbitMQContainer; + +import org.springframework.beans.factory.annotation.Autowired; + +/** + * relay 卡住（broker 無回應，等 confirm 逾時）時，計時器迴圈仍照常推進。 + * 排程迴圈全部開啟（真實排程），排程執行緒池 = 4；confirm 逾時拉長到 PT8S，確保 relay 的阻塞期涵蓋觀察窗。 + * 舊行為（單一排程執行緒）下，relay 阻塞期間 AdvanceLoop 不會被排到。 + */ +@SpringBootTest( + properties = { + "cdp.engine.poll-interval=PT0.2S", + "cdp.relay.poll-interval=PT0.2S", + "cdp.relay.confirm-timeout=PT8S", + "cdp.relay.lease=PT30S", + "spring.task.scheduling.pool.size=4" + }) +@AutoConfigureMockMvc +@ActiveProfiles("test") +@Import(RelayLoopTestSupport.class) +class RelayStuckTimersAcceptanceTest extends RelayLoopBase { + + @Autowired RabbitMQContainer rabbitContainer; + + private static final Duration WAIT = Duration.ofSeconds(30); + + private String enrollOne(String eventName) throws Exception { + String profile = postEvent(eventName); + await().atMost(WAIT).until(() -> enrollmentOf(profile) != null); + return enrollmentOf(profile); + } + + @Test + void relayBlockedOnConfirm_timersStillAdvance() throws Exception { + // 三個旅程：等 1／2／3 小時；同一時間點全部建好 Enrollment + String warmEvent = createJourney("PT1H"); + String slowEvent = createJourney("PT2H"); + String probeEvent = createJourney("PT3H"); + String warm = enrollOne(warmEvent); + String slow = enrollOne(slowEvent); + String probe = enrollOne(probeEvent); + + // 暖機：先走通一次完整發送，讓連線與 channel 建立 + clock.advance(Duration.ofHours(1).plusSeconds(1)); + await().atMost(WAIT).until(() -> outboxKeyOf(warm) != null); + String warmKey = outboxKeyOf(warm); + await().atMost(WAIT).until(() -> "SENT".equals(taskStatus(warmKey))); + + var docker = DockerClientFactory.instance().client(); + String containerId = rabbitContainer.getContainerId(); + docker.pauseContainerCmd(containerId).exec(); + try { + // 第二筆到期 → 寫入 outbox → relay 搶到後卡在等 confirm + clock.advance(Duration.ofHours(1)); + await().atMost(Duration.ofSeconds(5)).until(() -> outboxKeyOf(slow) != null); + String slowKey = outboxKeyOf(slow); + // 給 relay 時間進入等待 + Thread.sleep(1000); + assertThat(outboxStatus(slowKey)).as("relay 還卡在等 confirm，列未回寫為 PUBLISHED").isEqualTo("PENDING"); + + // 第三筆到期：relay 仍卡住（逾時 8 秒），計時器必須在 4 秒內照常推進 + long t0 = System.nanoTime(); + clock.advance(Duration.ofHours(1)); + await().atMost(Duration.ofSeconds(4)).until(() -> outboxKeyOf(probe) != null); + long elapsedMs = (System.nanoTime() - t0) / 1_000_000; + assertThat(elapsedMs).as("relay 阻塞期間計時器仍在數秒內推進").isLessThan(4_000); + assertThat(outboxStatus(slowKey)) + .as("觀察時 relay 仍未完成（確認測試窗口確實落在阻塞期內）") + .isEqualTo("PENDING"); + } finally { + docker.unpauseContainerCmd(containerId).exec(); + } + } +} From 526ab1098cd141f6b8e755731102a5456b9926b6 Mon Sep 17 00:00:00 2001 From: Marcus Date: Thu, 22 Oct 2026 02:10:43 +0800 Subject: [PATCH 2/3] =?UTF-8?q?feat(relay-loop-rewrite):=20relay=20?= =?UTF-8?q?=E6=90=B6=E5=88=97=EF=BC=8Flease=EF=BC=8F=E6=A2=9D=E4=BB=B6?= =?UTF-8?q?=E5=BC=8F=E5=9B=9E=E5=AF=AB=E3=80=81=E8=BF=B4=E5=9C=88=E5=B0=8F?= =?UTF-8?q?=E4=BA=A4=E6=98=93=E8=88=87=E6=BB=BF=E6=89=B9=E7=BA=8C=E5=81=9A?= =?UTF-8?q?=E3=80=81=E6=8E=92=E7=A8=8B=E6=B1=A0=E8=88=87=20broker=20?= =?UTF-8?q?=E8=A8=AD=E5=AE=9A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_0134QMUz6NTfqU8Yrh3BJekj --- .../java/com/cdp/common/fault/FaultPoint.java | 2 + .../com/cdp/delivery/DeliveryConsumer.java | 5 +- .../cdp/delivery/DeliveryListenerConfig.java | 17 ++ .../com/cdp/delivery/DeliveryProperties.java | 7 +- .../java/com/cdp/delivery/LoopProperties.java | 9 ++ .../cdp/delivery/LoopStartupValidator.java | 42 +++++ .../java/com/cdp/delivery/OutboxStore.java | 33 +++- .../com/cdp/delivery/relay/OutboxRelay.java | 153 ++++++++++++++---- .../cdp/delivery/relay/RelayProperties.java | 12 +- .../com/cdp/engine/runtime/AdvanceLoop.java | 57 +++++-- .../java/com/cdp/ingest/EventDispatcher.java | 64 ++++++-- .../com/cdp/persistence/JdbcOutboxStore.java | 93 +++++++++-- api/src/main/resources/application.yml | 12 ++ .../cdp/delivery/DeliveryConsumerTest.java | 2 +- .../ingest/EventDispatcherBackoffTest.java | 9 +- .../EventDispatcherUnparsableRowTest.java | 19 ++- api/src/test/resources/application-test.yml | 10 ++ 17 files changed, 458 insertions(+), 88 deletions(-) create mode 100644 api/src/main/java/com/cdp/delivery/DeliveryListenerConfig.java create mode 100644 api/src/main/java/com/cdp/delivery/LoopProperties.java create mode 100644 api/src/main/java/com/cdp/delivery/LoopStartupValidator.java diff --git a/api/src/main/java/com/cdp/common/fault/FaultPoint.java b/api/src/main/java/com/cdp/common/fault/FaultPoint.java index 05aa5c9..72e2243 100644 --- a/api/src/main/java/com/cdp/common/fault/FaultPoint.java +++ b/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 } diff --git a/api/src/main/java/com/cdp/delivery/DeliveryConsumer.java b/api/src/main/java/com/cdp/delivery/DeliveryConsumer.java index 079131d..f52b5ed 100644 --- a/api/src/main/java/com/cdp/delivery/DeliveryConsumer.java +++ b/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(), diff --git a/api/src/main/java/com/cdp/delivery/DeliveryListenerConfig.java b/api/src/main/java/com/cdp/delivery/DeliveryListenerConfig.java new file mode 100644 index 0000000..272350d --- /dev/null +++ b/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 deliveryShutdownCustomizer( + DeliveryProperties properties) { + return container -> container.setShutdownTimeout(properties.shutdownTimeout().toMillis()); + } +} diff --git a/api/src/main/java/com/cdp/delivery/DeliveryProperties.java b/api/src/main/java/com/cdp/delivery/DeliveryProperties.java index 88185c4..4a77684 100644 --- a/api/src/main/java/com/cdp/delivery/DeliveryProperties.java +++ b/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) {} diff --git a/api/src/main/java/com/cdp/delivery/LoopProperties.java b/api/src/main/java/com/cdp/delivery/LoopProperties.java new file mode 100644 index 0000000..796cf26 --- /dev/null +++ b/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) {} diff --git a/api/src/main/java/com/cdp/delivery/LoopStartupValidator.java b/api/src/main/java/com/cdp/delivery/LoopStartupValidator.java new file mode 100644 index 0000000..8c4e169 --- /dev/null +++ b/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"); + } + } +} diff --git a/api/src/main/java/com/cdp/delivery/OutboxStore.java b/api/src/main/java/com/cdp/delivery/OutboxStore.java index 12d9dd6..aa671c8 100644 --- a/api/src/main/java/com/cdp/delivery/OutboxStore.java +++ b/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 entries); - /** status=PENDING 且 available_at <= now，FOR UPDATE SKIP LOCKED；須在交易內呼叫。 */ - List lockDue(Instant now, int limit); + /** + * 搶列（單一敘述，不需外層交易）：status=PENDING 且 available_at <= now，FOR UPDATE SKIP LOCKED，同時把 + * available_at 設為 leaseUntil 並回傳列（回傳列的 availableAt 即 fencing 值）。 + */ + List 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 leases, Instant publishedAt); + + /** {@link #reschedule} 的批次版，回傳與輸入同序的各列影響筆數。 */ + int[] rescheduleAll(List 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 find(String idempotencyKey); } diff --git a/api/src/main/java/com/cdp/delivery/relay/OutboxRelay.java b/api/src/main/java/com/cdp/delivery/relay/OutboxRelay.java index 0584c6d..4716a14 100644 --- a/api/src/main/java/com/cdp/delivery/relay/OutboxRelay.java +++ b/api/src/main/java/com/cdp/delivery/relay/OutboxRelay.java @@ -1,12 +1,19 @@ package com.cdp.delivery.relay; +import java.sql.Statement; import java.time.Clock; +import java.time.Instant; +import java.time.temporal.ChronoUnit; +import java.util.ArrayList; import java.util.List; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.rabbit.connection.CorrelationData; +import org.springframework.amqp.rabbit.core.RabbitOperations; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; @@ -14,23 +21,26 @@ import com.cdp.common.fault.FaultInjector; import com.cdp.common.fault.FaultPoint; +import com.cdp.delivery.LoopProperties; import com.cdp.delivery.OutboxEntry; import com.cdp.delivery.OutboxStore; +import com.cdp.delivery.OutboxStore.Lease; import com.cdp.delivery.RabbitTopology; /** - * 全系統唯一可依賴 RabbitTemplate 的類別（ArchitectureTest）。每列：發佈 body = idempotency_key → 等 - * publisher confirm → markPublished；confirm 失敗則留 PENDING 下輪再推。 + * 全系統唯一可依賴 RabbitTemplate 的類別（ArchitectureTest）。spec 07 §12「relay」：搶列（短交易，順便設 lease）→ + * 交易外整批 convertAndSend（CorrelationData）→ 一起等 confirm（上限 confirmTimeout）→ 一個小交易 batchUpdate 條件式回寫。 + * 回寫以 {@code available_at = lease} 為 fencing：lease 已被別人接手或列已取消，回寫影響 0 列，結果丟棄。 */ @Component public class OutboxRelay { private static final Logger log = LoggerFactory.getLogger(OutboxRelay.class); - private static final long CONFIRM_TIMEOUT_MILLIS = 5000; private final OutboxStore outbox; private final RabbitTemplate rabbit; private final RelayProperties properties; + private final LoopProperties loop; private final FaultInjector faults; private final Clock clock; private final TransactionTemplate tx; @@ -39,74 +49,155 @@ public OutboxRelay( OutboxStore outbox, RabbitTemplate rabbit, RelayProperties properties, + LoopProperties loop, FaultInjector faults, Clock clock, TransactionTemplate tx) { this.outbox = outbox; this.rabbit = rabbit; this.properties = properties; + this.loop = loop; this.faults = faults; this.clock = clock; this.tx = tx; } + /** 一輪的結果：claimed = 搶到幾列，published = 其中成功標 PUBLISHED 的筆數，clean = 全數成功。 */ + private record Round(int claimed, int published, boolean clean) {} + + private record Pending(Lease lease, CorrelationData correlation) {} + + /** 滿批續做，直到不滿批、本輪有失敗（broker 異常時不空轉），或超過 CDP_LOOP_MAX_DRAIN。 */ @Scheduled(fixedDelayString = "${cdp.relay.poll-interval}") public void tick() { - runOnce(); + long deadline = System.nanoTime() + loop.maxDrain().toNanos(); + Round round; + do { + round = round(); + } while (round.claimed() >= properties.batchSize() + && round.clean() + && System.nanoTime() < deadline); } - /** 推送一批，回傳筆數（測試可直接呼叫）。 */ + /** 推送一批，回傳標為 PUBLISHED 的筆數（測試可直接呼叫）。 */ public int runOnce() { - Integer n = tx.execute(status -> publishBatch()); - return n == null ? 0 : n; + return round().published(); } - private int publishBatch() { - List due = outbox.lockDue(clock.instant(), properties.batchSize()); - int published = 0; - for (OutboxEntry entry : due) { + private Round round() { + Instant now = clock.instant(); + // Postgres timestamptz 只到微秒；先截斷，fencing 比對才不會差奈秒 + Instant leaseUntil = now.plus(properties.lease()).truncatedTo(ChronoUnit.MICROS); + List claimed = + tx.execute(status -> outbox.claimDue(now, leaseUntil, properties.batchSize())); + if (claimed == null || claimed.isEmpty()) { + return new Round(0, 0, true); + } + + List failed = new ArrayList<>(); + List succeeded = new ArrayList<>(); + try { + // 整批在同一條 channel 上送出並等 confirm（不在 DB 交易內）：每則各取 channel 時，帶著未 confirm 訊息的 + // channel 不會回 cache，broker 無回應時下一則要開新 channel 而卡死 + rabbit.invoke( + ops -> { + sendAndConfirm(ops, claimed, failed, succeeded); + return null; + }); + } catch (RuntimeException e) { + log.warn("relay channel failure; unresolved rows keep their lease: {}", e.toString()); + } + + int published = writeBack(succeeded, failed); + boolean clean = failed.isEmpty() && published == claimed.size(); + return new Round(claimed.size(), published, clean); + } + + private void sendAndConfirm( + RabbitOperations ops, List claimed, List failed, List succeeded) { + List inFlight = new ArrayList<>(); + boolean sendBroken = false; + for (OutboxEntry entry : claimed) { String key = entry.idempotencyKey(); + Lease lease = new Lease(key, entry.availableAt()); String routingKey = routingKey(entry.effectType()); if (routingKey == null) { log.error("unknown effect_type {} for outbox {}; rescheduled", entry.effectType(), key); - outbox.reschedule(key, clock.instant().plus(properties.retryDelay())); + failed.add(lease); + continue; + } + if (sendBroken) { + failed.add(lease); // broker 不可用時不逐列等逾時 continue; } try { faults.at(FaultPoint.RELAY_BEFORE_PUBLISH, key); } catch (RuntimeException e) { + // 模擬在此中斷：不發佈也不回寫，列維持被 lease 住，到期後由下一輪重搶 log.warn("relay interrupted before publish for {}: {}", key, e.toString()); - break; + continue; } - String failure = publish(routingKey, key); + CorrelationData correlation = new CorrelationData(key); + try { + ops.convertAndSend(RabbitTopology.EXCHANGE, routingKey, key, correlation); + inFlight.add(new Pending(lease, correlation)); + } catch (RuntimeException e) { + log.warn("publish failed for outbox {}; rescheduled: {}", key, e.toString()); + failed.add(lease); + sendBroken = true; + } + } + + long deadline = System.nanoTime() + properties.confirmTimeout().toNanos(); + for (Pending p : inFlight) { + String key = p.lease().idempotencyKey(); + String failure = awaitConfirm(p.correlation(), deadline); if (failure != null) { log.warn("publish failed for outbox {}; rescheduled: {}", key, failure); - outbox.reschedule(key, clock.instant().plus(properties.retryDelay())); - break; // 一批只等一次失敗（broker 不可用時不逐列等逾時） + failed.add(p.lease()); + continue; } try { faults.at(FaultPoint.RELAY_AFTER_CONFIRM, key); + faults.at(FaultPoint.RELAY_BEFORE_WRITEBACK, key); } catch (RuntimeException e) { - log.warn("relay interrupted after confirm for {}: {}", key, e.toString()); - break; + log.warn("relay interrupted before writeback for {}: {}", key, e.toString()); + continue; + } + succeeded.add(p.lease()); + } + + } + + /** 一個小交易、batchUpdate 條件式回寫；回傳實際標為 PUBLISHED 的筆數。 */ + private int writeBack(List succeeded, List failed) { + if (succeeded.isEmpty() && failed.isEmpty()) { + return 0; + } + Instant at = clock.instant(); + int[][] counts = new int[1][]; + tx.executeWithoutResult( + status -> { + counts[0] = outbox.markPublishedAll(succeeded, at); + outbox.rescheduleAll(failed, at.plus(properties.retryDelay())); + }); + int published = 0; + for (int i = 0; i < succeeded.size(); i++) { + if (counts[0][i] > 0 || counts[0][i] == Statement.SUCCESS_NO_INFO) { + published++; + } else { + log.info("lease lost for outbox {}; late confirm discarded", succeeded.get(i).idempotencyKey()); } - outbox.markPublished(key, clock.instant()); - published++; } return published; } - /** correlated confirm：回傳 null 表示 broker 已 ack 且未被 returned，否則為失敗原因。 */ - private String publish(String routingKey, String key) { - CorrelationData correlation = new CorrelationData(key); + /** 回傳 null 表示 broker 已 ack 且未被 returned，否則為失敗原因；所有列共用同一個截止時間。 */ + private static String awaitConfirm(CorrelationData correlation, long deadlineNanos) { try { - rabbit.invoke( - ops -> { - ops.convertAndSend(RabbitTopology.EXCHANGE, routingKey, key, correlation); - return null; - }); + long remaining = Math.max(0, deadlineNanos - System.nanoTime()); CorrelationData.Confirm confirm = - correlation.getFuture().get(CONFIRM_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); + correlation.getFuture().get(remaining, TimeUnit.NANOSECONDS); if (!confirm.isAck()) { return "nack: " + confirm.getReason(); } @@ -114,10 +205,12 @@ private String publish(String routingKey, String key) { return "returned: " + correlation.getReturned().getReplyText(); } return null; + } catch (TimeoutException e) { + return "confirm timeout"; } catch (InterruptedException e) { Thread.currentThread().interrupt(); return e.toString(); - } catch (Exception e) { + } catch (ExecutionException | RuntimeException e) { return e.toString(); } } diff --git a/api/src/main/java/com/cdp/delivery/relay/RelayProperties.java b/api/src/main/java/com/cdp/delivery/relay/RelayProperties.java index b22660b..23dc691 100644 --- a/api/src/main/java/com/cdp/delivery/relay/RelayProperties.java +++ b/api/src/main/java/com/cdp/delivery/relay/RelayProperties.java @@ -4,6 +4,14 @@ import org.springframework.boot.context.properties.ConfigurationProperties; -/** cdp.relay.*（變數見 DESIGN-01「設定」）。 */ +/** + * cdp.relay.*（變數見 DESIGN-01、DESIGN-03「設定」）。confirmTimeout = OUTBOX_RELAY_CONFIRM_TIMEOUT，lease = + * OUTBOX_RELAY_LEASE（必須大於 confirmTimeout，由 {@link com.cdp.delivery.LoopStartupValidator} 啟動時驗證）。 + */ @ConfigurationProperties("cdp.relay") -public record RelayProperties(Duration pollInterval, int batchSize, Duration retryDelay) {} +public record RelayProperties( + Duration pollInterval, + int batchSize, + Duration retryDelay, + Duration confirmTimeout, + Duration lease) {} diff --git a/api/src/main/java/com/cdp/engine/runtime/AdvanceLoop.java b/api/src/main/java/com/cdp/engine/runtime/AdvanceLoop.java index 930be13..edfae7b 100644 --- a/api/src/main/java/com/cdp/engine/runtime/AdvanceLoop.java +++ b/api/src/main/java/com/cdp/engine/runtime/AdvanceLoop.java @@ -2,21 +2,25 @@ import java.time.Clock; import java.time.Instant; +import java.util.HashSet; import java.util.List; +import java.util.Set; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; -import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionTemplate; +import com.cdp.delivery.LoopProperties; import com.cdp.engine.StateStore; import com.cdp.engine.model.EnrollmentState; import com.cdp.engine.model.WakeSignal; /** - * 計時器喚醒（R1）：每 cdp.engine.poll-interval 以 Clock 的現在時間 lockDue，逐筆 advance(TimerFired)。 - * lockDue 的鎖與推進在同一交易；單筆失敗只回滾那一筆（advance 為 NESTED），其餘照常，失敗者留待下輪重試。 + * 計時器喚醒（R1）：每 cdp.engine.poll-interval 以 Clock 的現在時間逐筆推進到期的 Enrollment。 + * 每一筆一個交易（lockDue → advance(TimerFired)，DESIGN-03「relay 與迴圈」）：失敗只回滾那一筆，本輪略過、下輪重試。 + * tick 滿批續做到 CDP_LOOP_MAX_DRAIN。 */ @Component public class AdvanceLoop { @@ -26,39 +30,70 @@ public class AdvanceLoop { private final StateStore stateStore; private final EnrollmentAdvancer advancer; private final EngineProperties properties; + private final LoopProperties loop; + private final TransactionTemplate tx; private final Clock clock; public AdvanceLoop( StateStore stateStore, EnrollmentAdvancer advancer, EngineProperties properties, + LoopProperties loop, + TransactionTemplate tx, Clock clock) { this.stateStore = stateStore; this.advancer = advancer; this.properties = properties; + this.loop = loop; + this.tx = tx; this.clock = clock; } @Scheduled(fixedDelayString = "${cdp.engine.poll-interval}") - @Transactional public void tick() { - runOnce(); + long deadline = System.nanoTime() + loop.maxDrain().toNanos(); + Set failed = new HashSet<>(); + int n; + do { + n = runBatch(failed); + } while (n >= properties.batchSize() && System.nanoTime() < deadline); } - /** 推進一批到期 Enrollment，回傳筆數（測試可直接呼叫）。 */ - @Transactional + /** 推進一批到期 Enrollment（最多 batchSize 筆），回傳處理筆數（測試可直接呼叫）。 */ public int runOnce() { + return runBatch(new HashSet<>()); + } + + private int runBatch(Set failed) { Instant now = clock.instant(); - List due = stateStore.lockDue(now, properties.batchSize()); - int advanced = 0; + int processed = 0; + while (processed < properties.batchSize()) { + Boolean ok = tx.execute(status -> advanceNext(now, failed)); + if (ok == null) { + break; // 沒有可處理的到期列 + } + processed++; + } + return processed; + } + + /** 一個交易：鎖一筆（略過本輪失敗者）並推進。回傳 null = 沒有可處理的列；true／false = 成功／失敗。 */ + private Boolean advanceNext(Instant now, Set failed) { + // StateStore.lockDue 沒有排除清單：多取「失敗數 + 1」列，取第一個沒失敗過的 + List due = stateStore.lockDue(now, failed.size() + 1); for (EnrollmentState state : due) { + if (failed.contains(state.enrollmentId())) { + continue; + } try { advancer.advance(state, new WakeSignal.TimerFired(now)); - advanced++; + return Boolean.TRUE; } catch (RuntimeException e) { + failed.add(state.enrollmentId()); log.error("advance failed enrollment_id={}", state.enrollmentId(), e); + return Boolean.FALSE; } } - return advanced; + return null; } } diff --git a/api/src/main/java/com/cdp/ingest/EventDispatcher.java b/api/src/main/java/com/cdp/ingest/EventDispatcher.java index 5e88e36..3937201 100644 --- a/api/src/main/java/com/cdp/ingest/EventDispatcher.java +++ b/api/src/main/java/com/cdp/ingest/EventDispatcher.java @@ -3,15 +3,18 @@ import java.time.Clock; import java.time.Duration; import java.time.Instant; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; -import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionTemplate; +import com.cdp.delivery.LoopProperties; import com.cdp.engine.JourneyRuntime; import com.cdp.engine.model.TriggerSnapshot; import com.cdp.engine.runtime.EngineProperties; @@ -19,8 +22,9 @@ import com.cdp.journey.JourneyDefinitionStore; /** - * A 類扇出（Postgres 輪詢）：同一交易內 lockUndispatched → 對每個命中的 EventSourceBinding 呼叫 - * JourneyRuntime.start → markDispatched。遲到 > 24h 只標 LATE、不建立（01 §6）。 + * A 類扇出（Postgres 輪詢）：每個事件一個交易（DESIGN-03「relay 與迴圈」）：lockUndispatched → 對每個命中的 + * EventSourceBinding 呼叫 JourneyRuntime.start → markDispatched。遲到 > 24h 只標 LATE、不建立（01 §6）。tick 滿批 + * 續做到 CDP_LOOP_MAX_DRAIN；失敗的事件本輪略過。 * *

start 為 NESTED，單一綁定失敗只回滾它自己；該事件記一次失敗並依指數退避延後，達上限標 EXHAUSTED。已建立的 * Enrollment 靠 dedup_key（= event_id）去重，重試時只補失敗的綁定。規則見 spec 07 §11。 @@ -37,6 +41,8 @@ public class EventDispatcher { private final JourneyRuntime runtime; private final EngineProperties properties; private final DispatchProperties dispatchProperties; + private final LoopProperties loop; + private final TransactionTemplate tx; private final Clock clock; public EventDispatcher( @@ -45,46 +51,76 @@ public EventDispatcher( JourneyRuntime runtime, EngineProperties properties, DispatchProperties dispatchProperties, + LoopProperties loop, + TransactionTemplate tx, Clock clock) { this.events = events; this.journeys = journeys; this.runtime = runtime; this.properties = properties; this.dispatchProperties = dispatchProperties; + this.loop = loop; + this.tx = tx; this.clock = clock; } @Scheduled(fixedDelayString = "${cdp.engine.poll-interval}") - @Transactional public void tick() { - runOnce(); + long deadline = System.nanoTime() + loop.maxDrain().toNanos(); + Set failed = new HashSet<>(); + int n; + do { + n = runBatch(failed); + } while (n >= properties.batchSize() && System.nanoTime() < deadline); } - /** 派發一批，回傳事件筆數（測試可直接呼叫）。 */ - @Transactional + /** 派發一批（最多 batchSize 個事件，每個事件一個交易），回傳事件筆數（測試可直接呼叫）。 */ public int runOnce() { + return runBatch(new HashSet<>()); + } + + private int runBatch(Set failed) { + int processed = 0; + while (processed < properties.batchSize()) { + Boolean ok = tx.execute(status -> dispatchNext(failed)); + if (ok == null) { + break; // 沒有可派發的事件 + } + processed++; + } + return processed; + } + + /** 一個交易：鎖一個事件（略過本輪失敗者）並派發。回傳 null = 沒有可處理的事件。 */ + private Boolean dispatchNext(Set failed) { Instant now = clock.instant(); - List batch = events.lockUndispatched(properties.batchSize(), now); - for (UndispatchedEvent row : batch) { + // EventStore.lockUndispatched 沒有排除清單：多取「失敗數 + 1」列，取第一個沒失敗過的 + List rows = events.lockUndispatched(failed.size() + 1, now); + for (UndispatchedEvent row : rows) { + if (failed.contains(row.eventId())) { + continue; + } NormalizedEvent event = row.event(); if (event == null) { events.recordFailure(row.eventId(), row.parseError()); events.markDispatched(row.eventId(), now, EXHAUSTED); log.error("event row unparsable, giving up event_id={}", row.eventId()); - continue; + return Boolean.TRUE; } if (isLate(event)) { events.markDispatched(event.eventId(), now, "LATE"); - continue; + return Boolean.TRUE; } RuntimeException failure = dispatch(event); if (failure == null) { events.markDispatched(event.eventId(), now, "DISPATCHED"); - } else { - onFailure(event, failure, now); + return Boolean.TRUE; } + failed.add(event.eventId()); + onFailure(event, failure, now); + return Boolean.FALSE; } - return batch.size(); + return null; } private void onFailure(NormalizedEvent event, RuntimeException failure, Instant now) { diff --git a/api/src/main/java/com/cdp/persistence/JdbcOutboxStore.java b/api/src/main/java/com/cdp/persistence/JdbcOutboxStore.java index 2ba73a7..eecaf11 100644 --- a/api/src/main/java/com/cdp/persistence/JdbcOutboxStore.java +++ b/api/src/main/java/com/cdp/persistence/JdbcOutboxStore.java @@ -1,5 +1,6 @@ package com.cdp.persistence; +import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; import java.time.Instant; @@ -9,6 +10,8 @@ import java.util.Map; import java.util.Optional; +import org.springframework.jdbc.core.BatchPreparedStatementSetter; +import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.jdbc.core.simple.JdbcClient; import org.springframework.stereotype.Repository; @@ -28,11 +31,23 @@ public class JdbcOutboxStore implements OutboxStore { + " visit_seq, effect_index, effect_type, payload, status, available_at," + " publish_count, created_at, published_at"; + /** 條件式回寫（fencing）：status 仍為 PENDING 且 available_at 仍是搶列時的 lease。參數：at、key、lease。 */ + private static final String MARK_PUBLISHED = + "update effect_outbox set status = 'PUBLISHED', published_at = ?," + + " publish_count = publish_count + 1" + + " where idempotency_key = ? and status = 'PENDING' and available_at = ?"; + + private static final String RESCHEDULE = + "update effect_outbox set available_at = ?" + + " where idempotency_key = ? and status = 'PENDING' and available_at = ?"; + private final JdbcClient jdbc; + private final JdbcTemplate jdbcTemplate; private final JsonMapper json; - public JdbcOutboxStore(JdbcClient jdbc, JsonMapper json) { + public JdbcOutboxStore(JdbcClient jdbc, JdbcTemplate jdbcTemplate, JsonMapper json) { this.jdbc = jdbc; + this.jdbcTemplate = jdbcTemplate; this.json = json; } @@ -64,11 +79,13 @@ public void insertAll(List entries) { } @Override - public List lockDue(Instant now, int limit) { + public List claimDue(Instant now, Instant leaseUntil, int limit) { return jdbc.sql( - "select " + COLUMNS + " from effect_outbox where status = 'PENDING'" + "update effect_outbox set available_at = :lease where idempotency_key in (" + + " select idempotency_key from effect_outbox where status = 'PENDING'" + " and available_at <= :now order by available_at, idempotency_key" - + " limit :limit for update skip locked") + + " limit :limit for update skip locked) returning " + COLUMNS) + .param("lease", ts(leaseUntil)) .param("now", ts(now)) .param("limit", limit) .query((rs, n) -> map(rs)) @@ -76,22 +93,66 @@ public List lockDue(Instant now, int limit) { } @Override - public void markPublished(String idempotencyKey, Instant publishedAt) { - jdbc.sql( - "update effect_outbox set status = 'PUBLISHED', published_at = :at," - + " publish_count = publish_count + 1 where idempotency_key = :key") - .param("at", ts(publishedAt)) - .param("key", idempotencyKey) + public boolean markPublished(String idempotencyKey, Instant leaseUntil, Instant publishedAt) { + return jdbcTemplate.update( + MARK_PUBLISHED, ts(publishedAt), idempotencyKey, ts(leaseUntil)) + == 1; + } + + @Override + public boolean reschedule(String idempotencyKey, Instant leaseUntil, Instant availableAt) { + return jdbcTemplate.update(RESCHEDULE, ts(availableAt), idempotencyKey, ts(leaseUntil)) + == 1; + } + + @Override + public int[] markPublishedAll(List leases, Instant publishedAt) { + return batch(MARK_PUBLISHED, leases, publishedAt); + } + + @Override + public int[] rescheduleAll(List leases, Instant availableAt) { + return batch(RESCHEDULE, leases, availableAt); + } + + private int[] batch(String sql, List leases, Instant at) { + if (leases.isEmpty()) { + return new int[0]; + } + OffsetDateTime atValue = ts(at); + return jdbcTemplate.batchUpdate( + sql, + new BatchPreparedStatementSetter() { + @Override + public void setValues(PreparedStatement ps, int i) throws SQLException { + Lease lease = leases.get(i); + ps.setObject(1, atValue); + ps.setString(2, lease.idempotencyKey()); + ps.setObject(3, ts(lease.leaseUntil())); + } + + @Override + public int getBatchSize() { + return leases.size(); + } + }); + } + + @Override + public int cancelPendingByEnrollment(String enrollmentId) { + return jdbc.sql( + "update effect_outbox set status = 'CANCELLED'" + + " where enrollment_id = :id and status = 'PENDING'") + .param("id", enrollmentId) .update(); } @Override - public void reschedule(String idempotencyKey, Instant availableAt) { - jdbc.sql( - "update effect_outbox set status = 'PENDING', available_at = :at" - + " where idempotency_key = :key") - .param("at", ts(availableAt)) - .param("key", idempotencyKey) + public int cancelPendingByJourney(String journeyId) { + return jdbc.sql( + "update effect_outbox set status = 'CANCELLED'" + + " where journey_id = :id and status = 'PENDING'") + .param("id", journeyId) .update(); } diff --git a/api/src/main/resources/application.yml b/api/src/main/resources/application.yml index d20ee11..1905627 100644 --- a/api/src/main/resources/application.yml +++ b/api/src/main/resources/application.yml @@ -11,6 +11,7 @@ spring: port: ${RABBITMQ_PORT} username: ${RABBITMQ_USER} password: ${RABBITMQ_PASSWORD} + connection-timeout: ${RABBITMQ_CONNECTION_TIMEOUT} # relay 等 broker confirm 後才標 PUBLISHED（DESIGN-01 做法 #4） publisher-confirm-type: correlated publisher-returns: true @@ -21,6 +22,11 @@ spring: acknowledge-mode: auto # 例外預設進 DLQ；要重投的情況由 consumer 丟 ImmediateRequeueAmqpException default-requeue-rejected: false + task: + scheduling: + pool: + # 輪詢迴圈各佔一條執行緒，relay 卡住不能拖住計時器；啟動驗證 >= 迴圈數（DESIGN-03） + size: ${CDP_SCHEDULER_POOL_SIZE} jackson: # 對外 JSON 一律 snake_case（docs/spec/README.md「命名」） property-naming-strategy: SNAKE_CASE @@ -51,8 +57,14 @@ cdp: poll-interval: ${OUTBOX_RELAY_POLL_INTERVAL} batch-size: ${OUTBOX_RELAY_BATCH_SIZE} retry-delay: ${OUTBOX_RELAY_RETRY_DELAY} + confirm-timeout: ${OUTBOX_RELAY_CONFIRM_TIMEOUT} + # 必須大於 confirm-timeout（啟動驗證） + lease: ${OUTBOX_RELAY_LEASE} + loop: + max-drain: ${CDP_LOOP_MAX_DRAIN} delivery: requeue-delay: ${DELIVERY_REQUEUE_DELAY} + shutdown-timeout: ${DELIVERY_SHUTDOWN_TIMEOUT} security: # 契約 §1.1、§1.2；機密只在 .env ingest-api-key: ${CDP_INGEST_API_KEY} diff --git a/api/src/test/java/com/cdp/delivery/DeliveryConsumerTest.java b/api/src/test/java/com/cdp/delivery/DeliveryConsumerTest.java index 160388b..06a408d 100644 --- a/api/src/test/java/com/cdp/delivery/DeliveryConsumerTest.java +++ b/api/src/test/java/com/cdp/delivery/DeliveryConsumerTest.java @@ -45,7 +45,7 @@ private DeliveryConsumer consumer() { mock(FaultInjector.class), Clock.fixed(Instant.parse("2026-01-01T00:00:00Z"), ZoneOffset.UTC), channel, - new DeliveryProperties(DELAY)); + new DeliveryProperties(DELAY, Duration.ofSeconds(5))); } @Test diff --git a/api/src/test/java/com/cdp/ingest/EventDispatcherBackoffTest.java b/api/src/test/java/com/cdp/ingest/EventDispatcherBackoffTest.java index 6c28d46..95550ba 100644 --- a/api/src/test/java/com/cdp/ingest/EventDispatcherBackoffTest.java +++ b/api/src/test/java/com/cdp/ingest/EventDispatcherBackoffTest.java @@ -11,7 +11,14 @@ class EventDispatcherBackoffTest { private static EventDispatcher with(Duration initial, Duration max) { return new EventDispatcher( - null, null, null, null, new DispatchProperties(5, initial, max), Clock.systemUTC()); + null, + null, + null, + null, + new DispatchProperties(5, initial, max), + null, + null, + Clock.systemUTC()); } @Test diff --git a/api/src/test/java/com/cdp/ingest/EventDispatcherUnparsableRowTest.java b/api/src/test/java/com/cdp/ingest/EventDispatcherUnparsableRowTest.java index e00b57f..b87934f 100644 --- a/api/src/test/java/com/cdp/ingest/EventDispatcherUnparsableRowTest.java +++ b/api/src/test/java/com/cdp/ingest/EventDispatcherUnparsableRowTest.java @@ -1,6 +1,9 @@ package com.cdp.ingest; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import java.lang.reflect.Proxy; import java.time.Clock; @@ -11,7 +14,11 @@ import java.util.List; import org.junit.jupiter.api.Test; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.SimpleTransactionStatus; +import org.springframework.transaction.support.TransactionTemplate; +import com.cdp.delivery.LoopProperties; import com.cdp.engine.runtime.EngineProperties; import com.cdp.journey.EventSourceBinding; import com.cdp.journey.JourneyDefinitionStore; @@ -36,19 +43,21 @@ private static NormalizedEvent goodEvent() { @Test void unparsableRowIsExhaustedAndOthersStillDispatched() { List calls = new ArrayList<>(); + List pending = new ArrayList<>(List.of( + new UndispatchedEvent("bad", null, "parse boom"), + new UndispatchedEvent("good", goodEvent(), null))); EventStore store = (EventStore) Proxy.newProxyInstance( getClass().getClassLoader(), new Class[] {EventStore.class}, (p, m, a) -> switch (m.getName()) { - case "lockUndispatched" -> List.of( - new UndispatchedEvent("bad", null, "parse boom"), - new UndispatchedEvent("good", goodEvent(), null)); + case "lockUndispatched" -> List.copyOf(pending); case "recordFailure" -> { calls.add("fail:" + a[0]); yield 1; } case "markDispatched" -> { calls.add("mark:" + a[0] + ":" + a[2]); + pending.removeIf(r -> r.eventId().equals(a[0])); yield null; } default -> throw new UnsupportedOperationException(m.getName()); @@ -62,12 +71,16 @@ void unparsableRowIsExhaustedAndOthersStillDispatched() { } throw new UnsupportedOperationException(m.getName()); }); + PlatformTransactionManager txManager = mock(PlatformTransactionManager.class); + when(txManager.getTransaction(any())).thenReturn(new SimpleTransactionStatus()); EventDispatcher d = new EventDispatcher( store, journeys, null, new EngineProperties(Duration.ofSeconds(1), 10), new DispatchProperties(5, Duration.ofSeconds(10), Duration.ofMinutes(10)), + new LoopProperties(Duration.ofSeconds(10)), + new TransactionTemplate(txManager), Clock.fixed(NOW, ZoneOffset.UTC)); assertThat(d.runOnce()).isEqualTo(2); diff --git a/api/src/test/resources/application-test.yml b/api/src/test/resources/application-test.yml index 02dc28c..a16ee65 100644 --- a/api/src/test/resources/application-test.yml +++ b/api/src/test/resources/application-test.yml @@ -10,10 +10,15 @@ spring: port: 5672 username: test password: test + connection-timeout: PT5S publisher-confirm-type: correlated publisher-returns: true template: mandatory: true + task: + scheduling: + pool: + size: 4 server: port: 0 servlet: @@ -34,8 +39,13 @@ cdp: poll-interval: PT0.2S batch-size: 100 retry-delay: PT30S + confirm-timeout: PT5S + lease: PT30S + loop: + max-drain: PT10S delivery: requeue-delay: PT0.2S + shutdown-timeout: PT5S # 測試專用假憑證（非機密）；驗收測試以此呼叫各角色 security: ingest-api-key: test-ingest-key From f6a908ad9a0d3eb2db6ebf030a3efe8e2790fe41 Mon Sep 17 00:00:00 2001 From: Marcus Date: Thu, 22 Oct 2026 12:20:59 +0800 Subject: [PATCH 3/3] =?UTF-8?q?test(relay-loop-rewrite):=20=E4=BE=9D?= =?UTF-8?q?=E4=BB=B2=E8=A3=81=E4=BF=AE=E6=AD=A3=20delivery=20=E5=85=A9?= =?UTF-8?q?=E6=B8=AC=E8=A9=A6=EF=BC=88lease=20=E5=BF=AB=E8=BD=89=EF=BC=89?= =?UTF-8?q?=E8=88=87=20import=20=E9=A0=86=E5=BA=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_0134QMUz6NTfqU8Yrh3BJekj --- .../acceptance/delivery/DeliveryPipelineAcceptanceTest.java | 6 ++++-- .../relay-loop-rewrite/RelayStuckTimersAcceptanceTest.java | 3 +-- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/api/src/test/java/com/cdp/acceptance/delivery/DeliveryPipelineAcceptanceTest.java b/api/src/test/java/com/cdp/acceptance/delivery/DeliveryPipelineAcceptanceTest.java index fc33935..9cae93a 100644 --- a/api/src/test/java/com/cdp/acceptance/delivery/DeliveryPipelineAcceptanceTest.java +++ b/api/src/test/java/com/cdp/acceptance/delivery/DeliveryPipelineAcceptanceTest.java @@ -48,7 +48,7 @@ * 觀察。輪詢間隔調成 1 小時，relay 只由 runOnce() 驅動；consumer（@RabbitListener）照常自動啟動。 */ @SpringBootTest( - properties = {"cdp.engine.poll-interval=PT1H", "cdp.relay.poll-interval=PT1H"}) + properties = {"cdp.engine.poll-interval=PT1H", "cdp.relay.poll-interval=PT1H", "cdp.relay.lease=PT10S"}) @AutoConfigureMockMvc @ActiveProfiles("test") @Import(DeliveryTestSupport.class) @@ -366,6 +366,7 @@ void relayInterruptedBeforePublish_thenRerun_sendsExactlyOnce() throws Exception faults.disarm(FaultPoint.RELAY_BEFORE_PUBLISH, s.key()); } + clock.advance(Duration.ofSeconds(11)); relay.runOnce(); await().atMost(WAIT).until(() -> "SENT".equals(taskStatus(s.key()))); relay.runOnce(); @@ -394,7 +395,8 @@ void relayInterruptedAfterConfirm_thenRestart_republishes_butMockSendsExactlyOnc faults.disarm(FaultPoint.RELAY_AFTER_CONFIRM, s.key()); } - // 「重啟」：relay 再取到同一列，重複推送 + // 「重啟」：lease 過期後 relay 再取到同一列，重複推送 + clock.advance(Duration.ofSeconds(11)); relay.runOnce(); assertThat(outboxStatus(s.key())).isEqualTo("PUBLISHED"); settle(); diff --git a/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayStuckTimersAcceptanceTest.java b/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayStuckTimersAcceptanceTest.java index d0747a3..d3270b0 100644 --- a/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayStuckTimersAcceptanceTest.java +++ b/api/src/test/java/com/cdp/acceptance/relay-loop-rewrite/RelayStuckTimersAcceptanceTest.java @@ -6,6 +6,7 @@ import java.time.Duration; import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.boot.webmvc.test.autoconfigure.AutoConfigureMockMvc; import org.springframework.context.annotation.Import; @@ -13,8 +14,6 @@ import org.testcontainers.DockerClientFactory; import org.testcontainers.rabbitmq.RabbitMQContainer; -import org.springframework.beans.factory.annotation.Autowired; - /** * relay 卡住（broker 無回應，等 confirm 逾時）時，計時器迴圈仍照常推進。 * 排程迴圈全部開啟（真實排程），排程執行緒池 = 4；confirm 逾時拉長到 PT8S，確保 relay 的阻塞期涵蓋觀察窗。