From 48a27b26dd1b1d57247fde5a3ce31b25690dae10 Mon Sep 17 00:00:00 2001 From: windyboy Date: Tue, 8 Sep 2026 13:04:43 +0800 Subject: [PATCH] =?UTF-8?q?feat(jobs):=20=E6=8F=90=E4=BA=A4=E5=90=8E?= =?UTF-8?q?=E5=9B=9E=E5=A1=AB=E8=A1=A5=E5=81=BF=E8=90=BD=E5=BA=93=20backfi?= =?UTF-8?q?ll=5Ftodo=20=E5=B9=B6=E5=88=B0=E6=9C=9F=E9=87=8D=E8=AF=95=20(AC?= =?UTF-8?q?M2-29=20P2-3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - V1.3.2 迁移：backfill_todo 持久补偿表（attempts/next_attempt_at/last_error） - BackfillTodoRepository 接口 + PG 实现与 stub；record 幂等 upsert - BackfillSweepJob：到期重试，成功即删，失败指数退避（30s 起步封顶 15 分钟）； 触发点 = InboxPoller 每轮心跳（重启即对账）+ JobExecutor BACKFILL_SWEEP 类型 - MessageProcessor/SnapshotFlow 回填失败落待办（成功终态不降级，业务不重放）； SKIPPED 重复报文同样补偿共享信箱回填，防止重复信件被反复轮询 - 新测试：扫描重试/退避/恢复对账 4 用例；快照回填失败落库断言； 跳过路径补偿落库断言 验证：MSGX_PG_PORT=5433 真实 PG 全量 102 用例 0 失败 0 跳过 (ACM2-29) --- .../infra/persistence/Repositories.kt | 31 +++++++ .../persistence/jdbc/JdbcPgRepositories.kt | 73 +++++++++++++++ .../infra/stub/StubRepositories.kt | 47 ++++++++++ .../omms/msgexchange/ingress/InboxPoller.kt | 15 ++++ .../omms/msgexchange/jobs/BackfillSweepJob.kt | 53 +++++++++++ .../gzzn/omms/msgexchange/jobs/JobExecutor.kt | 2 + .../gzzn/omms/msgexchange/processing/Pump.kt | 28 ++++++ .../msgexchange/processing/SnapshotFlow.kt | 12 +++ .../db/migration/V1.3.2__backfill_todo.sql | 18 ++++ .../msgexchange/jobs/BackfillSweepJobTest.kt | 90 +++++++++++++++++++ .../processing/MessageProcessorTest.kt | 26 +++++- .../processing/SnapshotFlowBackfillTest.kt | 10 ++- 12 files changed, 402 insertions(+), 3 deletions(-) create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/jobs/BackfillSweepJob.kt create mode 100644 src/main/resources/db/migration/V1.3.2__backfill_todo.sql create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/jobs/BackfillSweepJobTest.kt diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt index 2f5c417..1b93d36 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt @@ -162,6 +162,37 @@ interface PipelineTransactionManager { fun inTransaction(block: () -> T): T } +/** + * v2 §5（ACM2-29 P2-3）：提交后共享信箱回填的持久补偿待办。 + * 业务事务已提交（PROC_STATE=SUCCEEDED）后 backfillOnSuccess 失败 → 落本表重试； + * 回填失败不得把已成功的业务事务重新标记为失败，也不得重放业务变更。 + */ +interface BackfillTodoRepository { + data class BackfillTask( + val cminmsgsId: Long, + val sndr: String, + val type: String, + val styp: String, + val seqn: Long, + val attempts: Int = 0, + ) + + /** 失败即落库（幂等 upsert，同 ID 重复失败只刷新错误与重试时间）。 */ + fun record(task: BackfillTask, lastError: String?, now: Instant = Instant.now()) + + /** 到期待重试的补偿任务（next_attempt_at <= now，按到期时间升序）。 */ + fun findDue(now: Instant = Instant.now(), limit: Int = 50): List + + /** 重试失败：累加 attempts 并按退避推后 next_attempt_at。 */ + fun markFailed(cminmsgsId: Long, lastError: String?, nextAttemptAt: Instant, now: Instant = Instant.now()) + + /** 重试成功：移除补偿待办。 */ + fun delete(cminmsgsId: Long) + + /** 当前待办数量（测试/对账用）。 */ + fun count(): Int +} + @Deprecated("Replaced by FlightSchdRepository in ACM2-28", ReplaceWith("FlightSchdRepository")) interface RefDataRepository { data class GenMeta(val flids: List, val version: Long) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt index 52914c6..2e2b9a1 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt @@ -9,6 +9,7 @@ import com.gzzn.omms.msgexchange.domain.MsgEvent import com.gzzn.omms.msgexchange.domain.ProcState import com.gzzn.omms.msgexchange.domain.ProcStatus import com.gzzn.omms.msgexchange.domain.RefUpsert +import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository import com.gzzn.omms.msgexchange.infra.persistence.FlightFields import com.gzzn.omms.msgexchange.infra.persistence.FlightSchdRepository import com.gzzn.omms.msgexchange.infra.persistence.FlightStateRepository @@ -311,6 +312,78 @@ class JdbcPipelineTransactionManager( } } +/** v2 §5（ACM2-29 P2-3）：提交后共享信箱回填补偿待办的 PG 实现。 */ +@Singleton +@Requires(property = "msgx.stubs", notEquals = "true") +@Requires(property = "datasources.default.enabled", value = "true") +class JdbcBackfillTodoRepository( + private val ds: DataSource, +) : BackfillTodoRepository { + + override fun record(task: BackfillTodoRepository.BackfillTask, lastError: String?, now: Instant) { + val ts = now.toSqlTimestamp() + ds.update( + """ + INSERT INTO backfill_todo (cminmsgs_id, sndr, type, styp, seqn, attempts, next_attempt_at, last_error, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, 0, ?, ?, ?, ?) + ON CONFLICT (cminmsgs_id) DO UPDATE SET + last_error = EXCLUDED.last_error, + next_attempt_at = EXCLUDED.next_attempt_at, + updated_at = EXCLUDED.updated_at + """.trimIndent(), + ) { ps -> + ps.setLong(1, task.cminmsgsId) + ps.setString(2, task.sndr) + ps.setString(3, task.type) + ps.setString(4, task.styp) + ps.setLong(5, task.seqn) + ps.setTimestamp(6, ts) + ps.setString(7, lastError?.take(512)) + ps.setTimestamp(8, ts) + ps.setTimestamp(9, ts) + } + } + + override fun findDue(now: Instant, limit: Int): List = + ds.query( + """ + SELECT cminmsgs_id, sndr, type, styp, seqn, attempts FROM backfill_todo + WHERE next_attempt_at <= ? ORDER BY next_attempt_at ASC LIMIT ? + """.trimIndent(), + { ps -> + ps.setTimestamp(1, now.toSqlTimestamp()) + ps.setInt(2, limit) + }, + ) { rs -> + BackfillTodoRepository.BackfillTask( + cminmsgsId = rs.getLong("cminmsgs_id"), + sndr = rs.getString("sndr"), + type = rs.getString("type"), + styp = rs.getString("styp") ?: "", + seqn = rs.getLong("seqn"), + attempts = rs.getInt("attempts"), + ) + } + + override fun markFailed(cminmsgsId: Long, lastError: String?, nextAttemptAt: Instant, now: Instant) { + ds.update( + "UPDATE backfill_todo SET attempts = attempts + 1, next_attempt_at = ?, last_error = ?, updated_at = ? WHERE cminmsgs_id = ?", + ) { ps -> + ps.setTimestamp(1, nextAttemptAt.toSqlTimestamp()) + ps.setString(2, lastError?.take(512)) + ps.setTimestamp(3, now.toSqlTimestamp()) + ps.setLong(4, cminmsgsId) + } + } + + override fun delete(cminmsgsId: Long) { + ds.update("DELETE FROM backfill_todo WHERE cminmsgs_id = ?") { ps -> ps.setLong(1, cminmsgsId) } + } + + override fun count(): Int = + ds.queryOne("SELECT COUNT(*) AS c FROM backfill_todo", {}) { rs -> rs.getInt("c") } ?: 0 +} + @Singleton @Requires(property = "msgx.stubs", notEquals = "true") @Requires(property = "datasources.default.enabled", value = "true") diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt index 5333c83..1c74464 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt @@ -6,6 +6,7 @@ import com.gzzn.omms.msgexchange.domain.MsgEvent import com.gzzn.omms.msgexchange.domain.ProcState import com.gzzn.omms.msgexchange.domain.ProcStatus import com.gzzn.omms.msgexchange.domain.RefUpsert +import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository import com.gzzn.omms.msgexchange.infra.persistence.FlightFields import com.gzzn.omms.msgexchange.infra.persistence.FlightSchdRepository @@ -435,3 +436,49 @@ class StubFlightState( override fun findByDay(day: String): List> = flightSchd.findByDay(day) } + +@Requires(property = "msgx.stubs", value = "true") +@Singleton +class StubBackfillTodo : BackfillTodoRepository { + private data class Row(val task: BackfillTodoRepository.BackfillTask, var nextAttemptAt: Instant) + + private val rows = linkedMapOf() + private val errors = linkedMapOf() + + @Synchronized + fun clear() { + rows.clear() + errors.clear() + } + + @Synchronized + fun tasks(): List = rows.values.map { it.task } + + @Synchronized + fun lastErrorOf(cminmsgsId: Long): String? = errors[cminmsgsId] + + @Synchronized + override fun record(task: BackfillTodoRepository.BackfillTask, lastError: String?, now: Instant) { + rows[task.cminmsgsId] = Row(task, now) + errors[task.cminmsgsId] = lastError + } + + @Synchronized + override fun findDue(now: Instant, limit: Int): List = + rows.values.filter { it.nextAttemptAt <= now }.map { it.task }.take(limit) + + @Synchronized + override fun markFailed(cminmsgsId: Long, lastError: String?, nextAttemptAt: Instant, now: Instant) { + rows[cminmsgsId]?.let { rows[cminmsgsId] = it.copy(task = it.task.copy(attempts = it.task.attempts + 1), nextAttemptAt = nextAttemptAt) } + errors[cminmsgsId] = lastError + } + + @Synchronized + override fun delete(cminmsgsId: Long) { + rows.remove(cminmsgsId) + errors.remove(cminmsgsId) + } + + @Synchronized + override fun count(): Int = rows.size +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt index 02a238e..120b4e9 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt @@ -2,6 +2,7 @@ package com.gzzn.omms.msgexchange.ingress import com.gzzn.omms.msgexchange.config.PipelineProps import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository +import com.gzzn.omms.msgexchange.jobs.BackfillSweepJob import jakarta.inject.Singleton /** @@ -12,6 +13,7 @@ import jakarta.inject.Singleton class InboxPoller( private val inbox: CminmsgInboxRepository, private val enqueue: InboxEnqueue, + private val backfillSweep: BackfillSweepJob, private val props: PipelineProps, ) { private val log = org.slf4j.LoggerFactory.getLogger(InboxPoller::class.java) @@ -30,9 +32,22 @@ class InboxPoller( log.info("polled cminmsgsId={}", id) } } + sweepBackfillTodos() return enqueued } + /** v2 §5：每轮心跳顺带对账回填补偿待办（重启即恢复；空表只花一次索引 SELECT）。 */ + private fun sweepBackfillTodos() { + try { + val outcome = backfillSweep.sweep() + if (outcome.inspected > 0) { + log.info("backfill sweep inspected={} succeeded={} failed={}", outcome.inspected, outcome.succeeded, outcome.failed) + } + } catch (e: Exception) { + log.error("backfill sweep tick failed", e) + } + } + fun loop() { running = true log.info("inbox poller loop started") diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/BackfillSweepJob.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/BackfillSweepJob.kt new file mode 100644 index 0000000..09dcc01 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/BackfillSweepJob.kt @@ -0,0 +1,53 @@ +package com.gzzn.omms.msgexchange.jobs + +import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository +import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository +import jakarta.inject.Singleton +import java.time.Duration +import java.time.Instant + +/** + * v2 §5（ACM2-29 P2-3）：共享信箱回填补偿扫描。 + * + * 业务事务提交（SUCCEEDED）后 backfillOnSuccess 失败 → 落 backfill_todo； + * 本任务到期重试：成功即删除待办，失败按指数退避推后（30s 起步，封顶 15 分钟）。 + * 回填失败绝不重放航班业务变更——待办里只有信箱回填所需的路由元数据。 + * + * 触发点：InboxPoller 心跳（重启即对账）+ JobExecutor("BACKFILL_SWEEP")（FIFO/手工）。 + */ +@Singleton +class BackfillSweepJob( + private val todo: BackfillTodoRepository, + private val inbox: CminmsgInboxRepository, + @Suppress("unused") private val props: PipelineProps, +) { + data class SweepOutcome(val inspected: Int, val succeeded: Int, val failed: Int) + + companion object { + private val INITIAL_BACKOFF: Duration = Duration.ofSeconds(30) + private val MAX_BACKOFF: Duration = Duration.ofMinutes(15) + + fun backoffDelayFor(attempts: Int): Duration { + val shift = (attempts - 1).coerceIn(0, 20) + return INITIAL_BACKOFF.multipliedBy(1L shl shift).coerceAtMost(MAX_BACKOFF) + } + } + + fun sweep(now: Instant = Instant.now()): SweepOutcome { + val due = todo.findDue(now) + var succeeded = 0 + var failed = 0 + for (task in due) { + try { + inbox.backfillOnSuccess(task.cminmsgsId, task.sndr, task.type, task.styp, task.seqn) + todo.delete(task.cminmsgsId) + succeeded++ + } catch (e: Exception) { + todo.markFailed(task.cminmsgsId, e.message, now.plus(backoffDelayFor(task.attempts + 1)), now) + failed++ + } + } + return SweepOutcome(due.size, succeeded, failed) + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobExecutor.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobExecutor.kt index c0ec732..e0d9539 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobExecutor.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobExecutor.kt @@ -18,11 +18,13 @@ class JobExecutor( private val historySweep: HistorySweepJob, private val archive: ArchiveJob, private val projectionRebuild: ProjectionRebuildJob, + private val backfillSweep: BackfillSweepJob, ) { fun execute(job: PumpJobRepository.Job) = when (job.kind) { "HISTORY_SWEEP" -> historySweep.run() "ARCHIVE" -> archive.run() "PROJECTION_REBUILD" -> projectionRebuild.run() + "BACKFILL_SWEEP" -> backfillSweep.sweep() else -> error("unknown job ${job.kind}") } } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt index a51f507..5dd557d 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt @@ -8,6 +8,7 @@ import com.gzzn.omms.msgexchange.domain.MsgEvent import com.gzzn.omms.msgexchange.domain.ProcState import com.gzzn.omms.msgexchange.domain.ProcStatus import com.gzzn.omms.msgexchange.domain.Targets +import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository @@ -121,6 +122,7 @@ class MessageProcessor( private val procFailure: ProcFailure, private val props: PipelineProps, private val txManager: PipelineTransactionManager, + private val backfillTodo: BackfillTodoRepository? = null, ) { private val log = org.slf4j.LoggerFactory.getLogger(MessageProcessor::class.java) private val flidRegex = Regex("(.*?)", RegexOption.IGNORE_CASE) @@ -188,6 +190,8 @@ class MessageProcessor( val owner = procState.ownerOfIdentity(identity) ?: -1L log.info("duplicate-of:{} -> SKIPPED id={}", owner, head.cminmsgsId) procState.update(head.cminmsgsId, ProcStatus.SKIPPED, lastError = "duplicate-of:$owner") + // 重复报文自身也是一条信箱记录：同样回填共享信箱，防止被反复轮询（失败落补偿） + compensateBackfill(head, decoded) return } } @@ -243,9 +247,33 @@ class MessageProcessor( inbox.backfillOnSuccess(head.cminmsgsId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn) } catch (e: Exception) { log.error("backfill failed after SUCCEEDED id={} (compensation required)", head.cminmsgsId, e) + recordBackfillTodo(head.cminmsgsId, decoded, e) } log.info("SUCCEEDED id={} events={} flightChanges={}", head.cminmsgsId, events.size, decision.flightChanges.size) } + + /** v2 §5：回填失败只落补偿待办（成功终态不降级，业务不重放）；由 BackfillSweepJob 到期重试。 */ + private fun compensateBackfill(head: ProcState, decoded: DecodedMessage) { + try { + inbox.backfillOnSuccess(head.cminmsgsId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn) + } catch (e: Exception) { + log.error("backfill failed for SKIPPED duplicate id={} (compensation required)", head.cminmsgsId, e) + recordBackfillTodo(head.cminmsgsId, decoded, e) + } + } + + private fun recordBackfillTodo(cminmsgsId: Long, decoded: DecodedMessage, e: Exception) { + backfillTodo?.record( + BackfillTodoRepository.BackfillTask( + cminmsgsId = cminmsgsId, + sndr = decoded.meta.sndr, + type = decoded.meta.type, + styp = decoded.meta.styp, + seqn = decoded.meta.seqn, + ), + e.message ?: e.javaClass.simpleName, + ) ?: log.warn("no backfill-todo repository bound; compensation NOT persisted id={}", cminmsgsId) + } } /** 延迟装配占位（阶段 1 后续以 Micronaut Bean 替换直连构造）。 */ diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlow.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlow.kt index 22be3a7..0f6eadb 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlow.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlow.kt @@ -8,6 +8,7 @@ import com.gzzn.omms.msgexchange.domain.ProcStatus import com.gzzn.omms.msgexchange.domain.MsgEvent import com.gzzn.omms.msgexchange.domain.MsgKind import com.gzzn.omms.msgexchange.domain.Targets +import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository import com.gzzn.omms.msgexchange.infra.persistence.FlightFields import com.gzzn.omms.msgexchange.infra.persistence.FlightFieldsJson @@ -33,6 +34,7 @@ class SnapshotFlow( private val procFailure: ProcFailure, private val txManager: PipelineTransactionManager, private val inbox: CminmsgInboxRepository, + private val backfillTodo: BackfillTodoRepository? = null, ) { private val log = org.slf4j.LoggerFactory.getLogger(SnapshotFlow::class.java) @@ -139,6 +141,16 @@ class SnapshotFlow( inbox.backfillOnSuccess(head.cminmsgsId, msg.meta.sndr, msg.meta.type, msg.meta.styp, msg.meta.seqn) } catch (e: Exception) { log.error("backfill failed after SUCCEEDED id={} (compensation required)", head.cminmsgsId, e) + backfillTodo?.record( + BackfillTodoRepository.BackfillTask( + cminmsgsId = head.cminmsgsId, + sndr = msg.meta.sndr, + type = msg.meta.type, + styp = msg.meta.styp, + seqn = msg.meta.seqn, + ), + e.message ?: e.javaClass.simpleName, + ) ?: log.warn("no backfill-todo repository bound; compensation NOT persisted id={}", head.cminmsgsId) } } diff --git a/src/main/resources/db/migration/V1.3.2__backfill_todo.sql b/src/main/resources/db/migration/V1.3.2__backfill_todo.sql new file mode 100644 index 0000000..b5c1f41 --- /dev/null +++ b/src/main/resources/db/migration/V1.3.2__backfill_todo.sql @@ -0,0 +1,18 @@ +-- flight-state-design-v2 §5 (ACM2-29 P2-3): 提交后共享信箱回填的持久补偿待办。 +-- 业务事务已提交（PROC_STATE=SUCCEEDED）后回填失败 → 落本表重试； +-- 回填失败不得把已成功的业务事务重新标记为失败，也不得重放业务变更。 + +CREATE TABLE IF NOT EXISTS backfill_todo ( + cminmsgs_id BIGINT PRIMARY KEY, + sndr VARCHAR(64) NOT NULL, + type VARCHAR(32) NOT NULL, + styp VARCHAR(32), + seqn BIGINT, + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at TIMESTAMP(6) WITH TIME ZONE NOT NULL, + last_error VARCHAR(512), + created_at TIMESTAMP(6) WITH TIME ZONE NOT NULL, + updated_at TIMESTAMP(6) WITH TIME ZONE NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_backfill_todo_due ON backfill_todo (next_attempt_at); diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/BackfillSweepJobTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/BackfillSweepJobTest.kt new file mode 100644 index 0000000..a043b4e --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/BackfillSweepJobTest.kt @@ -0,0 +1,90 @@ +package com.gzzn.omms.msgexchange.jobs + +import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository +import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository +import com.gzzn.omms.msgexchange.infra.stub.StubBackfillTodo +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import java.time.Duration +import java.time.Instant + +/** + * v2 §5（ACM2-29 P2-3）：提交后回填补偿——失败落库、到期重试、成功即清、退避推后。 + * 重启对账语义：待办持久化后，任意时刻新的 sweep 都能继续推进（不依赖内存状态）。 + */ +class BackfillSweepJobTest { + + private class FakeInbox(var fail: Boolean = false) : CminmsgInboxRepository { + val backfilled = mutableSetOf() + override fun insertRaw(rawXml: String) = 1L + override fun rawOf(cminmsgsId: Long): String? = null + override fun pollUnprocessed(afterId: Long, limit: Int): List = emptyList() + override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) { + if (fail) throw IllegalStateException("mysql-down") + backfilled += cminmsgsId + } + } + + private val t0: Instant = Instant.parse("2026-09-08T03:00:00Z") + + private fun task(id: Long = 901L) = BackfillTodoRepository.BackfillTask(id, "AODB", "SCHD", "DNLD", 9L) + + @Test + fun `due task is retried and removed after successful backfill`() { + val todo = StubBackfillTodo() + val inbox = FakeInbox(fail = false) + todo.record(task(), "mysql-down", t0) + val job = BackfillSweepJob(todo, inbox, PipelineProps()) + + val outcome = job.sweep(t0.plusSeconds(60)) + + assertEquals(BackfillSweepJob.SweepOutcome(inspected = 1, succeeded = 1, failed = 0), outcome) + assertEquals(0, todo.count()) + assertTrue(901L in inbox.backfilled) + } + + @Test + fun `failed retry increments attempts and applies exponential backoff`() { + val todo = StubBackfillTodo() + val inbox = FakeInbox(fail = true) + todo.record(task(), "mysql-down", t0) + val job = BackfillSweepJob(todo, inbox, PipelineProps()) + + val outcome = job.sweep(t0) + + assertEquals(BackfillSweepJob.SweepOutcome(inspected = 1, succeeded = 0, failed = 1), outcome) + assertEquals(1, todo.count()) + assertEquals(1, todo.tasks().single().attempts) + assertEquals("mysql-down", todo.lastErrorOf(901L)) + // 一次失败后 30s 内不再到期（退避生效） + assertEquals(emptyList(), todo.findDue(t0.plusSeconds(29))) + assertEquals(1, todo.findDue(t0.plusSeconds(30)).size) + } + + @Test + fun `recovered inbox completes compensation on a later sweep as if restarted`() { + val todo = StubBackfillTodo() + val inbox = FakeInbox(fail = true) + todo.record(task(), "mysql-down", t0) + val job = BackfillSweepJob(todo, inbox, PipelineProps()) + job.sweep(t0) + job.sweep(t0.plusSeconds(30)) // 第二次仍失败 → attempts=2 + + inbox.fail = false // 共享信箱恢复（模拟重启后对账） + val outcome = job.sweep(t0.plusSeconds(90)) + + assertEquals(BackfillSweepJob.SweepOutcome(inspected = 1, succeeded = 1, failed = 0), outcome) + assertEquals(0, todo.count()) + assertTrue(901L in inbox.backfilled) + } + + @Test + fun `backoff delay doubles per attempt and caps at fifteen minutes`() { + assertEquals(Duration.ofSeconds(30), BackfillSweepJob.backoffDelayFor(1)) + assertEquals(Duration.ofMinutes(2), BackfillSweepJob.backoffDelayFor(3)) + assertEquals(Duration.ofMinutes(15), BackfillSweepJob.backoffDelayFor(10)) + assertEquals(Duration.ofMinutes(15), BackfillSweepJob.backoffDelayFor(50)) + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/MessageProcessorTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/MessageProcessorTest.kt index 87d1333..c1aef3d 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/MessageProcessorTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/MessageProcessorTest.kt @@ -17,6 +17,7 @@ import com.gzzn.omms.msgexchange.domain.ProcState import com.gzzn.omms.msgexchange.domain.ProcStatus import com.gzzn.omms.msgexchange.domain.SchdPush import com.gzzn.omms.msgexchange.domain.Targets +import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository import com.gzzn.omms.msgexchange.infra.persistence.FlightFields import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository @@ -96,6 +97,7 @@ class MessageProcessorTest { private class FakeInbox : CminmsgInboxRepository { var raws = mutableMapOf() var throwOnRawOf: Throwable? = null + var throwOnBackfill: Throwable? = null val backfilled = mutableListOf>() override fun insertRaw(rawXml: String): Long = 0 @@ -108,6 +110,7 @@ class MessageProcessorTest { } override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) { + throwOnBackfill?.let { throw it } backfilled += listOf(cminmsgsId, sndr, type, styp, seqn) } } @@ -219,14 +222,15 @@ class MessageProcessorTest { private fun processor( procState: FakeProcState, inbox: FakeInbox, events: FakeEvents, codec: FakeCodec, registry: HandlerRegistry = HandlerRegistry(emptyList()), + backfillTodo: BackfillTodoRepository? = null, ): MessageProcessor { val props = PipelineProps() val scheduler = FailureScheduler(props, clock) val procFailure = ProcFailure(procState, scheduler) val flightSchd = FakeFlightSchd() val txManager = FakeTxManager() - val snapshot = SnapshotFlow(procState, flightSchd, events, FakeReqTrack(), procFailure, txManager, inbox) - return MessageProcessor(inbox, procState, events, CodecHolder(codec), HandlerHolder(registry), flightSchd, snapshot, procFailure, props, txManager) + val snapshot = SnapshotFlow(procState, flightSchd, events, FakeReqTrack(), procFailure, txManager, inbox, backfillTodo = backfillTodo) + return MessageProcessor(inbox, procState, events, CodecHolder(codec), HandlerHolder(registry), flightSchd, snapshot, procFailure, props, txManager, backfillTodo) } // ---------- tests ---------- @Test @@ -308,6 +312,24 @@ class MessageProcessorTest { assertTrue(ev.inserted.isEmpty()) } + @Test + fun `skipped duplicate still compensates mailbox backfill and persists todo on failure`() { + val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents() + ps.record[1] = head() + inbox.raws[1] = "" + inbox.throwOnBackfill = IllegalStateException("mysql-down") + ps.bound["AODB|FLOP|DELY|1"] = 2L + val todo = com.gzzn.omms.msgexchange.infra.stub.StubBackfillTodo() + + processor(ps, inbox, ev, FakeCodec(), backfillTodo = todo).processOne(ps.state(1)) + + // SKIPPED 不变，但共享信箱回填的补偿待办必须落库（否则重复报文被反复轮询） + assertEquals(ProcStatus.SKIPPED, ps.state(1).state) + assertEquals(1, todo.count()) + assertEquals(1L, todo.tasks().single().cminmsgsId) + assertEquals("mysql-down", todo.lastErrorOf(1L)) + } + // ---------- U08 边界化（评审要求①③） ---------- @Test fun `unexpected exception maps to FAILED INFRA with backoff at boundary`() { diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlowBackfillTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlowBackfillTest.kt index 8616e80..0405884 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlowBackfillTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlowBackfillTest.kt @@ -56,11 +56,19 @@ class SnapshotFlowBackfillTest { SnapshotFlow.StageResult.Ok(day, listOf("FL_X" to mapOf("FLID" to "FL_X"))) } - val flow = SnapshotFlow(procState, flightSchd, msgEvents, reqTrack, procFailure, txManager, inbox) + // v2 §5（P2-3）：回填失败必须落持久补偿待办（否则重启后无从对账） + val todo = com.gzzn.omms.msgexchange.infra.stub.StubBackfillTodo() + val flow = SnapshotFlow(procState, flightSchd, msgEvents, reqTrack, procFailure, txManager, inbox, backfillTodo = todo) flow.publishSnapshot(ProcState(headId, ProcStatus.PENDING), decoded) assertEquals(ProcStatus.SUCCEEDED, procState.snapshotOf(headId)!!.state) assertTrue(flightSchd.findByFlid("FL_X") != null) + assertEquals(1, todo.count()) + val recorded = todo.tasks().single() + assertEquals(901L, recorded.cminmsgsId) + assertEquals("SCHD", recorded.type) + assertEquals("DNLD", recorded.styp) + assertEquals("mysql-down", todo.lastErrorOf(901L)) } @Test