From 09b53c77bbd0454832b8fab3ce498beb219baaeb Mon Sep 17 00:00:00 2001 From: windyboy Date: Thu, 10 Sep 2026 10:59:58 +0800 Subject: [PATCH] =?UTF-8?q?refactor(ingress):=20=E4=BF=A1=E7=AE=B1?= =?UTF-8?q?=E8=BE=B9=E7=95=8C=E5=B1=82=E5=A5=91=E7=BA=A6=E9=87=8D=E6=9E=84?= =?UTF-8?q?=E2=80=94=E2=80=94=E5=8F=91=E7=8E=B0=E4=B8=8E=E5=A4=84=E7=90=86?= =?UTF-8?q?=E6=A0=87=E8=AE=B0=E8=A7=A3=E8=80=A6=E3=80=81=E5=9B=9E=E5=A1=AB?= =?UTF-8?q?=E4=BA=8B=E5=AE=9E=E5=B9=B6=E5=85=A5=20PROC=5FSTATE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 收报扫描不再以 DATE_PROCESSED 为谓词：终态而未回填的行（解码失败死信等）会永久占据 有限批次，累积到 claim-batch 后收报整体停摆（US-01 条目 3 / message-lifecycle §5.3）。 ingress/InboxPoller.kt：按 ID 区间升序有界读取（ID > W），水位落 INBOX_CURSOR 并与入队 同一 PG 事务推进（中断后重扫补建）；遇空洞即停，空洞超过 pipeline.max-commit-delay 判定 为永久并放行——否则水位永久停摆于一次自增回滚留下的空位。删除 InboxEnqueue（改由 insertIfAbsent 幂等入队）与 poller 内的回填扫描（消除 ingress→jobs 反向依赖）。 infra/persistence：端口按事实重画为 readRange/maxId/markProcessedIfUnmarked，标记 UPDATE 带 DATE_PROCESSED IS NULL 守卫，只把空标写为已处理（§11 单调，重复执行无副作用）。 回填事实并入 PROC_STATE（RECEIVED_AT/BACKFILL_AT/NEXT_AT/ATTEMPTS/ERROR），BACKFILL_TODO 随 V2 迁移下线；终态与回填意图是同一条 UPDATE，由处理器在自己的业务事务内落库， message-lifecycle §4 登记的两个崩溃窗口（提交后回填前崩溃、待办二次落账失败）不再是缺口。 processing/BackfillService.kt（取代 BackfillSweepJob）：终态提交后立即尝试一次，失败按 30s→15min 指数退避重试；扫描条件「终态 + 未确认标记 +（已到期 或 接收时间早于 NOW − R）」 使 §5.2 的超期期限 R 覆盖退避，中间态永不补写。死信同样可补写——回填只需消息 ID， 不再依赖 META。Pump 改用可注入 Clock。 infra/health：InboxLifecycleHealthIndicator 输出积压条数、最老未处理信龄、未回填终态数与 水位滞后（OPS-2 / §5.3 验收）。预计消化时长需吞吐采样，留待接入指标注册表时补。 配置：pipeline.max-commit-delay / overdue-backfill / backfill-batch、mailbox.processed-value （Q2/Q6/Q7 未书面确认前取保守初值，不得为提速下调）。 不变量回归测试：死信不阻断后续发现、水位遇空洞即停与老化放行、终态+意图同事务、 超期 R 覆盖退避、中间态不补写、标记单调；InboxLifecycleJdbcSqlTest 以 H2 的 PostgreSQL 兼容模式直连验证上述 SQL 语义（不依赖 docker）。libs.h2 由 testRuntimeOnly 提为 testImplementation 以支持该用例。 验证：gradle clean test --offline → 78 tests / 0 failures / 1 skipped （PG Testcontainers 集成用例在本机无 docker 时按既有约定 assumeTrue 跳过）。 --- build.gradle.kts | 4 +- .../omms/msgexchange/config/MailboxProps.kt | 6 + .../omms/msgexchange/config/PipelineProps.kt | 16 ++ .../gzzn/omms/msgexchange/domain/ProcState.kt | 17 +- .../health/InboxLifecycleHealthIndicator.kt | 74 ++++++ .../infra/persistence/Repositories.kt | 96 +++++--- .../jdbc/JdbcCminmsgInboxRepository.kt | 40 +++- .../persistence/jdbc/JdbcPgRepositories.kt | 221 ++++++++++++------ .../msgexchange/infra/retry/ProcFailure.kt | 24 +- .../infra/stub/StubRepositories.kt | 157 +++++++++---- .../omms/msgexchange/ingress/InboxEnqueue.kt | 16 -- .../omms/msgexchange/ingress/InboxPoller.kt | 93 +++++--- .../omms/msgexchange/ingress/InboxService.kt | 14 +- .../omms/msgexchange/jobs/BackfillSweepJob.kt | 53 ----- .../gzzn/omms/msgexchange/jobs/JobRunner.kt | 8 +- .../msgexchange/processing/BackfillService.kt | 74 ++++++ .../processing/DynamicProcessors.kt | 39 ++-- .../gzzn/omms/msgexchange/processing/Pump.kt | 143 +++++------- .../processing/ScheduleProcessor.kt | 15 +- src/main/resources/application.yml | 9 +- .../db/migration/V2__inbox_lifecycle.sql | 44 ++++ .../omms/msgexchange/PipelineSmokeTest.kt | 25 ++ .../infra/health/HealthIndicatorsTest.kt | 37 +++ .../persistence/jdbc/FlywayMigrationTest.kt | 45 +++- .../jdbc/InboxLifecycleJdbcSqlTest.kt | 221 ++++++++++++++++++ .../infra/retry/ReplayServiceTest.kt | 17 +- .../msgexchange/ingress/InboxPollerTest.kt | 121 ++++++++-- .../msgexchange/jobs/BackfillSweepJobTest.kt | 90 ------- .../processing/BackfillServiceTest.kt | 170 ++++++++++++++ .../processing/FdelAndAdftProcessorTest.kt | 23 +- .../processing/ScheduleProcessorTest.kt | 27 +-- src/test/resources/application-test.yml | 11 + 32 files changed, 1402 insertions(+), 548 deletions(-) create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/InboxLifecycleHealthIndicator.kt delete mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxEnqueue.kt delete mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/jobs/BackfillSweepJob.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/processing/BackfillService.kt create mode 100644 src/main/resources/db/migration/V2__inbox_lifecycle.sql create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/InboxLifecycleJdbcSqlTest.kt delete mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/jobs/BackfillSweepJobTest.kt create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/processing/BackfillServiceTest.kt diff --git a/build.gradle.kts b/build.gradle.kts index 405335b..5e2bd59 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -59,7 +59,9 @@ dependencies { runtimeOnly(libs.snakeyaml) // U02：显式版本入 catalog（原无版本号依赖 BOM 覆盖，snakeyaml 不在 micronaut BOM 内） testImplementation(libs.junit.jupiter) - testRuntimeOnly(libs.h2) // U03：配置绑定回归测试的测试库（内存 datasource，冒烟用） + // U03：配置绑定回归测试的测试库（内存 datasource，冒烟用）； + // V2 起 InboxLifecycleJdbcSqlTest 以 H2 的 PostgreSQL 兼容模式直连验证生命周期 SQL 语义 + testImplementation(libs.h2) testImplementation("io.micronaut.test:micronaut-test-junit5") testImplementation("org.jetbrains.kotlin:kotlin-test") // P3-A（ACM2-29）：无外部 PG 时 JDBC 集成测试自动起隔离容器；有 MSGX_PG_* 环境变量时不启用 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/config/MailboxProps.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/config/MailboxProps.kt index 63f2041..2aa1184 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/config/MailboxProps.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/config/MailboxProps.kt @@ -4,6 +4,12 @@ import io.micronaut.context.annotation.ConfigurationProperties @ConfigurationProperties("mailbox") class MailboxProps { + /** + * 处理标记写入值（message-lifecycle §5.2/Q7）：仅限库方认可的 legacy 值集； + * 值集与写权限书面确认前保持 legacy 现役值。 + */ + var processedValue: String = "PROCESSED" + var sharedMysql: SharedMysql = SharedMysql() @ConfigurationProperties("shared-mysql") diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt index f16015d..ec11ed1 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt @@ -25,6 +25,22 @@ class PipelineProps { var backoffCapMs: Long = 60_000 var headDeadline: Duration = Duration.ofMinutes(10) // 最坏 HOL 上界（毒丸升级） + /** + * message-lifecycle §5.1 空洞老化：水位 W+1 处的空洞持续超过该时延即判定为永久并放行。 + * 取值口径 = 库方承诺的最大提交时延（Q2）；太小会把迟到消息判成永久空洞（FIFO 越序风险）。 + */ + var maxCommitDelay: Duration = Duration.ofMinutes(5) + + /** + * message-lifecycle §5.2 超期补写期限 R：终态后仍无处理标记的行到达该期限即强制补写， + * 保证库方清除前提「边界内无未标记行」在有限时间内成立。 + * R ≥ 人工重放期限 + 人工处置期限（Q6）；确认前不得下调。 + */ + var overdueBackfill: Duration = Duration.ofDays(30) + + /** 回填扫描单批条数。 */ + var backfillBatch: Int = 100 + /** U07：启动即拉起 Pump/Dispatcher 循环（默认关——需要真实仓储或 msgx.stubs=true 才可安全开启）。 */ var autostart: Boolean = false diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/ProcState.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/ProcState.kt index 6953365..f0ad5ba 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/ProcState.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/ProcState.kt @@ -1,10 +1,15 @@ package com.gzzn.omms.msgexchange.domain +import java.time.Instant + /** * PROC_STATE 处理伴生状态（docs/design.md §2.1/§2.3）：每消息一行， * MSG_ID = 信箱 ID 主键防重复入队；IDENTITY_KEY 唯一约束防业务重复； * 处理状态机与错误分类同设计文档 §2.3。SUCCEEDED 终态兼作 SCHD 快照 * 重放判定（docs/flight-state.md §4 步骤 / design.md §4.1）。 + * + * 回填事实（message-lifecycle.md §5.2/§11）与处理事实同体同行：终态与回填意图 + * 由同一条 UPDATE 落下，因此不存在"业务已提交、回填待办未记"的崩溃窗口。 */ enum class ProcStatus { PENDING, FAILED, SUCCEEDED, SKIPPED, DEAD } @@ -16,8 +21,16 @@ data class ProcState( val state: ProcStatus, val identityKey: String? = null, // SNDR|TYPE|STYP|SEQN（design.md §2.2）；decode 后首次绑定，FAILED 重试不重绑 val attempts: Int = 0, - val nextAttemptAt: java.time.Instant? = null, + val nextAttemptAt: Instant? = null, val errorClass: ErrorClass? = null, val lastError: String? = null, - val updatedAt: java.time.Instant = java.time.Instant.now(), + /** 信箱 CMINMSGS_DATE_RECEIVED：§5.2 超期补写的 R 判据与 OPS-2「最老未处理信龄」锚点。 */ + val receivedAt: Instant? = null, + /** 非空 = 已确认信箱行持有处理标记（回填完成）。 */ + val backfillAt: Instant? = null, + /** 非空 = 待回填；终态事务内登记为 now，失败按退避推后。 */ + val backfillNextAt: Instant? = null, + val backfillAttempts: Int = 0, + val backfillError: String? = null, + val updatedAt: Instant = Instant.now(), ) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/InboxLifecycleHealthIndicator.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/InboxLifecycleHealthIndicator.kt new file mode 100644 index 0000000..bf499cc --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/InboxLifecycleHealthIndicator.kt @@ -0,0 +1,74 @@ +package com.gzzn.omms.msgexchange.infra.health + +import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository +import com.gzzn.omms.msgexchange.infra.persistence.InboxCursorRepository +import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository +import io.micronaut.context.BeanProvider +import io.micronaut.core.async.publisher.Publishers +import io.micronaut.health.HealthStatus +import io.micronaut.management.health.indicator.HealthIndicator +import io.micronaut.management.health.indicator.HealthResult +import jakarta.inject.Singleton +import org.reactivestreams.Publisher +import java.time.Duration +import java.time.Instant + +/** + * 信箱生命周期观测（docs/message-lifecycle.md §5.3 验收 / user-stories.md OPS-2）： + * 输出剩余积压、最老未处理信龄、未回填终态条数与水位滞后，供积压消化期间持续观察。 + * + * 端口缺省（未接通共享信箱或自有 PG）时报告未绑定而不判 DOWN——可用性由各依赖自身的 + * 健康指示器承担，本指示器只反映生命周期状态；端口查询失败判 DOWN。 + */ +@Singleton +class InboxLifecycleHealthIndicator( + private val procState: BeanProvider, + private val cursor: BeanProvider, + private val mailbox: BeanProvider, +) : HealthIndicator { + + override fun getResult(): Publisher = + Publishers.just( + runCatching { + lifecycleHealth( + procState = if (procState.isPresent) procState.get() else null, + cursor = if (cursor.isPresent) cursor.get() else null, + mailbox = if (mailbox.isPresent) mailbox.get() else null, + ) + }.getOrElse { down(it) }, + ) +} + +internal fun lifecycleHealth( + procState: ProcStateRepository?, + cursor: InboxCursorRepository?, + mailbox: CminmsgInboxRepository?, + now: Instant = Instant.now(), +): HealthResult { + if (procState == null) { + return HealthResult.builder(NAME).status(HealthStatus.UP) + .details(mapOf("message" to "proc_state repository not bound (stub off, impl pending)")) + .build() + } + val backlog = procState.backlog() + val watermark = cursor?.load()?.committedUpTo + val maxId = runCatching { mailbox?.maxId() }.getOrNull() + return HealthResult.builder(NAME).status(HealthStatus.UP).details( + linkedMapOf( + "backlog" to backlog.unfinished, + "oldestUnprocessedSeconds" to ( + backlog.oldestReceivedAt?.let { Duration.between(it, now).seconds } ?: -1L + ), + "unmarkedTerminal" to backlog.unmarkedTerminal, + "watermark" to (watermark ?: -1L), + "watermarkLag" to if (watermark != null && maxId != null) maxId - watermark else -1L, + ), + ).build() +} + +private fun down(error: Throwable): HealthResult = + HealthResult.builder(NAME).status(HealthStatus.DOWN) + .details(mapOf("message" to (error.message ?: error.javaClass.simpleName))) + .build() + +private const val NAME = "inbox-lifecycle" 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 6243294..118511f 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 @@ -29,12 +29,17 @@ interface PipelineLockRepository { fun lock() } -/** PROC_STATE：每消息一行（design.md §2.1）；SUCCEEDED 终态兼作日计划重放判定。 */ +/** + * PROC_STATE：每消息一行（design.md §2.1）；SUCCEEDED 终态兼作日计划重放判定。 + * 回填事实（BACKFILL_AT/NEXT_AT/ATTEMPTS/ERROR）与处理事实同行，取代独立的回填待办表。 + */ interface ProcStateRepository { - fun insert(msgId: Long, state: ProcStatus = ProcStatus.PENDING) - - /** 主路径轮询/compat 入队前判重（PG 已有行则跳过）。 */ - fun exists(msgId: Long): Boolean + /** + * 入队：MSG_ID 主键幂等（重复扫描与 compat 入口并发都不会重复建行）。 + * @param receivedAt 信箱 DATE_RECEIVED，用于 §5.2 超期判据与信龄观测。 + * @return true = 本次实际新建 + */ + fun insertIfAbsent(msgId: Long, receivedAt: Instant?): Boolean fun find(msgId: Long): ProcState? @@ -48,6 +53,7 @@ interface ProcStateRepository { fun ownerOfIdentity(identityKey: String): Long? + /** 非终态迁移（PENDING / FAILED 及退避），不触碰回填列。 */ fun update( msgId: Long, state: ProcStatus, @@ -57,10 +63,44 @@ interface ProcStateRepository { lastError: String? = null, ) + /** + * 终态 + 回填意图同一条 UPDATE（message-lifecycle §2/§4），由处理器在自己的业务 + * 事务内调用：航班变更、事件、终态、回填意图同提交同回滚。 + */ + fun markTerminal( + msgId: Long, + state: ProcStatus, + errorClass: ErrorClass? = null, + lastError: String? = null, + attempts: Int? = null, + now: Instant = Instant.now(), + ) + + /** 信箱行已确认持有处理标记。 */ + fun markBackfilled(msgId: Long, now: Instant = Instant.now()) + + /** 回填失败：次数 +1、按退避推后、留错误；终态不得回改（§11）。 */ + fun recordBackfillFailure(msgId: Long, error: String?, attempts: Int, nextAttemptAt: Instant, now: Instant) + + /** + * 待回填待办（message-lifecycle §3/§5.2）：终态 + 未确认标记， + * 且（已到期 或 接收时间已达超期期限 overdueBefore）。 + */ + fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List + /** 显式重放入口：仅把给定 errorClass 集合中的行从 FAILED/DEAD 置回 PENDING（ATTEMPTS=0）。 */ fun requeueByErrorClasses(errorClasses: List): Int + + /** OPS-2 观测口径（message-lifecycle §5.3 验收）：积压、最老信龄锚点、未回填终态。 */ + fun backlog(): Backlog } +/** 某条待回填记录（扫描输入）。 */ +data class BackfillDue(val msgId: Long, val attempts: Int) + +/** 处理侧积压快照。 */ +data class Backlog(val unfinished: Int, val oldestReceivedAt: Instant?, val unmarkedTerminal: Int) + /** MSG_EVENT outbox（flight-state.md §5）。KAFKA_SCHD 合并同 FLID 未发事件按最新 STATE_VERSION 输出。 */ interface MsgEventRepository { fun insertAll(events: List): List @@ -168,41 +208,39 @@ interface ReqTrackRepository { } /** - * 共享信箱回填补偿待办（design.md §3.3/§6.1）：业务事务内预登记， - * 提交后由 BackfillSweepJob 重试；回填失败不得把 SUCCEEDED 改回 FAILED。 + * 消费水位 W（message-lifecycle §5.1）：单行游标，只随新 ID 成功入队推进、遇空洞即停， + * 与入队在同一 PG 事务提交——中断后 W 未前进，重扫即补建（§4 第一行）。 + * + * `holeSince` 记录 W+1 处空洞首次被观测到的时刻：超过最大提交时延（Q2 承诺）即判定为 + * 永久空洞并放行，否则水位会永久停摆于一次自增回滚留下的空位，后续 ID 再无入队机会。 */ -interface BackfillTodoRepository { - data class BackfillTask( - val msgId: Long, - val sndr: String, - val type: String, - val styp: String, - val seqn: Long, - val attempts: Int = 0, - ) +interface InboxCursorRepository { + data class Cursor(val committedUpTo: Long = 0L, val holeSince: Instant? = null) - /** 失败即落库（幂等 upsert，同 ID 重复失败只刷新错误与重试时间）。 */ - fun record(task: BackfillTask, lastError: String?, now: Instant = Instant.now()) + fun load(): Cursor - fun findDue(now: Instant = Instant.now(), limit: Int = 50): List - - fun markFailed(msgId: Long, lastError: String?, nextAttemptAt: Instant, now: Instant = Instant.now()) - - fun delete(msgId: Long) - - fun count(): Int + fun save(cursor: Cursor) } /** - * 共享 MySQL 信箱 CMINMSGS 访问（他人系统库，本系统不建表）。 - * 主路径 JDBC 轮询读 + 处理回填；compat HTTP 写；出站写 COUTMSGS 由出站适配层承担。 + * 共享 MySQL 信箱 CMINMSGS 访问（他人系统库，本系统不建表，只做 DML）。 + * 三个事实互不替代（message-lifecycle §5.1/§11）：**发现**按 ID 区间读、**水位**只表示 + * 读取进度、**处理标记**只用于回填与库方清除——标记不得作为扫描谓词。 */ interface CminmsgInboxRepository { fun insertRaw(rawXml: String): Long fun rawOf(msgId: Long): String? - fun pollUnprocessed(afterId: Long, limit: Int): List + /** 按 ID 区间升序有界读取（`ID > fromExclusive`），不以处理标记为谓词。 */ + fun readRange(fromExclusive: Long, limit: Int): List - fun backfillOnSuccess(msgId: Long, sndr: String, type: String, styp: String, seqn: Long) + /** 信箱当前最大 ID（空表 null）；仅用于观测水位滞后。 */ + fun maxId(): Long? + + /** 只把空标写为已处理（§11 单调）；返回 true = 本次实际写入，重复执行无副作用。 */ + fun markProcessedIfUnmarked(msgId: Long, value: String): Boolean } + +/** 信箱行读取结果（发现阶段只需要身份与接收时间，原文按需再取）。 */ +data class MailboxRow(val msgId: Long, val receivedAt: Instant?) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcCminmsgInboxRepository.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcCminmsgInboxRepository.kt index 940a6c4..27b81e8 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcCminmsgInboxRepository.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcCminmsgInboxRepository.kt @@ -1,14 +1,18 @@ package com.gzzn.omms.msgexchange.infra.persistence.jdbc import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository +import com.gzzn.omms.msgexchange.infra.persistence.MailboxRow import io.micronaut.context.annotation.Requires import jakarta.inject.Named import jakarta.inject.Singleton import javax.sql.DataSource /** - * 共享 MySQL CMINMSGS 信箱适配（ACM2-12：仅 DML，不建表）。 + * 共享 MySQL CMINMSGS 信箱适配（仅 DML，不建表；message-lifecycle §5.1/§6/§11）。 * 列名与 legacy `entity/Cminmsg.java` 一致。 + * + * 发现按 ID 区间读（`ID > ?`），**不以 `DATE_PROCESSED` 为扫描谓词**：已入队但尚未 + * 回填的行否则会永久占据批次，正是 §5.1 与 US-01 条目 3 要求排除的场景。 */ @Singleton @Requires(property = "msgx.stubs", notEquals = "true") @@ -33,31 +37,45 @@ class JdbcCminmsgInboxRepository( { ps -> ps.setLong(1, msgId) }, ) { rs -> rs.getString("CMINMSGS_CLOB_MSG") } - override fun pollUnprocessed(afterId: Long, limit: Int): List = + override fun readRange(fromExclusive: Long, limit: Int): List = ds.query( """ - SELECT CMINMSGS_ID FROM cminmsgs - WHERE CMINMSGS_ID > ? AND CMINMSGS_DATE_PROCESSED IS NULL + SELECT CMINMSGS_ID, CMINMSGS_DATE_RECEIVED FROM cminmsgs + WHERE CMINMSGS_ID > ? ORDER BY CMINMSGS_ID ASC LIMIT ? """.trimIndent(), { ps -> - ps.setLong(1, afterId) + ps.setLong(1, fromExclusive) ps.setInt(2, limit) }, - ) { rs -> rs.getLong("CMINMSGS_ID") } + ) { rs -> + MailboxRow( + msgId = rs.getLong("CMINMSGS_ID"), + receivedAt = rs.getTimestamp("CMINMSGS_DATE_RECEIVED")?.toInstant(), + ) + } - override fun backfillOnSuccess(msgId: Long, sndr: String, type: String, styp: String, seqn: Long) { + override fun maxId(): Long? = + ds.queryOne("SELECT MAX(CMINMSGS_ID) AS max_id FROM cminmsgs", {}) { rs -> + rs.getLong("max_id").takeIf { !rs.wasNull() } + } + + /** + * §11 单调：`DATE_PROCESSED IS NULL` 守卫保证只把空标写为已处理，已有值不回撤、 + * 不覆盖；影响 0 行 = 已被其他路径标记，调用方按幂等成功处理。 + * 写入值（DATE_PROCESSED 时间语义与 STATUS 值集）以库方契约为准（Q7）。 + */ + override fun markProcessedIfUnmarked(msgId: Long, value: String): Boolean = ds.update( """ UPDATE cminmsgs SET CMINMSGS_DATE_PROCESSED = CURRENT_TIMESTAMP, CMINMSGS_STATUS = ? - WHERE CMINMSGS_ID = ? + WHERE CMINMSGS_ID = ? AND CMINMSGS_DATE_PROCESSED IS NULL """.trimIndent(), ) { ps -> - ps.setString(1, "PROCESSED") + ps.setString(1, value) ps.setLong(2, msgId) - } - } + } == 1 } 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 f1e39eb..e9672c9 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 @@ -11,9 +11,11 @@ import com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot import com.gzzn.omms.msgexchange.domain.flight.FlightState import com.gzzn.omms.msgexchange.domain.flight.HistoryCandidate import com.gzzn.omms.msgexchange.domain.flight.HistoryRules -import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository +import com.gzzn.omms.msgexchange.infra.persistence.BackfillDue +import com.gzzn.omms.msgexchange.infra.persistence.Backlog import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository import com.gzzn.omms.msgexchange.infra.persistence.FlightStateRepository +import com.gzzn.omms.msgexchange.infra.persistence.InboxCursorRepository import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository import com.gzzn.omms.msgexchange.infra.persistence.PersistOutcome import com.gzzn.omms.msgexchange.infra.persistence.PipelineLockRepository @@ -68,22 +70,20 @@ class JdbcPipelineLockRepository( class JdbcProcStateRepository( private val ds: DataSource, ) : ProcStateRepository { - override fun insert(msgId: Long, state: ProcStatus) { + /** MSG_ID 主键幂等入队：重复扫描与 compat 入口并发都不重复建行（§5.1）。 */ + override fun insertIfAbsent(msgId: Long, receivedAt: Instant?): Boolean = ds.update( - "INSERT INTO proc_state (msg_id, state, updated_at) VALUES (?, ?, ?)", - { ps -> ps.setLong(1, msgId); ps.setString(2, state.name); ps.setTimestamp(3, Instant.now().toSqlTimestamp()) }, - ) - } - - override fun exists(msgId: Long): Boolean = - ds.queryOne("SELECT 1 FROM proc_state WHERE msg_id = ?", { ps -> ps.setLong(1, msgId) }) { 1 } != null + "INSERT INTO proc_state (msg_id, state, received_at, updated_at) VALUES (?, 'PENDING', ?, ?) " + + "ON CONFLICT (msg_id) DO NOTHING", + { ps -> + ps.setLong(1, msgId) + ps.setTimestamp(2, receivedAt?.toSqlTimestamp()) + ps.setTimestamp(3, Instant.now().toSqlTimestamp()) + }, + ) == 1 override fun find(msgId: Long): ProcState? = - ds.queryOne( - "SELECT msg_id, state, identity_key, attempts, next_attempt_at, error_class, last_error, updated_at FROM proc_state WHERE msg_id = ?", - { ps -> ps.setLong(1, msgId) }, - ::mapProcState, - ) + ds.queryOne("$SELECT_PROC WHERE msg_id = ?", { ps -> ps.setLong(1, msgId) }, ::mapProcState) override fun findSuccessTerminal(msgId: Long): Boolean = ds.queryOne( @@ -93,8 +93,7 @@ class JdbcProcStateRepository( override fun headUnfinished(): ProcState? = ds.queryOne( - "SELECT msg_id, state, identity_key, attempts, next_attempt_at, error_class, last_error, updated_at " + - "FROM proc_state WHERE state IN ('PENDING', 'FAILED') ORDER BY msg_id ASC LIMIT 1", + "$SELECT_PROC WHERE state IN ('PENDING', 'FAILED') ORDER BY msg_id ASC LIMIT 1", {}, ::mapProcState, ) @@ -145,6 +144,80 @@ class JdbcProcStateRepository( ) } + /** 终态与回填意图同一条 UPDATE：业务事务内调用即原子提交（message-lifecycle §2/§4）。 */ + override fun markTerminal( + msgId: Long, + state: ProcStatus, + errorClass: ErrorClass?, + lastError: String?, + attempts: Int?, + now: Instant, + ) { + ds.update( + """ + UPDATE proc_state + SET state = ?, error_class = ?, last_error = ?, attempts = COALESCE(?, attempts), + next_attempt_at = NULL, + backfill_at = NULL, backfill_next_at = ?, backfill_attempts = 0, backfill_error = NULL, + updated_at = ? + WHERE msg_id = ? + """.trimIndent(), + { ps -> + ps.setString(1, state.name) + ps.setString(2, errorClass?.name) + ps.setString(3, lastError?.take(1000)) + attempts?.let { ps.setInt(4, it) } ?: ps.setNull(4, java.sql.Types.INTEGER) + ps.setTimestamp(5, now.toSqlTimestamp()) + ps.setTimestamp(6, now.toSqlTimestamp()) + ps.setLong(7, msgId) + }, + ) + } + + override fun markBackfilled(msgId: Long, now: Instant) { + ds.update( + "UPDATE proc_state SET backfill_at = ?, backfill_next_at = NULL, backfill_error = NULL, updated_at = ? WHERE msg_id = ?", + { ps -> + ps.setTimestamp(1, now.toSqlTimestamp()) + ps.setTimestamp(2, now.toSqlTimestamp()) + ps.setLong(3, msgId) + }, + ) + } + + override fun recordBackfillFailure(msgId: Long, error: String?, attempts: Int, nextAttemptAt: Instant, now: Instant) { + ds.update( + "UPDATE proc_state SET backfill_attempts = ?, backfill_next_at = ?, backfill_error = ?, updated_at = ? WHERE msg_id = ?", + { ps -> + ps.setInt(1, attempts) + ps.setTimestamp(2, nextAttemptAt.toSqlTimestamp()) + ps.setString(3, error?.take(512)) + ps.setTimestamp(4, now.toSqlTimestamp()) + ps.setLong(5, msgId) + }, + ) + } + + /** + * 待回填：终态 + 未确认标记，且已到期或已达 §5.2 超期期限（R 覆盖退避，保证 + * 有限时间内必然补写，否则库方清除的前提"边界内无未标记行"无法成立）。 + */ + override fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List = + ds.query( + """ + SELECT msg_id, backfill_attempts FROM proc_state + WHERE backfill_at IS NULL + AND state IN ('SUCCEEDED', 'SKIPPED', 'DEAD') + AND (backfill_next_at IS NULL OR backfill_next_at <= ? OR (received_at IS NOT NULL AND received_at < ?)) + ORDER BY msg_id ASC LIMIT ? + """.trimIndent(), + { ps -> + ps.setTimestamp(1, now.toSqlTimestamp()) + ps.setTimestamp(2, overdueBefore.toSqlTimestamp()) + ps.setInt(3, limit) + }, + ) { rs -> BackfillDue(rs.getLong("msg_id"), rs.getInt("backfill_attempts")) } + override fun requeueByErrorClasses(errorClasses: List): Int { if (errorClasses.isEmpty()) return 0 val placeholders = errorClasses.joinToString(",") { "?" } @@ -158,6 +231,25 @@ class JdbcProcStateRepository( ) } + /** OPS-2（message-lifecycle §5.3）：积压条数、最老未处理接收时刻、未回填终态条数。 */ + override fun backlog(): Backlog = + ds.queryOne( + """ + SELECT + count(*) FILTER (WHERE state IN ('PENDING', 'FAILED')) AS unfinished, + min(received_at) FILTER (WHERE state IN ('PENDING', 'FAILED')) AS oldest_received_at, + count(*) FILTER (WHERE state IN ('SUCCEEDED', 'SKIPPED', 'DEAD') AND backfill_at IS NULL) AS unmarked_terminal + FROM proc_state + """.trimIndent(), + {}, + ) { rs -> + Backlog( + unfinished = rs.getInt("unfinished"), + oldestReceivedAt = rs.getInstant("oldest_received_at"), + unmarkedTerminal = rs.getInt("unmarked_terminal"), + ) + } ?: Backlog(0, null, 0) + private fun mapProcState(rs: ResultSet) = ProcState( msgId = rs.getLong("msg_id"), state = ProcStatus.valueOf(rs.getString("state")), @@ -166,8 +258,45 @@ class JdbcProcStateRepository( nextAttemptAt = rs.getInstant("next_attempt_at"), errorClass = rs.getString("error_class")?.let(ErrorClass::valueOf), lastError = rs.getString("last_error"), + receivedAt = rs.getInstant("received_at"), + backfillAt = rs.getInstant("backfill_at"), + backfillNextAt = rs.getInstant("backfill_next_at"), + backfillAttempts = rs.getInt("backfill_attempts"), + backfillError = rs.getString("backfill_error"), updatedAt = rs.getInstant("updated_at") ?: Instant.now(), ) + + private companion object { + const val SELECT_PROC = + "SELECT msg_id, state, identity_key, attempts, next_attempt_at, error_class, last_error, " + + "received_at, backfill_at, backfill_next_at, backfill_attempts, backfill_error, updated_at FROM proc_state" + } +} + +/** 消费水位单行游标（message-lifecycle §5.1）；与入队同事务写入。 */ +@Singleton +@Requires(property = "datasources.default.enabled", value = "true") +@Requires(missingProperty = "msgx.stubs") +class JdbcInboxCursorRepository( + private val ds: DataSource, +) : InboxCursorRepository { + override fun load(): InboxCursorRepository.Cursor = + ds.queryOne( + "SELECT committed_up_to, hole_since FROM inbox_cursor WHERE cursor_id = 1", + {}, + ) { rs -> InboxCursorRepository.Cursor(rs.getLong("committed_up_to"), rs.getInstant("hole_since")) } + ?: InboxCursorRepository.Cursor() + + override fun save(cursor: InboxCursorRepository.Cursor) { + ds.update( + "UPDATE inbox_cursor SET committed_up_to = ?, hole_since = ?, updated_at = ? WHERE cursor_id = 1", + { ps -> + ps.setLong(1, cursor.committedUpTo) + ps.setTimestamp(2, cursor.holeSince?.toSqlTimestamp()) + ps.setTimestamp(3, Instant.now().toSqlTimestamp()) + }, + ) + } } @Singleton @@ -673,64 +802,4 @@ class JdbcReqTrackRepository( ) } -@Singleton -@Requires(property = "datasources.default.enabled", value = "true") -@Requires(missingProperty = "msgx.stubs") -class JdbcBackfillTodoRepository( - private val ds: DataSource, -) : BackfillTodoRepository { - override fun record(task: BackfillTodoRepository.BackfillTask, lastError: String?, now: Instant) { - ds.update( - """ - INSERT INTO backfill_todo (msg_id, sndr, type, styp, seqn, attempts, next_attempt_at, last_error, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, 0, ?, ?, ?, ?) - ON CONFLICT (msg_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.msgId) - ps.setString(2, task.sndr) - ps.setString(3, task.type) - ps.setString(4, task.styp) - ps.setLong(5, task.seqn) - ps.setTimestamp(6, now.toSqlTimestamp()) - ps.setString(7, lastError?.take(512)) - ps.setTimestamp(8, now.toSqlTimestamp()) - ps.setTimestamp(9, now.toSqlTimestamp()) - }, - ) - } - override fun findDue(now: Instant, limit: Int): List = - ds.query( - "SELECT msg_id, sndr, type, styp, seqn, attempts FROM backfill_todo WHERE next_attempt_at <= ? ORDER BY next_attempt_at ASC LIMIT ?", - { ps -> ps.setTimestamp(1, now.toSqlTimestamp()); ps.setInt(2, limit) }, - ) { rs -> - BackfillTodoRepository.BackfillTask( - msgId = rs.getLong("msg_id"), - sndr = rs.getString("sndr"), - type = rs.getString("type"), - styp = rs.getString("styp"), - seqn = rs.getLong("seqn"), - attempts = rs.getInt("attempts"), - ) - } - - override fun markFailed(msgId: Long, lastError: String?, nextAttemptAt: Instant, now: Instant) { - ds.update( - "UPDATE backfill_todo SET attempts = attempts + 1, last_error = ?, next_attempt_at = ?, updated_at = ? WHERE msg_id = ?", - { ps -> - ps.setString(1, lastError?.take(512)) - ps.setTimestamp(2, nextAttemptAt.toSqlTimestamp()) - ps.setTimestamp(3, now.toSqlTimestamp()) - ps.setLong(4, msgId) - }, - ) - } - - override fun delete(msgId: Long) { - ds.update("DELETE FROM backfill_todo WHERE msg_id = ?", { ps -> ps.setLong(1, msgId) }) - } - - override fun count(): Int = - ds.queryOne("SELECT count(*) AS n FROM backfill_todo", {}) { rs -> rs.getInt("n") } ?: 0 -} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/retry/ProcFailure.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/retry/ProcFailure.kt index c421af9..17642f0 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/retry/ProcFailure.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/retry/ProcFailure.kt @@ -8,7 +8,7 @@ import jakarta.inject.Singleton /** * ProcState 侧统一失败迁移（U08/U10）：处理/快照路径共用—— - * attempts+1 后若 exhausted → DEAD(EXHAUSTED)（终态，errorClass 规范化，原因保留在 lastError）； + * attempts+1 后若 exhausted → DEAD(EXHAUSTED)（终态 + 回填意图，errorClass 规范化，原因保留在 lastError）； * 否则 FAILED + attempts + nextAttemptAt(退避)（可重放）。任何“失败”都不得在无退避下直接终态化。 */ @Singleton @@ -16,23 +16,25 @@ class ProcFailure( private val procState: ProcStateRepository, val scheduler: FailureScheduler, ) { - fun fail(head: ProcState, ec: ErrorClass, reason: String) { + /** @return 是否已达终态（DEAD 才是终态；FAILED 仍可重放/重试） */ + fun fail(head: ProcState, ec: ErrorClass, reason: String): Boolean { val attempts = head.attempts + 1 if (scheduler.exhausted(attempts)) { - procState.update( + procState.markTerminal( head.msgId, ProcStatus.DEAD, attempts = attempts, errorClass = ErrorClass.EXHAUSTED, lastError = "$reason; attempts=$attempts", ) - } else { - procState.update( - head.msgId, ProcStatus.FAILED, - attempts = attempts, - nextAttemptAt = scheduler.nextAttemptAt(attempts), - errorClass = ec, - lastError = reason, - ) + return true } + procState.update( + head.msgId, ProcStatus.FAILED, + attempts = attempts, + nextAttemptAt = scheduler.nextAttemptAt(attempts), + errorClass = ec, + lastError = reason, + ) + return false } } 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 eed10d5..ec46669 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 @@ -11,8 +11,11 @@ import com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot import com.gzzn.omms.msgexchange.domain.flight.FlightState import com.gzzn.omms.msgexchange.domain.flight.HistoryCandidate import com.gzzn.omms.msgexchange.domain.flight.HistoryRules -import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository +import com.gzzn.omms.msgexchange.infra.persistence.BackfillDue +import com.gzzn.omms.msgexchange.infra.persistence.Backlog import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository +import com.gzzn.omms.msgexchange.infra.persistence.InboxCursorRepository +import com.gzzn.omms.msgexchange.infra.persistence.MailboxRow import com.gzzn.omms.msgexchange.infra.persistence.FlightStateRepository import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository import com.gzzn.omms.msgexchange.infra.persistence.PersistOutcome @@ -55,12 +58,12 @@ class StubProcState : ProcStateRepository { rows.clear(); bound.clear() } - override fun insert(msgId: Long, state: ProcStatus) { - rows.getOrPut(msgId) { ProcState(msgId, state) } + override fun insertIfAbsent(msgId: Long, receivedAt: Instant?): Boolean { + if (rows.containsKey(msgId)) return false + rows[msgId] = ProcState(msgId, ProcStatus.PENDING, receivedAt = receivedAt) + return true } - override fun exists(msgId: Long): Boolean = rows.containsKey(msgId) - override fun find(msgId: Long): ProcState? = rows[msgId] override fun findSuccessTerminal(msgId: Long): Boolean = rows[msgId]?.state == ProcStatus.SUCCEEDED @@ -97,6 +100,67 @@ class StubProcState : ProcStateRepository { ) } + /** 终态 + 回填意图同一动作（模拟单条 UPDATE 的原子性）。 */ + override fun markTerminal( + msgId: Long, + state: ProcStatus, + errorClass: ErrorClass?, + lastError: String?, + attempts: Int?, + now: Instant, + ) { + val old = rows[msgId] ?: ProcState(msgId, state) + rows[msgId] = old.copy( + state = state, + errorClass = errorClass, + lastError = lastError, + attempts = attempts ?: old.attempts, + nextAttemptAt = null, + backfillAt = null, + backfillNextAt = now, + backfillAttempts = 0, + backfillError = null, + updatedAt = now, + ) + } + + override fun markBackfilled(msgId: Long, now: Instant) { + rows[msgId]?.let { + rows[msgId] = it.copy(backfillAt = now, backfillNextAt = null, backfillError = null, updatedAt = now) + } + } + + override fun recordBackfillFailure(msgId: Long, error: String?, attempts: Int, nextAttemptAt: Instant, now: Instant) { + rows[msgId]?.let { + rows[msgId] = it.copy( + backfillAttempts = attempts, + backfillNextAt = nextAttemptAt, + backfillError = error, + updatedAt = now, + ) + } + } + + override fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List = + rows.values + .filter { it.state.isTerminal() && it.backfillAt == null } + .filter { + it.backfillNextAt == null || it.backfillNextAt <= now || + (it.receivedAt != null && it.receivedAt < overdueBefore) + } + .sortedBy { it.msgId } + .take(limit) + .map { BackfillDue(it.msgId, it.backfillAttempts) } + + override fun backlog(): Backlog { + val unfinished = rows.values.filter { it.state == ProcStatus.PENDING || it.state == ProcStatus.FAILED } + return Backlog( + unfinished = unfinished.size, + oldestReceivedAt = unfinished.mapNotNull { it.receivedAt }.minOrNull(), + unmarkedTerminal = rows.values.count { it.state.isTerminal() && it.backfillAt == null }, + ) + } + override fun requeueByErrorClasses(errorClasses: List): Int { var n = 0 rows.forEach { (id, s) -> @@ -300,60 +364,69 @@ class StubReqTrack : ReqTrackRepository { } } -@Singleton -@Requires(property = "msgx.stubs", value = "true") -class StubBackfillTodo : BackfillTodoRepository { - val tasks = linkedMapOf() - private val errors = linkedMapOf() - private val due = linkedMapOf() - - fun clear() { tasks.clear(); errors.clear(); due.clear() } - - fun lastErrorOf(msgId: Long): String? = errors[msgId] - - override fun record(task: BackfillTodoRepository.BackfillTask, lastError: String?, now: Instant) { - errors[task.msgId] = lastError - due.putIfAbsent(task.msgId, now) - tasks[task.msgId] = task.copy(attempts = tasks[task.msgId]?.attempts ?: 0) - } - - override fun findDue(now: Instant, limit: Int): List = - tasks.keys.filter { (due[it] ?: Instant.EPOCH) <= now }.take(limit).mapNotNull { tasks[it] } - - override fun markFailed(msgId: Long, lastError: String?, nextAttemptAt: Instant, now: Instant) { - errors[msgId] = lastError - due[msgId] = nextAttemptAt - tasks[msgId]?.let { tasks[msgId] = it.copy(attempts = it.attempts + 1) } - } - - override fun delete(msgId: Long) { - tasks.remove(msgId); errors.remove(msgId); due.remove(msgId) - } - - override fun count(): Int = tasks.size -} - @Singleton @Requires(property = "msgx.stubs", value = "true") class StubInbox : CminmsgInboxRepository { val raws = linkedMapOf() + private val received = linkedMapOf() + private val marks = linkedMapOf() private val ids = AtomicLong(0) - fun clear() = raws.clear() + fun clear() { + raws.clear(); received.clear(); marks.clear(); ids.set(0) // ID 自 1 重新分配，用例间不串扰 + } override fun insertRaw(rawXml: String): Long { val id = ids.incrementAndGet() raws[id] = rawXml + received[id] = Instant.now() return id } override fun rawOf(msgId: Long): String? = raws[msgId] - override fun pollUnprocessed(afterId: Long, limit: Int): List = - raws.keys.filter { it > afterId }.sorted().take(limit) + override fun readRange(fromExclusive: Long, limit: Int): List = + raws.keys.filter { it > fromExclusive }.sorted().take(limit).map { MailboxRow(it, received[it]) } - override fun backfillOnSuccess(msgId: Long, sndr: String, type: String, styp: String, seqn: Long) = Unit + override fun maxId(): Long? = raws.keys.maxOrNull() + + /** §11 单调：只把空标写为已处理；已有值不回撤、不覆盖。 */ + override fun markProcessedIfUnmarked(msgId: Long, value: String): Boolean { + if (!raws.containsKey(msgId) || marks.containsKey(msgId)) return false + marks[msgId] = value + return true + } + + fun markOf(msgId: Long): String? = marks[msgId] + + fun isMarked(msgId: Long): Boolean = marks.containsKey(msgId) /** 测试辅助：模拟上游外部写入共享信箱（不经本系统）。 */ fun simulateExternalWrite(rawXml: String): Long = insertRaw(rawXml) + + /** 测试辅助：模拟库方清除（原文不可读），用于 §9 原文缺失与空洞场景。 */ + fun removeRow(msgId: Long) { + raws.remove(msgId); received.remove(msgId); marks.remove(msgId) + } } + +/** 消费水位游标（stub）。 */ +@Singleton +@Requires(property = "msgx.stubs", value = "true") +class StubInboxCursor : InboxCursorRepository { + var cursor: InboxCursorRepository.Cursor = InboxCursorRepository.Cursor() + + fun clear() { + cursor = InboxCursorRepository.Cursor() + } + + override fun load(): InboxCursorRepository.Cursor = cursor + + override fun save(cursor: InboxCursorRepository.Cursor) { + this.cursor = cursor + } +} + +/** 终态判定（§2 状态总纲）。 */ +private fun ProcStatus.isTerminal(): Boolean = + this == ProcStatus.SUCCEEDED || this == ProcStatus.SKIPPED || this == ProcStatus.DEAD diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxEnqueue.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxEnqueue.kt deleted file mode 100644 index 7bcd81f..0000000 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxEnqueue.kt +++ /dev/null @@ -1,16 +0,0 @@ -package com.gzzn.omms.msgexchange.ingress - -import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository -import jakarta.inject.Singleton - -/** 流程 1 入队：共享库 CMINMSGS_ID → 自有 PG PROC_STATE(PENDING)，幂等跳过已存在行。 */ -@Singleton -class InboxEnqueue(private val procState: ProcStateRepository) { - - /** @return true 若新建 PENDING 行；false 若已存在（轮询重扫/compensate 幂等）。 */ - fun enqueue(msgId: Long): Boolean { - if (procState.exists(msgId)) return false - procState.insert(msgId) - return true - } -} 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 07185d4..a41b9b8 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt @@ -2,18 +2,34 @@ 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 com.gzzn.omms.msgexchange.infra.persistence.InboxCursorRepository +import com.gzzn.omms.msgexchange.infra.persistence.MailboxRow +import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager +import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository import jakarta.inject.Singleton +import java.time.Duration +import java.time.Instant /** - * ACMA-8 流程 1 · 主路径：JDBC 轮询共享 MySQL CMINMSGS（`DATE_PROCESSED IS NULL`）， - * 发现新信后入队自有 PG。与 legacy `MsgExchangeRunner.getNewMsgsAfterId(0L)` 同语义。 + * 收报（docs/message-lifecycle.md §5.1）：共享 MySQL 信箱按 **ID 区间**升序有界读取， + * 在自有 PG 建 `PROC_STATE(PENDING)`，并把消费水位推进到连续上界；入队与水位推进在同一 + * PG 事务内提交，中断即重扫补建（§4 第一行）。收报层不解析业务载荷、不写处理标记。 + * + * 三条红线： + * - **扫描谓词不含处理标记**：标记只用于回填与库方清除，不参与消息发现；否则已入队而未 + * 回填的行会永久占据批次，死信累积到批大小时收报整体停摆； + * - **水位遇空洞即停**：不越过空洞入队（越过后较小 ID 迟到即 FIFO 越序，architecture §5）； + * - **空洞老化**：超过最大提交时延（Q2 承诺）的空洞判定为永久并放行，否则水位会永久停摆于 + * 一次自增回滚留下的空位。 + * + * 空转代价为每轮一次区间 SELECT。 */ @Singleton class InboxPoller( - private val inbox: CminmsgInboxRepository, - private val enqueue: InboxEnqueue, - private val backfillSweep: BackfillSweepJob, + private val mailbox: CminmsgInboxRepository, + private val procState: ProcStateRepository, + private val cursor: InboxCursorRepository, + private val txManager: PipelineTransactionManager, private val props: PipelineProps, ) { private val log = org.slf4j.LoggerFactory.getLogger(InboxPoller::class.java) @@ -21,31 +37,40 @@ class InboxPoller( @Volatile private var running = false - /** legacy 现役：afterId=0，每轮扫全部未处理行；PROC_STATE 判重防重复入队。 */ - fun pollOnce(): Int { + /** @return 本轮新建的入队条数 */ + fun pollOnce(now: Instant = Instant.now()): Int { val batch = props.pipeline.claimBatch.coerceAtLeast(1) - val ids = inbox.pollUnprocessed(afterId = 0L, limit = batch) - var enqueued = 0 - for (id in ids) { - if (enqueue.enqueue(id)) { - enqueued++ - log.info("polled msgId={}", id) - } - } - sweepBackfillTodos() - return enqueued - } + val watermark = cursor.load() + val rows = mailbox.readRange(watermark.committedUpTo, batch) + if (rows.isEmpty()) return 0 - /** 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) + // 连续上界；读取区间内出现空洞时，只推进到连续部分，空洞之后的行暂不入队（防较小 ID 迟到被越过） + val contiguous = contiguousUpTo(watermark.committedUpTo, rows) ?: watermark.committedUpTo + var committedTo = contiguous + var holeSince: Instant? = null + if (rows.last().msgId > contiguous) { + val since = watermark.holeSince ?: now + if (Duration.between(since, now) < props.pipeline.maxCommitDelay) { + holeSince = since + } else { + // 空洞老化：超过最大提交时延仍缺席即判永久（Q2），放行水位，否则永久停摆 + committedTo = rows.first { it.msgId > contiguous }.msgId - 1 + log.warn("hole after W={} aged out, watermark advanced to {}", contiguous, committedTo) } - } catch (e: Exception) { - log.error("backfill sweep tick failed", e) } + + val enqueued = txManager.inTransaction { + var n = 0 + rows.takeWhile { it.msgId <= committedTo }.forEach { row -> + if (procState.insertIfAbsent(row.msgId, row.receivedAt)) n++ + } + cursor.save(InboxCursorRepository.Cursor(committedTo, holeSince)) + n + } + if (enqueued > 0) { + log.info("polled {} new messages, W {} -> {}", enqueued, watermark.committedUpTo, committedTo) + } + return enqueued } fun loop() { @@ -70,7 +95,19 @@ class InboxPoller( running = false } - private fun sleepQuietly(d: java.time.Duration) { + /** 连续上界：从 W+1 起 ID 逐 1 相邻的最后一个；首个空位即停（rows 为升序且覆盖该区间）。 */ + private fun contiguousUpTo(from: Long, rows: List): Long? { + var expected = from + 1 + var last: Long? = null + for (row in rows) { + if (row.msgId != expected) break + last = row.msgId + expected++ + } + return last + } + + private fun sleepQuietly(d: Duration) { try { Thread.sleep(d.toMillis().coerceAtLeast(1)) } catch (_: InterruptedException) { diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxService.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxService.kt index af31d13..c033e95 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxService.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxService.kt @@ -1,23 +1,27 @@ package com.gzzn.omms.msgexchange.ingress import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository +import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository import jakarta.inject.Singleton import java.time.Instant -/** ACMA-8 流程 1 · compat 写路径：HTTP 落信 + PG 入队（非生产主拓扑；主路径=InboxPoller JDBC 轮询）。 */ +/** + * ACMA-8 流程 1 · compat 写路径：HTTP 落信 + PG 入队（非生产主拓扑；主路径=InboxPoller 轮询）。 + * 两步不在同一事务：信箱成功而 PG 失败时原文不丢失，由轮询按 ID 区间补建（design.md §3.1）。 + */ @Singleton class InboxService( private val inbox: CminmsgInboxRepository, - private val enqueue: InboxEnqueue, + private val procState: ProcStateRepository, ) { private val log = org.slf4j.LoggerFactory.getLogger(InboxService::class.java) data class Receipt(val msgId: Long, val receivedAt: Instant) - fun accept(rawXml: String): Receipt { + fun accept(rawXml: String, now: Instant = Instant.now()): Receipt { val id = inbox.insertRaw(rawXml) - enqueue.enqueue(id) + procState.insertIfAbsent(id, now) log.info("compat-accepted msgId={}", id) - return Receipt(id, Instant.now()) + return Receipt(id, now) } } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/BackfillSweepJob.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/BackfillSweepJob.kt deleted file mode 100644 index fed4059..0000000 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/BackfillSweepJob.kt +++ /dev/null @@ -1,53 +0,0 @@ -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.msgId, task.sndr, task.type, task.styp, task.seqn) - todo.delete(task.msgId) - succeeded++ - } catch (e: Exception) { - todo.markFailed(task.msgId, 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/JobRunner.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt index 34fdc8c..da71eef 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt @@ -2,6 +2,7 @@ package com.gzzn.omms.msgexchange.jobs import com.gzzn.omms.msgexchange.config.HistoryProps import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.processing.BackfillService import jakarta.inject.Singleton import java.time.Duration import java.time.Instant @@ -11,11 +12,12 @@ import java.time.ZoneId /** * 维护作业调度（docs/design.md §6.1/§6.2）：单 daemon 线程，独立于主泵—— * 作业不参与消息 FIFO，也不使到期消息饥饿（PUMP_JOB 队列机制已随审计口径移除）。 - * 触发：回填补偿扫描 30s 固定间隔；历史归档/留痕清理每日机场时区 03:30 后首个 tick。 + * 触发：回填补写扫描 30s 固定间隔；历史归档/留痕清理每日机场时区 03:30 后首个 tick。 + * 回填意图由处理器在终态事务内登记（message-lifecycle §4），本线程只负责到期重试。 */ @Singleton class JobRunner( - private val backfillSweep: BackfillSweepJob, + private val backfill: BackfillService, private val historySweep: HistorySweepJob, @Suppress("unused") private val pipelineProps: PipelineProps, @Suppress("unused") private val historyProps: HistoryProps, @@ -47,7 +49,7 @@ class JobRunner( internal fun loop() { while (running) { try { - backfillSweep.sweep() + backfill.sweep() maybeHistorySweep() } catch (e: InterruptedException) { Thread.currentThread().interrupt() diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/BackfillService.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/BackfillService.kt new file mode 100644 index 0000000..5f4587c --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/BackfillService.kt @@ -0,0 +1,74 @@ +package com.gzzn.omms.msgexchange.processing + +import com.gzzn.omms.msgexchange.config.MailboxProps +import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository +import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository +import jakarta.inject.Singleton +import java.time.Clock +import java.time.Duration +import java.time.Instant + +/** + * 信箱处理标记回填（docs/message-lifecycle.md §3/§4/§5.2）。 + * + * 回填意图（`BACKFILL_NEXT_AT`）由处理器在终态事务内登记，与业务写入同提交同回滚， + * 因此不存在"业务已提交、待办未记"的窗口；本服务只做两件事： + * 1. [attempt]：终态提交后立即尝试一次（低延迟，失败静默留给扫描）； + * 2. [sweep]：到期或已达超期期限 R 的记录批量补写（§5.2），指数退避 30s 起步、封顶 15 分钟。 + * + * 回填失败绝不重放业务变更，也绝不回改终态（§11）；写入侧只把空标写为已处理， + * 重复执行无副作用。 + */ +@Singleton +class BackfillService( + private val procState: ProcStateRepository, + private val mailbox: CminmsgInboxRepository, + private val mailboxProps: MailboxProps, + private val props: PipelineProps, + private val clock: Clock, +) { + private val log = org.slf4j.LoggerFactory.getLogger(BackfillService::class.java) + + 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 attempt(msgId: Long, now: Instant = clock.instant()) { + record(msgId, attempts = 0, now = now)?.let { + log.warn("backfill failed msgId={} error={} (sweep will retry)", msgId, it) + } + } + + /** + * 批量补写（JobRunner 每 30s 触发；重启即继续，不依赖内存状态）。 + * @return 本批检查条数 + */ + fun sweep(now: Instant = clock.instant()): Int { + val due = procState.findBackfillDue(now, now.minus(props.pipeline.overdueBackfill), props.pipeline.backfillBatch) + due.forEach { record(it.msgId, it.attempts, now) } + return due.size + } + + /** @return 失败原因；null = 已确认标记（含"已被其他路径标记"的幂等成功） */ + private fun record(msgId: Long, attempts: Int, now: Instant): String? = + try { + // 影响 0 行 = 已有标记；按幂等成功处理（§11 标记单调：不回撤、不覆盖） + mailbox.markProcessedIfUnmarked(msgId, mailboxProps.processedValue) + procState.markBackfilled(msgId, now) + null + } catch (e: Exception) { + val reason = e.message ?: e.javaClass.simpleName + runCatching { + procState.recordBackfillFailure(msgId, reason, attempts + 1, now.plus(backoffDelayFor(attempts + 1)), now) + }.onFailure { log.error("record backfill failure failed msgId={}", msgId, it) } + reason + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/DynamicProcessors.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/DynamicProcessors.kt index dc79a22..768d2fa 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/DynamicProcessors.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/DynamicProcessors.kt @@ -8,17 +8,18 @@ import com.gzzn.omms.msgexchange.domain.EventType import com.gzzn.omms.msgexchange.domain.MsgEvent import com.gzzn.omms.msgexchange.domain.OperationDayCalculator 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.domain.flight.FlightSnapshot import com.gzzn.omms.msgexchange.domain.flight.FlightState import com.gzzn.omms.msgexchange.domain.flight.FlightStateEngine import com.gzzn.omms.msgexchange.domain.flight.MergeChange import com.gzzn.omms.msgexchange.domain.flight.ScheduleRecord -import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository import com.gzzn.omms.msgexchange.infra.persistence.FlightStateRepository import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository import com.gzzn.omms.msgexchange.infra.persistence.PipelineLockRepository import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager +import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository import jakarta.inject.Singleton import java.time.Instant import java.time.ZoneId @@ -34,19 +35,23 @@ class FlopProcessor( private val lock: PipelineLockRepository, private val flightState: FlightStateRepository, private val msgEvents: MsgEventRepository, - private val backfillTodo: BackfillTodoRepository?, + private val procState: ProcStateRepository, private val mapper: ObjectMapper, ) { fun apply(head: ProcState, msg: DecodedMessage, payload: FlopPayload): ApplyResult = txManager.inTransaction { lock.lock() val current = flightState.loadFullSnapshot(payload.flid) - ?: return@inTransaction idempotentAbsent(head, msg) // 迟到/未知航班：幂等成功，不创建 + if (current == null) { + // 迟到/未知航班：幂等成功，不创建（创建入口只有 SCHD/ADFT）；终态同事务落库 + procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED) + return@inTransaction ApplyResult.Succeeded + } val change = MergeChange(flid = payload.flid, scalars = payload.scalars, collections = payload.collections) val next = FlightStateEngine.mergedState(current, change) flightState.persistFullState(next, msgId = head.msgId, now = Instant.now()) msgEvents.insertAll(eventsFor(next, mapper)) - preRegisterBackfill(head, msg, backfillTodo) + procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED) ApplyResult.Succeeded } } @@ -61,7 +66,7 @@ class FdelProcessor( private val lock: PipelineLockRepository, private val flightState: FlightStateRepository, private val msgEvents: MsgEventRepository, - private val backfillTodo: BackfillTodoRepository?, + private val procState: ProcStateRepository, private val mapper: ObjectMapper, ) { fun apply(head: ProcState, msg: DecodedMessage, payload: FlopPayload): ApplyResult = txManager.inTransaction { @@ -95,9 +100,9 @@ class FdelProcessor( ), ), ) - preRegisterBackfill(head, msg, backfillTodo) } - ApplyResult.Succeeded // 未命中 = 迟到/重复，幂等成功（§3.3） + procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED) // 未命中 = 迟到/重复，幂等成功（§3.3） + ApplyResult.Succeeded } } @@ -112,7 +117,7 @@ class AdftProcessor( private val lock: PipelineLockRepository, private val flightState: FlightStateRepository, private val msgEvents: MsgEventRepository, - private val backfillTodo: BackfillTodoRepository?, + private val procState: ProcStateRepository, operationDayProps: OperationDayProps, private val mapper: ObjectMapper, ) { @@ -135,7 +140,7 @@ class AdftProcessor( msgEvents.insertAll(eventsFor(next, mapper)) } } - preRegisterBackfill(head, msg, backfillTodo) + procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED) return@inTransaction ApplyResult.Succeeded } @@ -161,7 +166,7 @@ class AdftProcessor( "operation-day guard violated flid=${record.flid}" } msgEvents.insertAll(eventsFor(next, mapper)) - preRegisterBackfill(head, msg, backfillTodo) + procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED) ApplyResult.Succeeded } @@ -177,9 +182,6 @@ class AdftProcessor( // 共享小工具（处理器层私有约定） // ===================================================================== -/** 航班不存在/迟到：幂等成功（§3.2/§3.3；不阻塞队头，不创建实例——创建入口只有 SCHD/ADFT）。 */ -private fun idempotentAbsent(head: ProcState, msg: DecodedMessage): ApplyResult = ApplyResult.Succeeded - /** KAFKA_SCHD 整态 + KAFKA_MSG 变化通知（flight-state.md §3.1/§5）。 */ internal fun eventsFor(next: FlightSnapshot, mapper: ObjectMapper): List { val payload = linkedMapOf( @@ -203,14 +205,3 @@ internal fun eventsFor(next: FlightSnapshot, mapper: ObjectMapper): List sleepQuietly(props.pipeline.pollInterval) - head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > Instant.now() -> + head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > clock.instant() -> if (poisoned(head)) { log.error("poison -> DEAD msgId={} attempts={} lastError={}", head.msgId, head.attempts, head.lastError) - procState.update( + procState.markTerminal( head.msgId, ProcStatus.DEAD, - errorClass = ErrorClass.EXHAUSTED, lastError = head.lastError ?: "head-deadline-exceeded", + errorClass = ErrorClass.EXHAUSTED, + lastError = head.lastError ?: "head-deadline-exceeded", + attempts = head.attempts, ) + backfill.attempt(head.msgId) } else { - sleepQuietly(Duration.between(Instant.now(), head.nextAttemptAt)) + sleepQuietly(Duration.between(clock.instant(), head.nextAttemptAt)) } // PENDING、或 FAILED 退避已到期：交处理入口（内部有边界化失败迁移与 attempts 守卫） else -> processor.processOne(head) @@ -75,7 +81,7 @@ class Pump( private fun poisoned(head: ProcState): Boolean = head.attempts >= props.pipeline.maxAttempts || - Duration.between(head.updatedAt, Instant.now()) > props.pipeline.headDeadline + Duration.between(head.updatedAt, clock.instant()) > props.pipeline.headDeadline private fun sleepQuietly(d: Duration) { if (!d.isNegative && !d.isZero) Thread.sleep(d.toMillis().coerceAtLeast(1)) @@ -83,7 +89,7 @@ class Pump( } /** - * processOne：解码 → 绑定 → 处理器（事务内决策+落库）→ 终态迁移 → 回填。 + * processOne：解码 → 绑定 → 处理器（事务内决策 + 落库 + 终态 + 回填意图）→ 提交后最佳努力回填。 * 边界化失败迁移（ProcFailure）：任何意外异常归于本条 head，FAILED(INFRA)+退避，不穿出杀泵； * MALFORMED / PROTOCOL 直接 DEAD 不重试（docs/design.md §2.3 错误分类）。 */ @@ -97,14 +103,14 @@ class MessageProcessor( private val fdelProcessor: FdelProcessor, private val adftProcessor: AdftProcessor, private val procFailure: ProcFailure, + private val backfill: BackfillService, private val props: PipelineProps, - private val backfillTodo: BackfillTodoRepository? = null, ) { private val log = org.slf4j.LoggerFactory.getLogger(MessageProcessor::class.java) fun processOne(head: ProcState) { TraceLog.withTrace(head.msgId) { - try { + val terminal = try { processInternal(head) } catch (e: InterruptedException) { Thread.currentThread().interrupt() @@ -114,21 +120,30 @@ class MessageProcessor( log.warn("processOne unexpected failure msgId={} ec=INFRA msg={}", head.msgId, e.message ?: e.javaClass.simpleName) procFailure.fail(head, ErrorClass.INFRA, e.message ?: e.javaClass.simpleName) } + // 终态已提交：立即尝试一次回填；失败留待回填扫描按退避重试（意图已在终态事务内登记）。 + // 中间态（PENDING/FAILED）不适用（message-lifecycle §5.2）。 + if (terminal) backfill.attempt(head.msgId) } } - private fun processInternal(head: ProcState) { + /** @return 是否已达终态（终态才允许回填信箱标记） */ + private fun processInternal(head: ProcState): Boolean { // 守卫：手工/遗留 FAILED 行若 attempts 已达上限，直接终态（防止退避到期后无限重试） if (head.state == ProcStatus.FAILED && procFailure.scheduler.exhausted(head.attempts)) { log.error("head exhausted at entry -> DEAD msgId={} attempts={}", head.msgId, head.attempts) - procState.update(head.msgId, ProcStatus.DEAD, errorClass = ErrorClass.EXHAUSTED, lastError = head.lastError ?: "max-attempts") - return + procState.markTerminal( + head.msgId, ProcStatus.DEAD, + errorClass = ErrorClass.EXHAUSTED, + lastError = head.lastError ?: "max-attempts", + attempts = head.attempts, + ) + return true } - val raw = inbox.rawOf(head.msgId) ?: run { + val raw = inbox.rawOf(head.msgId) + if (raw == null) { log.error("raw missing -> DEAD(MALFORMED) msgId={}", head.msgId) - procState.update(head.msgId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = "raw-missing") - return + return deadMalformed(head, "raw-missing") } val decoded = when (val r = codec.decode(raw)) { is com.gzzn.omms.msgexchange.codec.DecodeResult.Ok -> r.message @@ -136,12 +151,10 @@ class MessageProcessor( // MALFORMED（报文非法）→ DEAD 不重试；CODEC_ERROR（可随 codec 修复重放）→ FAILED 退避 if (r.failure.errorClass == ErrorClass.MALFORMED) { log.error("decode MALFORMED -> DEAD msgId={} detail={}", head.msgId, r.failure.detail) - procState.update(head.msgId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = r.failure.detail) - } else { - log.warn("decode {} -> FAILED msgId={} detail={}", r.failure.errorClass, head.msgId, r.failure.detail) - procFailure.fail(head, r.failure.errorClass, r.failure.detail) + return deadMalformed(head, r.failure.detail) } - return + log.warn("decode {} -> FAILED msgId={} detail={}", r.failure.errorClass, head.msgId, r.failure.detail) + return procFailure.fail(head, r.failure.errorClass, r.failure.detail) } } @@ -151,9 +164,8 @@ class MessageProcessor( if (!procState.tryBindIdentity(head.msgId, identity)) { val owner = procState.ownerOfIdentity(identity) ?: -1L log.info("duplicate-of:{} -> SKIPPED msgId={}", owner, head.msgId) - procState.update(head.msgId, ProcStatus.SKIPPED, lastError = "duplicate-of:$owner") - compensateBackfill(head, decoded) - return + procState.markTerminal(head.msgId, ProcStatus.SKIPPED, lastError = "duplicate-of:$owner") + return true } } @@ -161,93 +173,52 @@ class MessageProcessor( val result: ApplyResult = when (val kind = decoded.kind) { is MsgKind.Schd -> { val body = decoded.body as? ScheduleBody - if (body == null) { - deadMalformed(head, "missing-schd-body") - return - } + if (body == null) return deadMalformed(head, "missing-schd-body") when (kind.subtype) { MsgKind.SchdSubtype.DNLD, MsgKind.SchdSubtype.RESP -> scheduleProcessor.applyScheduleRecords(head, decoded) MsgKind.SchdSubtype.ADFT -> { val record = body.records.singleOrNull() - if (record == null) { - deadMalformed(head, "adft-needs-single-fltr") - return - } + if (record == null) return deadMalformed(head, "adft-needs-single-fltr") adftProcessor.apply(head, decoded, record) } } } MsgKind.Fdel -> { val payload = decoded.body as? FlopPayload - if (payload == null) { - deadMalformed(head, "missing-fdel-flid") - return - } + if (payload == null) return deadMalformed(head, "missing-fdel-flid") fdelProcessor.apply(head, decoded, payload) } is MsgKind.Flop -> { val payload = decoded.body as? FlopPayload - if (payload == null) { - deadMalformed(head, "missing-flop-body") - return - } + if (payload == null) return deadMalformed(head, "missing-flop-body") flopProcessor.apply(head, decoded, payload) } is MsgKind.Unsupported -> { // design.md §2.3：未支持类型 → FAILED(UNSUPPORTED) 退避重试，达阈值转 DEAD；绝不写终态 log.warn("unsupported type -> FAILED(UNSUPPORTED) msgId={} tag={}", head.msgId, kind.tag) - procFailure.fail(head, ErrorClass.UNSUPPORTED, "no-handler:${kind.tag}") - return + return procFailure.fail(head, ErrorClass.UNSUPPORTED, "no-handler:${kind.tag}") } } - when (result) { - is ApplyResult.Succeeded, ApplyResult.ReplaySkipped -> - procState.update(head.msgId, ProcStatus.SUCCEEDED) - is ApplyResult.DeadProtocol -> { - // design.md §2.3：整包拒绝 DEAD(PROTOCOL)，立即释放队头，交人工确认 - log.error("DEAD(PROTOCOL) msgId={} reason={} flags={}", head.msgId, result.reason, result.flags) - procState.update(head.msgId, ProcStatus.DEAD, errorClass = ErrorClass.PROTOCOL, lastError = result.reason.take(1000)) - compensateBackfill(head, decoded) // 拒绝包同样要回填信箱，防止反复轮询 - return - } + if (result is ApplyResult.DeadProtocol) { + // design.md §2.3：整包拒绝 DEAD(PROTOCOL)，立即释放队头，交人工确认 + log.error("DEAD(PROTOCOL) msgId={} reason={} flags={}", head.msgId, result.reason, result.flags) + procState.markTerminal( + head.msgId, ProcStatus.DEAD, + errorClass = ErrorClass.PROTOCOL, + lastError = result.reason.take(1000), + ) + return true } - backfill(head, decoded) + // Succeeded / ReplaySkipped：SUCCEEDED 终态与回填意图已由处理器在自己的事务内落库 log.info("SUCCEEDED msgId={} kind={}", head.msgId, decoded.typeTag) + return true } - /** design.md §3.3/§6.1：提交后回填共享信箱；失败不得把 SUCCEEDED 改回 FAILED，待办已事务内预登记。 */ - private fun backfill(head: ProcState, decoded: DecodedMessage) { - try { - inbox.backfillOnSuccess(head.msgId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn) - backfillTodo?.delete(head.msgId) - } catch (e: Exception) { - log.error("backfill failed after SUCCEEDED msgId={} (todo pre-registered, sweep will retry)", head.msgId, e) - } - } - - /** SKIPPED/DEAD(PROTOCOL) 包的回填（无预登记待办）：失败落补偿待办。 */ - private fun compensateBackfill(head: ProcState, decoded: DecodedMessage) { - try { - inbox.backfillOnSuccess(head.msgId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn) - } catch (e: Exception) { - log.error("backfill failed msgId={} (compensation required)", head.msgId, e) - backfillTodo?.record( - BackfillTodoRepository.BackfillTask( - msgId = head.msgId, - 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 msgId={}", head.msgId) - } - } - - private fun deadMalformed(head: ProcState, detail: String) { - procState.update(head.msgId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = detail) + private fun deadMalformed(head: ProcState, detail: String): Boolean { + procState.markTerminal(head.msgId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = detail) + return true } } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessor.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessor.kt index d4e6ac6..89ae6fa 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessor.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessor.kt @@ -6,6 +6,7 @@ import com.gzzn.omms.msgexchange.config.OperationDayProps import com.gzzn.omms.msgexchange.domain.DecodedMessage 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.SnapshotFlag import com.gzzn.omms.msgexchange.domain.SnapshotLogEntry import com.gzzn.omms.msgexchange.domain.SnapshotResult @@ -16,7 +17,6 @@ import com.gzzn.omms.msgexchange.domain.flight.FlightState import com.gzzn.omms.msgexchange.domain.flight.FlightStateEngine import com.gzzn.omms.msgexchange.domain.flight.ScheduleRecord import com.gzzn.omms.msgexchange.domain.flight.SnapshotValidation -import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository import com.gzzn.omms.msgexchange.infra.persistence.FlightStateRepository import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository import com.gzzn.omms.msgexchange.infra.persistence.PersistOutcome @@ -29,7 +29,7 @@ import java.time.Instant import java.time.LocalDate import java.time.ZoneId -/** 处理器执行结果——终态迁移由 MessageProcessor 统一落库（docs/design.md §2.3）。 */ +/** 处理器执行结果——终态与回填意图由处理器在自己的业务事务内落库（message-lifecycle §2/§4）。 */ sealed interface ApplyResult { /** 业务成功（含幂等成功）。 */ data object Succeeded : ApplyResult @@ -57,7 +57,6 @@ class ScheduleProcessor( private val flightState: FlightStateRepository, private val msgEvents: MsgEventRepository, private val snapshotLog: SnapshotLogRepository, - private val backfillTodo: BackfillTodoRepository?, operationDayProps: OperationDayProps, private val mapper: ObjectMapper, ) { @@ -133,14 +132,8 @@ class ScheduleProcessor( events += snapshotEvents(next) } if (events.isNotEmpty()) msgEvents.insertAll(events) - // 回填待办与业务终态同事务预登记（design.md §3.3/§6.1；提交后由 MessageProcessor 回填并删待办） - backfillTodo?.record( - BackfillTodoRepository.BackfillTask( - msgId = head.msgId, sndr = msg.meta.sndr, type = msg.meta.type, - styp = msg.meta.styp, seqn = msg.meta.seqn, - ), - null, - ) + // 终态与回填意图同一事务（message-lifecycle §2/§4）：业务写入、事件、终态、回填意图同提交同回滚 + procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED) written } logSnapshot(head, body, SnapshotResult.COMMITTED, upserted, flags, started) diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 02a42bb..5f5a378 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -19,6 +19,9 @@ msgx: backoff-ms: [1000, 2000, 4000, 8000, 16000] # 指数退避，单次封顶 60s backoff-cap-ms: 60000 head-deadline: 10m # 队头滞留上界 = 最坏 HOL 时长（毒丸升级） + max-commit-delay: 5m # §5.1 空洞老化：W+1 空洞超过该时延判定为永久（Q2 最大提交时延） + overdue-backfill: 30d # §5.2 超期补写期限 R：≥ 人工重放期限 + 人工处置期限（Q6） + backfill-batch: 100 # 回填扫描单批条数 autostart: false # U07：启动即拉起 Pump/Dispatcher 循环；需真实仓储或 msgx.stubs=true 才开启（dev 见 application-dev.yml） schd: flush-period: 3s # KEEP 现役推送节律 @@ -41,8 +44,9 @@ micronaut: port: 8080 # 管理端点（U03/N32）：/env、/beans 默认 sensitive；仅开发/影子环境放开——见 application-dev.yml -# 自有 PostgreSQL：全部内部状态（PIPELINE_LOCK/PROC_STATE/MSG_EVENT/REQ_TRACK/BACKFILL_TODO/ -# FLIGHT_SCHD 及明细表/SCHD_SNAP_LOG；迁移 db/migration/V1__flight_state_baseline.sql）。 +# 自有 PostgreSQL：全部内部状态（PIPELINE_LOCK/PROC_STATE/MSG_EVENT/REQ_TRACK/INBOX_CURSOR/ +# FLIGHT_SCHD 及明细表/SCHD_SNAP_LOG；迁移 db/migration/V1__flight_state_baseline.sql + +# V2__inbox_lifecycle.sql——V2 起回填意图并入 PROC_STATE，BACKFILL_TODO 已下线）。 # 数据层实装前 enabled=false（stub 模式不建连）。 datasources: default: @@ -62,6 +66,7 @@ flyway: # 处理回填；出站写 COUTMSGS；不建表/schema，ACM2-12）。驱动/依赖与信箱适配层 # （CminmsgMailbox/OutboxMailbox）随 U05 批次引入。 mailbox: + processed-value: PROCESSED # §5.2 处理标记写入值（Q7：值集与写权限以库方契约为准） shared-mysql: enabled: false url: ${MSGX_MAILBOX_URL} diff --git a/src/main/resources/db/migration/V2__inbox_lifecycle.sql b/src/main/resources/db/migration/V2__inbox_lifecycle.sql new file mode 100644 index 0000000..888016f --- /dev/null +++ b/src/main/resources/db/migration/V2__inbox_lifecycle.sql @@ -0,0 +1,44 @@ +-- ===================================================================== +-- V2：信箱生命周期边界层（权威设计：docs/message-lifecycle.md §2/§4/§5.1/§5.2） +-- --------------------------------------------------------------------- +-- 三处契约级修正，均只动自有 PG（共享 MySQL 不建表、不改结构，红线不破）： +-- 1. 消费水位 W 落库（INBOX_CURSOR）：只随新 ID 成功入队推进、遇空洞即停， +-- 与入队同事务——中断后 W 未前进，重扫即补建（§4 第一行）； +-- HOLE_SINCE 支撑"空洞老化"，避免一次自增回滚空位永久停摆水位（§5.1）。 +-- 2. 回填事实并入 PROC_STATE：终态与回填意图同一条 UPDATE 落下， +-- 消除"业务已提交、回填待办未记"与"待办二次落账失败"两个崩溃窗口（§4）； +-- BACKFILL_TODO 随之下线（它承载的 sndr/type/styp/seqn 从未参与回填， +-- 共享信箱回填只需要 MSG_ID——§5.2 补写同样如此）。 +-- 3. 超期补写判据 RECEIVED_AT（§5.2 的 R）与 OPS-2「最老未处理信龄」锚点。 +-- ===================================================================== + +-- ① 消费水位：单行游标（CURSOR_ID 恒为 1） +CREATE TABLE INBOX_CURSOR ( + CURSOR_ID INT NOT NULL PRIMARY KEY, -- 恒为 1 + COMMITTED_UP_TO BIGINT NOT NULL, -- 水位 W：连续上界，无空洞 + HOLE_SINCE TIMESTAMP(6) WITH TIME ZONE, -- W+1 处空洞首次观测时刻；无空洞为 NULL + UPDATED_AT TIMESTAMP(6) WITH TIME ZONE NOT NULL +); +-- 初值 0：首轮从最小 ID 起重扫全部信箱行，入队幂等（ON CONFLICT DO NOTHING）， +-- 上线即自愈既有"已入队未回填"造成的窗口污染。 +INSERT INTO INBOX_CURSOR (CURSOR_ID, COMMITTED_UP_TO, HOLE_SINCE, UPDATED_AT) VALUES (1, 0, NULL, now()); + +-- ② PROC_STATE：接收时间 + 回填事实（替代 BACKFILL_TODO） +ALTER TABLE PROC_STATE ADD COLUMN RECEIVED_AT TIMESTAMP(6) WITH TIME ZONE; +ALTER TABLE PROC_STATE ADD COLUMN BACKFILL_AT TIMESTAMP(6) WITH TIME ZONE; -- 非空 = 已确认信箱行持有标记 +ALTER TABLE PROC_STATE ADD COLUMN BACKFILL_NEXT_AT TIMESTAMP(6) WITH TIME ZONE; -- 非空 = 待回填（终态同事务登记） +ALTER TABLE PROC_STATE ADD COLUMN BACKFILL_ATTEMPTS INT NOT NULL DEFAULT 0; +ALTER TABLE PROC_STATE ADD COLUMN BACKFILL_ERROR VARCHAR(512); + +-- 存量行接收时间：无法跨库回读 DATE_RECEIVED，以入队时间为下界（§5.2 判据只会偏晚，不会提前补写） +UPDATE PROC_STATE SET RECEIVED_AT = UPDATED_AT WHERE RECEIVED_AT IS NULL; + +-- 存量终态行补登记回填意图：立即交给回填扫描（已持有标记的行由空标守卫返回 0 行，幂等） +UPDATE PROC_STATE SET BACKFILL_NEXT_AT = now() +WHERE STATE IN ('SUCCEEDED', 'SKIPPED', 'DEAD') AND BACKFILL_AT IS NULL; + +-- 回填扫描索引：仅覆盖未确认标记的行 +CREATE INDEX idx_proc_backfill_due ON PROC_STATE (BACKFILL_NEXT_AT) WHERE BACKFILL_AT IS NULL; + +-- ③ BACKFILL_TODO 下线（回填意图已并入 PROC_STATE，机制收敛为一套；索引随表一并释放） +DROP TABLE BACKFILL_TODO; diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt index 7324eff..8dd3afb 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt @@ -45,6 +45,7 @@ class PipelineSmokeTest { fun cleanStubs() { ctx.getBean(StubProcState::class.java).clear() ctx.getBean(com.gzzn.omms.msgexchange.infra.stub.StubInbox::class.java).clear() + ctx.getBean(com.gzzn.omms.msgexchange.infra.stub.StubInboxCursor::class.java).clear() ctx.getBean(StubMsgEvents::class.java).clear() ctx.getBean(StubDeliveryPort::class.java).clear() } @@ -100,6 +101,30 @@ class PipelineSmokeTest { assertEquals(0, reopened.attempts) } + /** + * 回归（message-lifecycle §5.2/§5.3）：死信到达终态时同时登记回填意图并立即回填—— + * 否则永不回填的行会永久占据发现窗口，累积到批大小后收报整体停摆。 + */ + @Test + fun `dead letter reaches terminal state, gets marked and cannot block later discovery`() { + val inbox = ctx.getBean(com.gzzn.omms.msgexchange.infra.stub.StubInbox::class.java) + val proc = ctx.getBean(StubProcState::class.java) + val dead = inbox.simulateExternalWrite("") + ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce() + + pump.tick() // 解码 MALFORMED → DEAD + 回填意图（同一条 UPDATE）→ 提交后立即回填 + + val row = proc.find(dead)!! + assertEquals(ProcStatus.DEAD, row.state) + assertEquals(ErrorClass.MALFORMED, row.errorClass) + assertNotNull(row.backfillAt) + assertTrue(inbox.isMarked(dead)) + + val fresh = inbox.simulateExternalWrite("") + ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce() + assertNotNull(proc.find(fresh)) // 死信不阻断后续发现 + } + @Test fun `dispatcher flushes schd batch through stub port`() { val events = ctx.getBean(StubMsgEvents::class.java) diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/health/HealthIndicatorsTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/health/HealthIndicatorsTest.kt index 1498547..44146f9 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/health/HealthIndicatorsTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/health/HealthIndicatorsTest.kt @@ -1,6 +1,12 @@ package com.gzzn.omms.msgexchange.infra.health +import com.gzzn.omms.msgexchange.MutableClock import com.gzzn.omms.msgexchange.delivery.DeliveryPort +import com.gzzn.omms.msgexchange.domain.ProcStatus +import com.gzzn.omms.msgexchange.infra.persistence.InboxCursorRepository +import com.gzzn.omms.msgexchange.infra.stub.StubInbox +import com.gzzn.omms.msgexchange.infra.stub.StubInboxCursor +import com.gzzn.omms.msgexchange.infra.stub.StubProcState import io.micronaut.health.HealthStatus import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Test @@ -32,4 +38,35 @@ class HealthIndicatorsTest { assertEquals(HealthStatus.DOWN, kafkaHealth(ThrowingPort()).status) assertEquals(HealthStatus.DOWN, kafkaHealth(null).status) // bean 缺失 } + + /** OPS-2（message-lifecycle §5.3）：积压、最老未处理信龄、未回填终态与水位滞后。 */ + @Test + fun `inbox lifecycle reports backlog, oldest age, unmarked terminals and watermark lag`() { + val proc = StubProcState() + val inbox = StubInbox().apply { clear() } + val cursor = StubInboxCursor() + val oldest = MutableClock.BASE.minusSeconds(3600) + proc.insertIfAbsent(1L, oldest) + proc.insertIfAbsent(2L, MutableClock.BASE) + proc.markTerminal(2L, ProcStatus.SUCCEEDED) + val newest = inbox.insertRaw("") + cursor.save(InboxCursorRepository.Cursor(committedUpTo = newest - 2)) + + val result = lifecycleHealth(proc, cursor, inbox, now = MutableClock.BASE) + + assertEquals(HealthStatus.UP, result.status) + @Suppress("UNCHECKED_CAST") + val details = result.details as Map + assertEquals(1, details["backlog"]) + assertEquals(3600L, details["oldestUnprocessedSeconds"]) + assertEquals(1, details["unmarkedTerminal"]) + assertEquals(2L, details["watermarkLag"]) + } + + @Test + fun `inbox lifecycle stays up when ports are not bound`() { + val result = lifecycleHealth(procState = null, cursor = null, mailbox = null) + + assertEquals(HealthStatus.UP, result.status) // 可用性由依赖自身指示器承担 + } } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt index de564b7..c77cd95 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt @@ -9,10 +9,12 @@ import org.junit.jupiter.api.Test import java.sql.DriverManager /** - * Flyway 迁移引擎端到端验证（docs/flight-state.md §2 权威模型 + design.md §2.1）： - * 1. V1__flight_state_baseline.sql 在真实 PostgreSQL 上自动迁移成功； + * Flyway 迁移引擎端到端验证（docs/flight-state.md §2 权威模型 + design.md §2.1 + + * message-lifecycle.md §5.1/§5.2）： + * 1. V1 基线 + V2 信箱生命周期迁移在真实 PostgreSQL 上自动成功； * 2. flyway_schema_history 落库且 success = true； - * 3. 决策层/管道层/留痕层全表就绪；PIPELINE_LOCK 单行种子就位。 + * 3. 决策层/管道层/留痕层全表就绪；PIPELINE_LOCK 与 INBOX_CURSOR 单行种子就位； + * 4. V2 收敛结果成立：回填事实并入 PROC_STATE，BACKFILL_TODO 下线。 */ class FlywayMigrationTest { @@ -20,11 +22,11 @@ class FlywayMigrationTest { "pipeline_lock", "proc_state", "flight_schd", "flight_gate", "flight_checkin", "flight_belt", "flight_stand_plan", "flight_chute", "flight_delay", "flight_bridge_op", "flight_chock_op", "flight_route_point", - "msg_event", "req_track", "backfill_todo", "schd_snap_log", + "msg_event", "req_track", "inbox_cursor", "schd_snap_log", ) @Test - fun `Flyway migrates flight-state baseline onto real PostgreSQL`() { + fun `Flyway migrates flight-state baseline and inbox lifecycle onto real PostgreSQL`() { assumeTrue(PgTestSupport.canConnect(), PgTestSupport.skipMessage()) val url = PgTestSupport.jdbcUrl val user = PgTestSupport.user @@ -39,7 +41,7 @@ class FlywayMigrationTest { DriverManager.getConnection(url, user, pass).use { conn -> conn.createStatement().use { stmt -> - // 迁移记录：单基线 V1 + // 迁移记录：V1 基线 + V2 信箱生命周期 stmt.executeQuery( "SELECT version, script, success FROM flyway_schema_history ORDER BY installed_rank ASC", ).use { rs -> @@ -47,13 +49,15 @@ class FlywayMigrationTest { while (rs.next()) { records.add(Triple(rs.getString("version"), rs.getString("script"), rs.getBoolean("success"))) } - assertTrue(records.isNotEmpty(), "flyway_schema_history must record migrations") + assertTrue(records.size >= 2, "flyway_schema_history must record both migrations") assertEquals("1", records[0].first) assertEquals("V1__flight_state_baseline.sql", records[0].second) - assertTrue(records[0].third) + assertEquals("2", records[1].first) + assertEquals("V2__inbox_lifecycle.sql", records[1].second) + assertTrue(records.all { it.third }) } - // §3.2 全表就绪 + // 全表就绪（V2 后 backfill_todo 下线） stmt.executeQuery( "SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'", ).use { rs -> @@ -61,6 +65,7 @@ class FlywayMigrationTest { while (rs.next()) tables.add(rs.getString("table_name")) val missing = expectedTables - tables assertTrue(missing.isEmpty(), "missing tables: $missing") + assertTrue("backfill_todo" !in tables, "BACKFILL_TODO 已由 PROC_STATE 回填列取代") } // 身份不变量基础结构：FLIGHT_SCHD 主键 + STATE 列 @@ -73,11 +78,31 @@ class FlywayMigrationTest { assertEquals(setOf("flid", "operation_day", "state", "state_version", "last_msg_id"), cols) } - // PIPELINE_LOCK 单行种子（§3.2） + // 回填事实并入 PROC_STATE（§5.2 判据 + 回填待办） + stmt.executeQuery( + "SELECT column_name FROM information_schema.columns WHERE table_name = 'proc_state' " + + "AND column_name IN ('received_at', 'backfill_at', 'backfill_next_at', " + + "'backfill_attempts', 'backfill_error')", + ).use { rs -> + val cols = mutableSetOf() + while (rs.next()) cols.add(rs.getString("column_name")) + assertEquals( + setOf("received_at", "backfill_at", "backfill_next_at", "backfill_attempts", "backfill_error"), + cols, + ) + } + + // PIPELINE_LOCK 与 INBOX_CURSOR 单行种子（§3.2 / §5.1） stmt.executeQuery("SELECT count(*) FROM pipeline_lock WHERE lock_id = 1").use { rs -> assertTrue(rs.next()) assertEquals(1, rs.getInt(1)) } + stmt.executeQuery("SELECT committed_up_to, hole_since FROM inbox_cursor WHERE cursor_id = 1").use { rs -> + assertTrue(rs.next(), "INBOX_CURSOR 单行种子必须就位") + assertEquals(0L, rs.getLong("committed_up_to")) + rs.getTimestamp("hole_since") + assertTrue(rs.wasNull(), "初始无空洞观测") + } } } } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/InboxLifecycleJdbcSqlTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/InboxLifecycleJdbcSqlTest.kt new file mode 100644 index 0000000..0e1bfa9 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/InboxLifecycleJdbcSqlTest.kt @@ -0,0 +1,221 @@ +package com.gzzn.omms.msgexchange.infra.persistence.jdbc + +import com.gzzn.omms.msgexchange.domain.ErrorClass +import com.gzzn.omms.msgexchange.domain.ProcStatus +import com.gzzn.omms.msgexchange.infra.persistence.InboxCursorRepository +import org.h2.jdbcx.JdbcDataSource +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.Test +import java.sql.Connection +import java.sql.Timestamp +import java.time.Duration +import java.time.Instant +import java.util.UUID +import javax.sql.DataSource + +/** + * 信箱生命周期 SQL 语义（message-lifecycle §4/§5.1/§5.2/§11）在真实 JDBC 上的验证。 + * 以 H2 的 PostgreSQL 兼容模式承载与 V1+V2 等价的表结构——不依赖 docker/外接库， + * 覆盖 PG 侧（终态与回填意图同体、到期/超期筛选、积压观测、水位游标）与 MySQL 侧 + * （区间发现、标记单调）的实际语句行为。 + * + * 说明：H2 不支持 `INSERT ... ON CONFLICT DO NOTHING`， + * [JdbcProcStateRepository.insertIfAbsent] 的入队幂等由 PG 语义与 + * `InboxPollerTest`（重复轮询不再入队）分别保证。 + */ +class InboxLifecycleJdbcSqlTest { + + private lateinit var ds: DataSource + private lateinit var proc: JdbcProcStateRepository + private lateinit var mailbox: JdbcCminmsgInboxRepository + private lateinit var cursor: JdbcInboxCursorRepository + + private val t0: Instant = Instant.parse("2026-09-08T03:00:00Z") + private val overdue: Instant = t0.minus(Duration.ofDays(31)) + + @BeforeEach + fun setUp() { + ds = JdbcDataSource().apply { + setURL("jdbc:h2:mem:inbox-${UUID.randomUUID()};DB_CLOSE_DELAY=-1;MODE=PostgreSQL") + setUser("sa") + setPassword("") + } + ds.connection.use { conn -> + conn.createStatement().use { st -> + st.execute(PROC_STATE_DDL) + st.execute(CURSOR_DDL) + st.execute(MAILBOX_DDL) + st.execute( + "INSERT INTO inbox_cursor (cursor_id, committed_up_to, hole_since, updated_at) " + + "VALUES (1, 0, NULL, CURRENT_TIMESTAMP)", + ) + } + } + proc = JdbcProcStateRepository(ds) + mailbox = JdbcCminmsgInboxRepository(ds) + cursor = JdbcInboxCursorRepository(ds) + } + + @Test + fun `terminal write carries the backfill intent in the same statement`() { + seed(11L, t0) + + proc.markTerminal(11L, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = "raw-missing", now = t0) + + val row = proc.find(11L)!! + assertEquals(ProcStatus.DEAD, row.state) + assertEquals(ErrorClass.MALFORMED, row.errorClass) + assertEquals(t0, row.backfillNextAt) // 终态与回填意图同一条 UPDATE（§2/§4） + assertNull(row.backfillAt) + assertEquals(t0, row.receivedAt) + } + + @Test + fun `backfill failure is recorded on the same row without touching the terminal state`() { + seed(11L, t0) + proc.markTerminal(11L, ProcStatus.SUCCEEDED, now = t0) + + proc.recordBackfillFailure(11L, "mysql-down", attempts = 2, nextAttemptAt = t0.plusSeconds(120), now = t0) + + val row = proc.find(11L)!! + assertEquals(ProcStatus.SUCCEEDED, row.state) // §11 终态不可逆 + assertEquals(2, row.backfillAttempts) + assertEquals(t0.plusSeconds(120), row.backfillNextAt) + assertEquals("mysql-down", row.backfillError) + } + + @Test + fun `due query honours due time, the overdue deadline and terminal-only filter`() { + // ① 到期 + seed(1L, t0) + proc.markTerminal(1L, ProcStatus.SUCCEEDED, now = t0) + // ② 未到期（退避推后） + seed(2L, t0) + proc.markTerminal(2L, ProcStatus.SUCCEEDED, now = t0) + proc.recordBackfillFailure(2L, "mysql-down", 1, t0.plus(Duration.ofMinutes(15)), t0) + // ③ 未到期但接收时间已达超期期限 R（§5.2 覆盖退避） + seed(3L, overdue) + proc.markTerminal(3L, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, now = t0) + proc.recordBackfillFailure(3L, "mysql-down", 1, t0.plus(Duration.ofMinutes(15)), t0) + // ④ 中间态：永不补写（§5.2） + seed(4L, overdue) + // ⑤ 已确认标记 + seed(5L, t0) + proc.markTerminal(5L, ProcStatus.SUCCEEDED, now = t0) + proc.markBackfilled(5L, t0) + + val due = proc.findBackfillDue(t0, t0.minus(Duration.ofDays(30)), limit = 100).map { it.msgId } + + assertEquals(listOf(1L, 3L), due) + assertEquals(listOf(1L), proc.findBackfillDue(t0, t0.minus(Duration.ofDays(30)), limit = 1).map { it.msgId }) + } + + @Test + fun `backlog reports unfinished messages, oldest receive time and unmarked terminals`() { + seed(1L, overdue) + seed(2L, t0) + proc.markTerminal(2L, ProcStatus.SUCCEEDED, now = t0) + + val backlog = proc.backlog() + + assertEquals(1, backlog.unfinished) + assertEquals(overdue, backlog.oldestReceivedAt) // OPS-2 最老未处理信龄锚点 + assertEquals(1, backlog.unmarkedTerminal) + } + + @Test + fun `watermark round trips with and without a pending hole`() { + assertEquals(InboxCursorRepository.Cursor(0L, null), cursor.load()) + + cursor.save(InboxCursorRepository.Cursor(42L, t0)) + + assertEquals(InboxCursorRepository.Cursor(42L, t0), cursor.load()) + } + + @Test + fun `mailbox range read ignores processing marks and marking never overwrites`() { + val first = mailbox.insertRaw("") + val second = mailbox.insertRaw("") + val third = mailbox.insertRaw("") + + // 发现按 ID 区间；处理标记不参与扫描谓词 + assertEquals(listOf(first, second, third), mailbox.readRange(0L, 50).map { it.msgId }) + assertNotNull(mailbox.readRange(0L, 50).first().receivedAt) + assertEquals(listOf(second, third), mailbox.readRange(first, 50).map { it.msgId }) + assertEquals(third, mailbox.maxId()) + + assertTrue(mailbox.markProcessedIfUnmarked(second, "PROCESSED")) + assertFalse(mailbox.markProcessedIfUnmarked(second, "OTHER")) // §11 只把空标写为已处理 + assertEquals("PROCESSED", statusOf(second)) + // 已标记行仍出现在区间读结果中（发现与标记彻底解耦） + assertEquals(listOf(first, second, third), mailbox.readRange(0L, 50).map { it.msgId }) + assertFalse(mailbox.markProcessedIfUnmarked(999L, "PROCESSED")) + } + + private fun seed(msgId: Long, receivedAt: Instant?, state: ProcStatus = ProcStatus.PENDING) { + ds.connection.use { conn: Connection -> + conn.prepareStatement( + "INSERT INTO proc_state (msg_id, state, received_at, backfill_attempts, updated_at) VALUES (?, ?, ?, 0, ?)", + ).use { ps -> + ps.setLong(1, msgId) + ps.setString(2, state.name) + ps.setTimestamp(3, receivedAt?.let { Timestamp.from(it) }) + ps.setTimestamp(4, Timestamp.from(t0)) + ps.executeUpdate() + } + } + } + + private fun statusOf(msgId: Long): String? = + ds.connection.use { conn: Connection -> + conn.prepareStatement("SELECT CMINMSGS_STATUS FROM cminmsgs WHERE CMINMSGS_ID = ?").use { ps -> + ps.setLong(1, msgId) + ps.executeQuery().use { rs -> if (rs.next()) rs.getString(1) else null } + } + } + + private companion object { + const val PROC_STATE_DDL = """ + CREATE TABLE proc_state ( + msg_id BIGINT PRIMARY KEY, + state VARCHAR(16) NOT NULL, + identity_key VARCHAR(200), + attempts INT NOT NULL DEFAULT 0, + next_attempt_at TIMESTAMP WITH TIME ZONE, + error_class VARCHAR(20), + last_error VARCHAR(1000), + received_at TIMESTAMP WITH TIME ZONE, + backfill_at TIMESTAMP WITH TIME ZONE, + backfill_next_at TIMESTAMP WITH TIME ZONE, + backfill_attempts INT NOT NULL DEFAULT 0, + backfill_error VARCHAR(512), + updated_at TIMESTAMP WITH TIME ZONE NOT NULL, + CONSTRAINT uk_proc_identity UNIQUE (identity_key) + ) + """ + + const val CURSOR_DDL = """ + CREATE TABLE inbox_cursor ( + cursor_id INT PRIMARY KEY, + committed_up_to BIGINT NOT NULL, + hole_since TIMESTAMP WITH TIME ZONE, + updated_at TIMESTAMP WITH TIME ZONE NOT NULL + ) + """ + + const val MAILBOX_DDL = """ + CREATE TABLE cminmsgs ( + CMINMSGS_ID BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + CMINMSGS_CLOB_MSG CLOB, + CMINMSGS_DATE_RECEIVED TIMESTAMP, + CMINMSGS_DATE_PROCESSED TIMESTAMP, + CMINMSGS_STATUS VARCHAR(32) + ) + """ + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt index f4ebccd..4182ca5 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt @@ -23,8 +23,7 @@ class ReplayServiceTest { rows[id] = ProcState(id, status, identityKey = "k$id", attempts = 3, errorClass = ec, lastError = "x") } - override fun insert(msgId: Long, state: ProcStatus) = Unit - override fun exists(msgId: Long): Boolean = rows.containsKey(msgId) + override fun insertIfAbsent(msgId: Long, receivedAt: Instant?): Boolean = rows.putIfAbsent(msgId, ProcState(msgId, ProcStatus.PENDING, receivedAt = receivedAt)) == null override fun find(msgId: Long): ProcState? = rows[msgId] override fun findSuccessTerminal(msgId: Long): Boolean = rows[msgId]?.state == ProcStatus.SUCCEEDED override fun headUnfinished(): ProcState? = null @@ -35,6 +34,20 @@ class ReplayServiceTest { errorClass: ErrorClass?, lastError: String?, ) = Unit + override fun markTerminal( + msgId: Long, state: ProcStatus, errorClass: ErrorClass?, lastError: String?, + attempts: Int?, now: Instant, + ) = Unit + + override fun markBackfilled(msgId: Long, now: Instant) = Unit + + override fun recordBackfillFailure(msgId: Long, error: String?, attempts: Int, nextAttemptAt: Instant, now: Instant) = Unit + + override fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int) = + emptyList() + + override fun backlog() = com.gzzn.omms.msgexchange.infra.persistence.Backlog(0, null, 0) + override fun requeueByErrorClasses(errorClasses: List): Int { requeueCalls += errorClasses var n = 0 diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPollerTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPollerTest.kt index ad239e9..1a8193d 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPollerTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPollerTest.kt @@ -1,47 +1,120 @@ package com.gzzn.omms.msgexchange.ingress +import com.gzzn.omms.msgexchange.config.PipelineProps import com.gzzn.omms.msgexchange.domain.ErrorClass import com.gzzn.omms.msgexchange.domain.ProcStatus import com.gzzn.omms.msgexchange.infra.stub.StubInbox +import com.gzzn.omms.msgexchange.infra.stub.StubInboxCursor +import com.gzzn.omms.msgexchange.infra.stub.StubPipelineTx import com.gzzn.omms.msgexchange.infra.stub.StubProcState -import com.gzzn.omms.msgexchange.processing.Pump -import io.micronaut.test.extensions.junit5.annotation.MicronautTest -import jakarta.inject.Inject import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertNull import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.Test +import java.time.Instant -/** 主路径：JDBC 轮询语义（stub 下 InboxPoller + 外部写信箱模拟）。 */ -@MicronautTest +/** + * 收报发现权不变量（docs/message-lifecycle.md §5.1/§5.3 + architecture.md §5 严格 FIFO）： + * - 扫描按 ID 区间，**不受处理标记影响**：终态而未回填的行不得占据批次、不得阻断新信发现； + * - 水位只随成功入队推进，且与入队同事务（中断后由重扫补建）； + * - 遇空洞即停（较小 ID 未入队时不得被后续消息越过）；空洞老化后放行（水位不得永久停摆）； + * - 收报层不写处理标记。 + */ class InboxPollerTest { - @Inject lateinit var poller: InboxPoller - @Inject lateinit var pump: Pump - @Inject lateinit var stubInbox: StubInbox - @Inject lateinit var stubProc: StubProcState + private val t0: Instant = Instant.parse("2026-09-08T03:00:00Z") + private val props = PipelineProps() + + private lateinit var inbox: StubInbox + private lateinit var proc: StubProcState + private lateinit var cursor: StubInboxCursor + private lateinit var poller: InboxPoller @BeforeEach - fun clean() { - stubInbox.clear() - stubProc.clear() + fun setUp() { + inbox = StubInbox().apply { clear() } + proc = StubProcState().apply { clear() } + cursor = StubInboxCursor().apply { clear() } + poller = InboxPoller(inbox, proc, cursor, StubPipelineTx(), props) } @Test - fun `poller enqueues externally written mailbox rows`() { - val id = stubInbox.simulateExternalWrite("") - assertEquals(1, poller.pollOnce()) - assertNotNull(stubProc.snapshotOf(id)) - assertEquals(ProcStatus.PENDING, stubProc.snapshotOf(id)!!.state) + fun `external rows are enqueued in id order and advance the watermark without marking the mailbox`() { + val first = inbox.simulateExternalWrite("") + val second = inbox.simulateExternalWrite("") + + assertEquals(2, poller.pollOnce(t0)) + + assertEquals(ProcStatus.PENDING, proc.find(first)!!.state) + assertEquals(ProcStatus.PENDING, proc.find(second)!!.state) + assertEquals(second, cursor.cursor.committedUpTo) + assertNull(cursor.cursor.holeSince) + // §5.3：消化阶段只写自有 PG，不触碰信箱标记 + assertFalse(inbox.isMarked(first)) + assertEquals(0, poller.pollOnce(t0)) // 重复扫描幂等 + } + + /** + * 回归（US-01 条目 3 / §5.3「积压挡批」）：终态且永不回填的行（解码失败死信等）曾占满 + * 有限批次使收报整体停摆——发现必须与处理标记彻底解耦。 + */ + @Test + fun `terminal rows without a mailbox mark do not block discovery of later messages`() { + props.pipeline.claimBatch = 3 + val dead = (1..3).map { inbox.simulateExternalWrite("") } + assertEquals(3, poller.pollOnce(t0)) + dead.forEach { + proc.markTerminal(it, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = "raw-missing") + } + + val fresh = inbox.simulateExternalWrite("") + + assertEquals(1, poller.pollOnce(t0)) + assertEquals(ProcStatus.PENDING, proc.find(fresh)!!.state) + assertEquals(fresh, cursor.cursor.committedUpTo) } @Test - fun `poller is idempotent for already enqueued rows`() { - val id = stubInbox.simulateExternalWrite("") - poller.pollOnce() - assertEquals(0, poller.pollOnce()) - pump.tick() - assertEquals(ProcStatus.DEAD, stubProc.snapshotOf(id)!!.state) - assertEquals(ErrorClass.MALFORMED, stubProc.snapshotOf(id)!!.errorClass) + fun `watermark stops at a hole so later ids cannot overtake a missing smaller id`() { + val first = inbox.simulateExternalWrite("") + val hole = inbox.simulateExternalWrite("") + val afterHole = inbox.simulateExternalWrite("") + inbox.removeRow(hole) + + assertEquals(1, poller.pollOnce(t0)) + + assertEquals(first, cursor.cursor.committedUpTo) + assertNotNull(cursor.cursor.holeSince) + assertNull(proc.find(afterHole)) // 不得越过空洞入队（FIFO） + assertEquals(0, poller.pollOnce(t0.plusSeconds(60))) // 宽限期内水位不推进 + assertEquals(first, cursor.cursor.committedUpTo) + } + + @Test + fun `an aged hole is released and later ids resume enqueuing`() { + val hole = inbox.simulateExternalWrite("") + val afterHole = inbox.simulateExternalWrite("") + inbox.removeRow(hole) + poller.pollOnce(t0) // 记录空洞观测时刻 + + val agedOut = t0.plus(props.pipeline.maxCommitDelay) + assertEquals(0, poller.pollOnce(agedOut)) // 空洞判永久：推进水位但不越过入队 + + assertNull(cursor.cursor.holeSince) + assertEquals(afterHole - 1, cursor.cursor.committedUpTo) + assertEquals(1, poller.pollOnce(agedOut)) // 下一轮恢复发现 + assertEquals(afterHole, cursor.cursor.committedUpTo) + assertNotNull(proc.find(afterHole)) + } + + @Test + fun `compat http path and poller do not double enqueue the same message`() { + val receipt = InboxService(inbox, proc).accept("", t0) + + assertEquals(0, poller.pollOnce(t0)) + assertEquals(receipt.msgId, cursor.cursor.committedUpTo) // 已在 PG：读取进度照常推进 + assertNotNull(proc.find(receipt.msgId)) } } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/BackfillSweepJobTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/BackfillSweepJobTest.kt deleted file mode 100644 index efa7ee3..0000000 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/BackfillSweepJobTest.kt +++ /dev/null @@ -1,90 +0,0 @@ -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.values.single().attempts) - assertEquals("mysql-down", todo.lastErrorOf(901L)) - // 一次失败后 30s 内不再到期（退避生效） - assertEquals(0, todo.findDue(t0.plusSeconds(29)).size) - 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/BackfillServiceTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/BackfillServiceTest.kt new file mode 100644 index 0000000..6b14e87 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/BackfillServiceTest.kt @@ -0,0 +1,170 @@ +package com.gzzn.omms.msgexchange.processing + +import com.gzzn.omms.msgexchange.config.MailboxProps +import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.domain.ErrorClass +import com.gzzn.omms.msgexchange.domain.ProcStatus +import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository +import com.gzzn.omms.msgexchange.infra.persistence.MailboxRow +import com.gzzn.omms.msgexchange.infra.stub.StubInbox +import com.gzzn.omms.msgexchange.infra.stub.StubProcState +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import java.time.Clock +import java.time.Duration +import java.time.Instant +import java.time.ZoneOffset + +/** + * 回填通道不变量（docs/message-lifecycle.md §3/§4/§5.2/§11）： + * 终态即刻补写、失败退避重试、超期期限 R 覆盖退避、标记单调只写空标、 + * 中间态不适用、回填失败不回改终态、重启后扫描不依赖内存状态。 + */ +class BackfillServiceTest { + + private val t0: Instant = Instant.parse("2026-09-08T03:00:00Z") + private val props = PipelineProps() + + /** 可注入故障的信箱：验证回填失败路径（§3 失败处理）。 */ + private class FakeMailbox(var fail: Boolean = false) : CminmsgInboxRepository { + val marked = linkedSetOf() + override fun insertRaw(rawXml: String): Long = 1L + override fun rawOf(msgId: Long): String? = null + override fun readRange(fromExclusive: Long, limit: Int): List = emptyList() + override fun maxId(): Long? = null + override fun markProcessedIfUnmarked(msgId: Long, value: String): Boolean { + if (fail) throw IllegalStateException("mysql-down") + return marked.add(msgId) + } + } + + private fun service(proc: StubProcState, mailbox: CminmsgInboxRepository, now: Instant = t0) = + BackfillService(proc, mailbox, MailboxProps(), props, Clock.fixed(now, ZoneOffset.UTC)) + + /** 终态 + 回填意图（固定时刻，避免依赖真实时钟）。 */ + private fun succeeded(proc: StubProcState, id: Long) { + proc.insertIfAbsent(id, t0) + proc.markTerminal(id, ProcStatus.SUCCEEDED, now = t0) + } + + @Test + fun `terminal message is marked immediately and the intent is cleared`() { + val proc = StubProcState() + val inbox = StubInbox().apply { clear() } + val id = inbox.insertRaw("") + succeeded(proc, id) + + service(proc, inbox).attempt(id) + + assertEquals("PROCESSED", inbox.markOf(id)) + val row = proc.find(id)!! + assertNotNull(row.backfillAt) + assertNull(row.backfillNextAt) + assertNull(row.backfillError) + } + + @Test + fun `marking is monotonic - an existing value is never overwritten`() { + val inbox = StubInbox().apply { clear() } + val id = inbox.insertRaw("") + + assertTrue(inbox.markProcessedIfUnmarked(id, "PROCESSED")) + assertFalse(inbox.markProcessedIfUnmarked(id, "OTHER")) // §11 只把空标写为已处理 + assertEquals("PROCESSED", inbox.markOf(id)) + } + + @Test + fun `already marked rows are idempotent and never recorded as failures`() { + val proc = StubProcState() + val inbox = StubInbox().apply { clear() } + val id = inbox.insertRaw("") + succeeded(proc, id) + + val backfill = service(proc, inbox) + backfill.attempt(id) + backfill.attempt(id) // 重复执行无副作用（§5.2 补写只针对空标记） + + assertNull(proc.find(id)!!.backfillError) + assertEquals(0, proc.find(id)!!.backfillAttempts) + } + + @Test + fun `failed attempt records backoff and never touches the terminal state`() { + val proc = StubProcState() + val mailbox = FakeMailbox(fail = true) + succeeded(proc, 901L) + + service(proc, mailbox).attempt(901L) + + val row = proc.find(901L)!! + assertEquals(ProcStatus.SUCCEEDED, row.state) // §11 终态不可逆 + assertEquals(1, row.backfillAttempts) + assertEquals("mysql-down", row.backfillError) + assertEquals(t0.plus(Duration.ofSeconds(30)), row.backfillNextAt) + assertNull(row.backfillAt) + } + + @Test + fun `sweep retries due rows and completes once the mailbox recovers`() { + val proc = StubProcState() + val mailbox = FakeMailbox(fail = true) + succeeded(proc, 901L) + val backfill = service(proc, mailbox) + + assertEquals(1, backfill.sweep(t0)) // 到期 → 失败 → 退避 + assertEquals(0, backfill.sweep(t0.plusSeconds(29))) // 未到期 + assertEquals(1, backfill.sweep(t0.plusSeconds(30))) // 到期再试 → 仍失败（重启后同样收敛） + + mailbox.fail = false + assertEquals(1, backfill.sweep(t0.plusSeconds(90))) + + assertTrue(901L in mailbox.marked) + assertNotNull(proc.find(901L)!!.backfillAt) + } + + /** §5.2：超期期限 R 覆盖退避，保证库方清除前提「边界内无未标记行」在有限时间内成立。 */ + @Test + fun `overdue rows bypass the retry backoff`() { + val proc = StubProcState() + val inbox = StubInbox().apply { clear() } + val id = inbox.insertRaw("") + val old = t0.minus(props.pipeline.overdueBackfill).minusSeconds(60) + proc.insertIfAbsent(id, old) // 接收时间早于 R + proc.markTerminal(id, ProcStatus.SUCCEEDED, now = t0) + proc.recordBackfillFailure(id, "mysql-down", 5, t0.plus(Duration.ofMinutes(15)), t0) // 退避推到很远之后 + + assertEquals(1, service(proc, inbox).sweep(t0)) + + assertNotNull(proc.find(id)!!.backfillAt) + } + + /** §5.2：中间态（PENDING / FAILED）不适用超期补写——处理未完成时不打标。 */ + @Test + fun `mid states are never marked even when far past the deadline`() { + val proc = StubProcState() + val inbox = StubInbox().apply { clear() } + val pending = inbox.insertRaw("") + val failed = inbox.insertRaw("") + val old = t0.minus(props.pipeline.overdueBackfill).minusSeconds(60) + proc.insertIfAbsent(pending, old) + proc.insertIfAbsent(failed, old) + proc.update(failed, ProcStatus.FAILED, errorClass = ErrorClass.INFRA) + + assertEquals(0, service(proc, inbox).sweep(t0)) + + assertFalse(inbox.isMarked(pending)) + assertFalse(inbox.isMarked(failed)) + } + + @Test + fun `backoff delay doubles per attempt and caps at fifteen minutes`() { + assertEquals(Duration.ofSeconds(30), BackfillService.backoffDelayFor(1)) + assertEquals(Duration.ofMinutes(2), BackfillService.backoffDelayFor(3)) + assertEquals(Duration.ofMinutes(15), BackfillService.backoffDelayFor(10)) + assertEquals(Duration.ofMinutes(15), BackfillService.backoffDelayFor(50)) + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FdelAndAdftProcessorTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FdelAndAdftProcessorTest.kt index 4a85f14..164368d 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FdelAndAdftProcessorTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FdelAndAdftProcessorTest.kt @@ -13,12 +13,13 @@ import com.gzzn.omms.msgexchange.domain.ProcStatus import com.gzzn.omms.msgexchange.domain.Targets import com.gzzn.omms.msgexchange.domain.flight.FlightState import com.gzzn.omms.msgexchange.infra.persistence.PersistOutcome -import com.gzzn.omms.msgexchange.infra.stub.StubBackfillTodo +import com.gzzn.omms.msgexchange.infra.stub.StubProcState import com.gzzn.omms.msgexchange.infra.stub.StubFlightState import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents import com.gzzn.omms.msgexchange.infra.stub.StubPipelineLock import com.gzzn.omms.msgexchange.infra.stub.StubPipelineTx import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNotNull import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.Test import java.time.LocalDate @@ -59,9 +60,10 @@ class FdelAndAdftProcessorTest { fun `active flight deletion publishes single tombstone and keeps details`() { val f = flights() val events = StubMsgEvents() - val todo = StubBackfillTodo() + val proc = StubProcState() + proc.insertIfAbsent(msgId, null) - val result = FdelProcessor(StubPipelineTx(), StubPipelineLock(), f, events, todo, ObjectMapper()) + val result = FdelProcessor(StubPipelineTx(), StubPipelineLock(), f, events, proc, ObjectMapper()) .apply(head(), msg(), FlopPayload("121")) assertEquals(ApplyResult.Succeeded, result) @@ -71,14 +73,17 @@ class FdelAndAdftProcessorTest { assertEquals(1, tombstones.size) assertEquals(Targets.KAFKA_SCHD, tombstones.single().target) assertTrue(tombstones.single().payloadJson.contains("\"deleted\":true")) - assertEquals(1, todo.count()) + // 终态 + 回填意图由处理器在自己的事务内落库（message-lifecycle §2/§4） + val row = proc.find(msgId)!! + assertEquals(ProcStatus.SUCCEEDED, row.state) + assertNotNull(row.backfillNextAt) } @Test fun `repeated FDEL is idempotent without version bump or duplicate event`() { val f = flights() val events = StubMsgEvents() - val proc = FdelProcessor(StubPipelineTx(), StubPipelineLock(), f, events, null, ObjectMapper()) + val proc = FdelProcessor(StubPipelineTx(), StubPipelineLock(), f, events, StubProcState(), ObjectMapper()) proc.apply(head(), msg(), FlopPayload("121")) val version = f.findMainRow("121")!!.stateVersion @@ -94,7 +99,7 @@ class FdelAndAdftProcessorTest { val f = StubFlightState() val events = StubMsgEvents() - val result = FdelProcessor(StubPipelineTx(), StubPipelineLock(), f, events, null, ObjectMapper()) + val result = FdelProcessor(StubPipelineTx(), StubPipelineLock(), f, events, StubProcState(), ObjectMapper()) .apply(head(), msg(), FlopPayload("999")) assertEquals(ApplyResult.Succeeded, result) // §3.3：迟到/不存在幂等成功 @@ -107,7 +112,7 @@ class FdelAndAdftProcessorTest { f.markDeleted("121", msgId = 2, now = java.time.Instant.now()) val events = StubMsgEvents() val adft = AdftProcessor( - StubPipelineTx(), StubPipelineLock(), f, events, null, + StubPipelineTx(), StubPipelineLock(), f, events, StubProcState(), OperationDayProps().apply { zone = "Asia/Shanghai" }, ObjectMapper(), ) val record = com.gzzn.omms.msgexchange.domain.flight.ScheduleRecord( @@ -128,7 +133,7 @@ class FdelAndAdftProcessorTest { val f = StubFlightState() val events = StubMsgEvents() val adft = AdftProcessor( - StubPipelineTx(), StubPipelineLock(), f, events, null, + StubPipelineTx(), StubPipelineLock(), f, events, StubProcState(), OperationDayProps().apply { zone = "Asia/Shanghai" }, ObjectMapper(), ) @@ -154,7 +159,7 @@ class FdelAndAdftProcessorTest { msgId = 1, now = java.time.Instant.now(), ) val adft = AdftProcessor( - StubPipelineTx(), StubPipelineLock(), f, StubMsgEvents(), null, + StubPipelineTx(), StubPipelineLock(), f, StubMsgEvents(), StubProcState(), OperationDayProps().apply { zone = "Asia/Shanghai" }, ObjectMapper(), ) diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt index fb65e02..e4a0827 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt @@ -12,7 +12,6 @@ import com.gzzn.omms.msgexchange.domain.SnapshotFlag import com.gzzn.omms.msgexchange.domain.SnapshotResult import com.gzzn.omms.msgexchange.domain.Targets import com.gzzn.omms.msgexchange.domain.flight.FlightState -import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository import com.gzzn.omms.msgexchange.infra.persistence.FlightStateRepository import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository import com.gzzn.omms.msgexchange.infra.persistence.PersistOutcome @@ -24,8 +23,8 @@ import com.gzzn.omms.msgexchange.infra.stub.StubFlightState import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents import com.gzzn.omms.msgexchange.infra.stub.StubProcState import com.gzzn.omms.msgexchange.infra.stub.StubSnapshotLog -import com.gzzn.omms.msgexchange.infra.stub.StubBackfillTodo import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNotNull import org.junit.jupiter.api.Assertions.assertNull import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.Test @@ -70,7 +69,6 @@ class ScheduleProcessorTest { flights: StubFlightState = StubFlightState(), events: StubMsgEvents = StubMsgEvents(), log: StubSnapshotLog = StubSnapshotLog(), - todo: StubBackfillTodo? = StubBackfillTodo(), tx: PipelineTransactionManager? = null, ): ScheduleProcessor { val txRunner = tx ?: object : PipelineTransactionManager { @@ -83,20 +81,20 @@ class ScheduleProcessorTest { flightState = flights, msgEvents = events, snapshotLog = log, - backfillTodo = todo, operationDayProps = OperationDayProps().apply { zone = "Asia/Shanghai"; cutoffHour = 0 }, mapper = ObjectMapper(), ) } @Test - fun `happy path persists snapshot with version bump, events, snap log and backfill todo`() { + fun `happy path persists snapshot, events, terminal state and backfill intent in one transaction`() { + val proc = StubProcState() + proc.insertIfAbsent(msgId, null) val flights = StubFlightState() val events = StubMsgEvents() val log = StubSnapshotLog() - val todo = StubBackfillTodo() - val result = processor(flights = flights, events = events, log = log, todo = todo) + val result = processor(proc = proc, flights = flights, events = events, log = log) .applyScheduleRecords(head(), message(makeBody("121" to "15DEC261723"))) assertEquals(ApplyResult.Succeeded, result) @@ -108,13 +106,17 @@ class ScheduleProcessorTest { assertEquals(1, log.entries.size) assertEquals(SnapshotResult.COMMITTED, log.entries.single().result) assertEquals(1, log.entries.single().upserted) - assertEquals(1, todo.count()) // design.md §3.3/§6.1 事务内预登记 + // message-lifecycle §2/§4：终态与回填意图随业务写入在同一事务提交（不依赖主泵补写） + val row = proc.find(msgId)!! + assertEquals(ProcStatus.SUCCEEDED, row.state) + assertNotNull(row.backfillNextAt) } @Test fun `replay of succeeded message records idempotent success without writes`() { val proc = StubProcState() - proc.insert(msgId, ProcStatus.SUCCEEDED) + proc.insertIfAbsent(msgId, null) + proc.markTerminal(msgId, ProcStatus.SUCCEEDED) val flights = StubFlightState() val log = StubSnapshotLog() @@ -190,17 +192,16 @@ class ScheduleProcessorTest { val proc = StubProcState() val flights = StubFlightState() val log = StubSnapshotLog() - val todo = StubBackfillTodo() - proc.insert(msgId) - val p = processor(proc = proc, flights = flights, log = log, todo = todo, tx = TxRunner { true }) + proc.insertIfAbsent(msgId, null) + val p = processor(proc = proc, flights = flights, log = log, tx = TxRunner { true }) // design.md §2.3：数据库/内部故障 → 异常上抛，MessageProcessor 记 FAILED(INFRA) 退避；不写任何终态 org.junit.jupiter.api.Assertions.assertThrows(IllegalStateException::class.java) { p.applyScheduleRecords(head(), message(makeBody("121" to "15DEC261723"))) } assertNull(flights.findMainRow("121")) - assertEquals(0, todo.count()) assertEquals(0, log.entries.size) assertEquals(ProcStatus.PENDING, proc.find(msgId)!!.state) + assertNull(proc.find(msgId)!!.backfillNextAt) // 事务回滚：终态与回填意图都不落库 } } diff --git a/src/test/resources/application-test.yml b/src/test/resources/application-test.yml index 865d627..7ec0e5e 100644 --- a/src/test/resources/application-test.yml +++ b/src/test/resources/application-test.yml @@ -4,6 +4,8 @@ # U07：stubs=true 提供内存仓储使全链路（Controller→Inbox→Pump→Dispatcher）可装配； # autostart=false：测试内不自动拉起后台循环（避免泄漏线程），按需手动 tick。 # ACM2-12：main 的 datasources.default（自有 PG）默认 enabled=false，测试以 H2 内存替代并显式启用。 +# V2：BackfillService 注入 MailboxProps，其 shared-mysql.* 占位符必须在测试环境可解析 +# （信箱不启用：enabled=false，仅需占位符取值）。 msgx: register-eureka: false stubs: true @@ -18,6 +20,15 @@ datasources: password: "" driver-class-name: org.h2.Driver +mailbox: + processed-value: PROCESSED + shared-mysql: + enabled: false + url: jdbc:mysql://127.0.0.1:3306/cdairport + username: msgx_test + password: msgx_test + driver-class-name: com.mysql.cj.jdbc.Driver + flyway: datasources: default: