From 891088235cd434579a5e9318f332315594058df2 Mon Sep 17 00:00:00 2001 From: windyboy Date: Mon, 21 Sep 2026 14:49:40 +0800 Subject: [PATCH] =?UTF-8?q?feat(ingress):=20=E6=8C=89=E5=A4=84=E7=90=86?= =?UTF-8?q?=E6=97=B6=E9=97=B4=E4=B8=BA=E7=A9=BA=E6=89=AB=E6=8F=8F=E4=BF=A1?= =?UTF-8?q?=E7=AE=B1=EF=BC=8C=E7=A7=BB=E9=99=A4=E6=B0=B4=E4=BD=8D=E6=B6=88?= =?UTF-8?q?=E8=B4=B9=EF=BC=88ACM2-78=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit readUnprocessed 替代 ID 水位；去掉 INBOX_CURSOR/cutover-watermark/max-commit-delay 及相关健康与指标。 Co-authored-by: Cursor --- deploy/dev/mysql-init/02-mailbox-messages.sql | 2 +- deploy/integration/seed-flight.sql | 2 +- docs/implementation.md | 1 + docs/reference.md | 6 +- docs/specification.md | 1 - .../omms/msgexchange/config/PipelineProps.kt | 30 ---- .../health/InboxLifecycleHealthIndicator.kt | 15 +- .../infra/metrics/PipelineMetrics.kt | 12 -- .../infra/persistence/Repositories.kt | 60 ++----- .../jdbc/JdbcCminmsgInboxRepository.kt | 20 +-- .../persistence/jdbc/JdbcPgRepositories.kt | 86 --------- .../infra/stub/StubRepositories.kt | 32 +--- .../omms/msgexchange/ingress/InboxPoller.kt | 126 +++---------- .../gzzn/omms/msgexchange/processing/Pump.kt | 26 --- src/main/resources/application.yml | 5 +- .../migration/V1__flight_state_baseline.sql | 46 ++--- .../db/migration/oracle11g/README.md | 2 +- .../omms/msgexchange/PipelineSmokeTest.kt | 45 ++--- .../config/PipelineConfigCheckTest.kt | 9 - .../msgexchange/config/PipelinePropsTest.kt | 13 -- .../infra/health/HealthIndicatorsTest.kt | 14 +- .../infra/metrics/PipelineMetricsTest.kt | 2 - .../persistence/jdbc/FlywayMigrationTest.kt | 15 +- .../jdbc/InboxLifecycleJdbcSqlTest.kt | 45 +---- .../msgexchange/ingress/CutoverSeedTest.kt | 164 ----------------- .../msgexchange/ingress/InboxPollerTest.kt | 169 +++++++----------- .../msgexchange/ingress/InboxServiceTest.kt | 11 +- .../processing/BackfillServiceTest.kt | 8 +- .../processing/IgnoreBranchTest.kt | 4 - 29 files changed, 166 insertions(+), 805 deletions(-) delete mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/ingress/CutoverSeedTest.kt diff --git a/deploy/dev/mysql-init/02-mailbox-messages.sql b/deploy/dev/mysql-init/02-mailbox-messages.sql index b271653..68e5b2d 100644 --- a/deploy/dev/mysql-init/02-mailbox-messages.sql +++ b/deploy/dev/mysql-init/02-mailbox-messages.sql @@ -5,7 +5,7 @@ -- -- 报文格式：SIS 接口规范（docs/legacy/SIS_AODB_RMS-V0.1.md）META + 类型体， -- XML 按 docs/legacy/unisysaodbsis.xsd；只 INSERT 契约内列，不建表不变更（C-2）。 --- INBOX_CURSOR 初值 W=0，应用首启即全部消费。SEQN 1–5，与 01-mailbox.sql +-- 处理时间为空，应用首启即全部消费。SEQN 1–5，与 01-mailbox.sql -- （仅建表、无数据）无冲突。 -- -- 两条参考数据报文验证的是【现状行为】，US-13 落地后行为会变，需同步更新： diff --git a/deploy/integration/seed-flight.sql b/deploy/integration/seed-flight.sql index 434c5c3..edd107d 100644 --- a/deploy/integration/seed-flight.sql +++ b/deploy/integration/seed-flight.sql @@ -9,7 +9,7 @@ -- 故按 FLID 先删后插（implementation「字段与集合」的完整合并口径）。 -- CREATED_AT/UPDATED_AT 为 NOT NULL，显式给值。 -- 只写决策层（FLIGHT_SCHD + 明细），不碰管道层（PROC_STATE/MSG_EVENT/ --- INBOX_CURSOR/PIPELINE_LOCK/REQ_TRACK/SCHD_SNAP_LOG）。 +-- PIPELINE_LOCK/REQ_TRACK/SCHD_SNAP_LOG）。 -- 执行：psql -h 127.0.0.1 -p 15432 -U msgx_dev -d msgx -f 本文件 -- ===================================================================== diff --git a/docs/implementation.md b/docs/implementation.md index cef8316..056f238 100644 --- a/docs/implementation.md +++ b/docs/implementation.md @@ -76,6 +76,7 @@ msgexchange-v2 怎么处理报文：记录模型、状态机、事务边界、 1. 信箱不可读 → 记日志，下轮再试（不是「无新消息」）。 2. 同事务 `insertIfAbsent(MSG_ID, RECEIVED_AT, ENQUEUED_AT)`；冲突 = 已有记录，跳过。 3. 已写回处理时间的行不再出现；终态未写回的行会重复扫到，幂等跳过。 +4. 读满一批就按编号续读下一批，续读起点只在本轮有效：否则未写回的行攒满一批之后，后面的新行再也读不到。 写回延迟期间扫描量会涨，须观测积压（[reference.md](reference.md) 指标）。 diff --git a/docs/reference.md b/docs/reference.md index 9f883c7..df5d48c 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -22,7 +22,7 @@ | `msgx.pipeline.event-retention` | `7d` | 发送成功的消息在数据库保留多久，从 `SENT_AT` 起算 | 暂定 | | `msgx.pipeline.terminal-retention` | `90d` | 处理记录（PROC_STATE 终态行）保留多久，从 `UPDATED_AT` 起算；到期且已回填才删除（`US-11`） | 暂定 | -### 写回信箱与编号扫描 +### 写回信箱 | 参数 | 默认值 | 用途与约束 | 依据 | |---|---|---|---| @@ -31,8 +31,6 @@ | `msgx.pipeline.backfill-max-attempts` | `100` 次 | 失败次数达到此值时告警，仍继续重试直到期限 `R` | 暂定 | | `msgx.pipeline.backfill-backoff-ms` | `30000` ms | 首次写回失败后等多久再试 | 暂定 | | `msgx.pipeline.backfill-backoff-cap-ms` | `900000` ms | 以后每次重试最长等多久 | 暂定 | -| `msgx.pipeline.max-commit-delay` | `5m` | 当前按编号读信箱时，遇到缺号最多等待多久；改用处理时间筛选后删除（`G-SCAN-PREDICATE`） | 待替换 | -| `msgx.pipeline.cutover-watermark` | 不设置 | 首次启动时从哪个编号开始读信箱；可填 `min`、`zero`、`max` 或具体编号；改用处理时间筛选后删除（`G-SCAN-PREDICATE`） | 待替换 | ### 固定调度 @@ -89,7 +87,6 @@ | `msgx.operation-day.cutoff-hour` | 取值 0–23 | `PARAM:msgx.operation-day.cutoff-hour` | | `kafka.producers.default.acks` / `enable-idempotence` / `max-in-flight-requests-per-connection` | 联合满足幂等生产前提（`all`+`true`+`1`；旧 Broker 降级组合见 `D2`） | `D2` | | `msgx.pipeline.backoff-ms` | 档位数必须等于 `max-attempts - 1` | 退避表（implementation.md「参数」） | -| `msgx.pipeline.cutover-watermark` | 仅 `min`/`zero`/`max`/数字 | 切流播种（`G-SCAN-PREDICATE` 前） | **缺值**：带环境变量占位的必需键（`MSGX_PG_URL`、`MSGX_MAILBOX_URL`、`MSGX_KAFKA_SERVERS` 等）缺失时由 DI 解析失败拒绝启动，无需自检代码。 @@ -114,7 +111,6 @@ | `msgx.pipeline.processing.ignored.total` | 被 `IgnoreRules` 跳过的消息数 | 现行清单仅 `LDM-*`；`REGN`/`RSTA`/`EROR` 已改走 US-13/US-09 | | `msgx.pipeline.delivery.send_failures.total{target}` | 该投递目标累计发送失败次数（进程内，重启归零） | 持续增长且 `dead` 非零：投递链路故障（`OPS-2`） | | `msgx.pipeline.delivery.dead{target}` | 该投递目标当前死信（`DEAD`）行数 | 非零即告警：死信需人工处置（`OPS-2`；状态语义见 [implementation.md](implementation.md)「状态与错误分类」） | -| `msgx.pipeline.watermark.lag` | 信箱最新编号比已扫描编号大多少 | 改用处理时间筛选后删除（`G-SCAN-PREDICATE`） | 未处理消息的统计使用同一份缓存；统计不可用显示 `NaN`，没有记录可比时年龄与编号差显示 `-1`。作业计数在进程重启后归零；作业已启动却连续三个检查周期没有成功运行时，`/health` 报 `DOWN`。 diff --git a/docs/specification.md b/docs/specification.md index 72a152f..552dec6 100644 --- a/docs/specification.md +++ b/docs/specification.md @@ -159,7 +159,6 @@ | `G-REQ-TRACK` | 出站请求跟踪未做 | `US-09` | | `G-REQ-TRACK-RETENTION` | `REQ_TRACK` 已结案保留期未定（`Q24`） | `US-09` | | `G-RESP-GUARD` | `SCHD-RESP` 过期判断未做 | `US-07` | -| `G-SCAN-PREDICATE` | 按「处理时间为空」扫描未做 | `US-01`；`INV-1` | | `G-SCHD-SNAPSHOT` | 日计划快照删除、清空、分批未做 | `INV-7`、`INV-9` | | `G-SRVT-VIPF` | `SRVT`、`VIPF` 明细未入库（`Q2`） | `US-05` | 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 4656e4e..4183f57 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt @@ -27,15 +27,6 @@ class PipelineProps { var backoffMs: List = listOf(1000, 2000, 4000, 8000) var backoffCapMs: Long = 60_000 - /** - * 缺口等待时长：水位后面缺了一个 ID 时，等这么久还没出现就认定它永远不会来了， - * 跳过缺口继续推进水位。 - * - * 取值应该等于库方承诺的"上游提交到消息可见的最长时间"。设太小，可能把一条 - * 迟到的消息误判成永久缺失，导致它排到后面的消息之后；设太大，收报会在缺口上白等。 - */ - var maxCommitDelay: Duration = Duration.ofMinutes(5) - /** * 超期补写期限 R：一条消息到达终态后，过了这么久处理标记仍未写回信箱 * （比如回填一直失败），扫描谓词的超期分支成立，无视退避强制补写（只会提前、从不推迟打标）。 @@ -59,16 +50,6 @@ class PipelineProps { /** 回填独立退避的封顶间隔。 */ var backfillBackoffCapMs: Long = 900_000 - /** - * 切流水位播种（一次性、显式）。取值： - * `min` = `W:MIN(ID)−1`（读当前全部现存行）、`zero` = `W:0`（按空洞规则从 0 扫）、 - * `max` = `W:MAX(ID)`（跳过当前可见存量）、或一个具体 ID。 - * - * **默认 null = 不播种**，保持既有行为。是否跳过存量属于切流决策，必须由人显式配置： - * 代码不做默认选择，也不会自动退化成 `max`；升级实例（已有水位或已有处理记录）会拒绝重新播种。 - */ - var cutoverWatermark: String? = null - /** * 普通事件（`KAFKA:msg`）每轮向一个目标领取的条数上限。 * 逐条领取会让投递吞吐被"每条一次 DB 往返 + 一轮一次 sleep"压到每秒 1 条。 @@ -119,17 +100,6 @@ class PipelineProps { "msgx.pipeline.backoff-ms has ${backoffMs.size} slots, " + "but max-attempts=$maxAttempts implies exactly ${maxAttempts - 1}" } - // 切流播种只接受四种取值；非法值必须在启动时挡掉，而不是每轮轮询刷错误日志。 - val cutover = cutoverWatermark - require( - cutover == null || - cutover.equals("min", ignoreCase = true) || - cutover.equals("zero", ignoreCase = true) || - cutover.equals("max", ignoreCase = true) || - cutover.toLongOrNull() != null, - ) { - "msgx.pipeline.cutover-watermark must be one of min|zero|max|, got '$cutover'" - } } } 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 index 0b57dc8..22a83d9 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/InboxLifecycleHealthIndicator.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/InboxLifecycleHealthIndicator.kt @@ -1,8 +1,6 @@ package com.gzzn.omms.msgexchange.infra.health 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.ProcStateRepository import io.micronaut.context.BeanProvider import io.micronaut.core.async.publisher.Publishers @@ -19,8 +17,7 @@ import java.time.Instant * 在 /health 里输出收报与回填的当前情况，用来观察积压消化得怎么样： * - `backlog`：还没处理完的消息条数； * - `oldestUnprocessedSeconds`：最老一条未处理消息从收到到现在过了多久； - * - `unmarkedTerminal`：已经处理完、但还没把标记写回信箱的条数（回填跟不上时这个数会涨）； - * - `watermark` / `watermarkLag`：收报读到哪个 ID 了、落后信箱最新 ID 多少。 + * - `unmarkedTerminal`：已经处理完、但还没把标记写回信箱的条数（回填跟不上时这个数会涨）。 * * 相关工作没接上时（比如没连共享信箱）只提示"未绑定"，不判 DOWN——依赖本身是否可用 * 由各自的健康指示器回答，这里只报告业务状态。只有查询出错才判 DOWN。 @@ -28,8 +25,6 @@ import java.time.Instant @Singleton class InboxLifecycleHealthIndicator( private val procState: BeanProvider, - private val cursor: BeanProvider, - private val mailbox: BeanProvider, private val clock: Clock, private val backlogs: BacklogSnapshotProvider, ) : HealthIndicator { @@ -39,8 +34,6 @@ class InboxLifecycleHealthIndicator( 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, now = clock.instant(), // 与 /metrics 共用同一份 30 秒缓存：两者都不该把全表聚合打成高频查询。 backlog = backlogs.snapshot(), @@ -52,8 +45,6 @@ class InboxLifecycleHealthIndicator( /** 组装上面那几个指标；单独抽出来是为了能在测试里直接调用。 */ internal fun lifecycleHealth( procState: ProcStateRepository?, - cursor: InboxCursorRepository?, - mailbox: CminmsgInboxRepository?, now: Instant, /** 允许调用方传入缓存/已算好的积压快照；为空时现查（`backlog()` 是全表聚合）。 */ backlog: Backlog? = null, @@ -64,8 +55,6 @@ internal fun lifecycleHealth( .build() } val snapshot = 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 snapshot.unfinished, @@ -79,8 +68,6 @@ internal fun lifecycleHealth( "oldestUnmarkedBackfillSeconds" to ( snapshot.oldestUnmarkedAt?.let { Duration.between(it, now).seconds } ?: -1L ), - "watermark" to (watermark ?: -1L), - "watermarkLag" to if (watermark != null && maxId != null) maxId - watermark else -1L, ), ).build() } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetrics.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetrics.kt index 5659841..29452fb 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetrics.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetrics.kt @@ -2,9 +2,6 @@ package com.gzzn.omms.msgexchange.infra.metrics import com.gzzn.omms.msgexchange.infra.health.BacklogSnapshotProvider import com.gzzn.omms.msgexchange.infra.health.DeliveryDeadSnapshotProvider -import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository -import com.gzzn.omms.msgexchange.infra.persistence.InboxCursorRepository -import io.micronaut.context.BeanProvider import io.micronaut.context.annotation.Context import io.micronaut.context.annotation.Requires import io.micrometer.core.instrument.Gauge @@ -22,7 +19,6 @@ import java.time.Duration * - `msgx.pipeline.backfill.unmarked_terminal`：已终态但未打标的条数 * - `msgx.pipeline.backfill.abandoned`：已放弃自动回填的条数（**非 0 需人工对账**） * - `msgx.pipeline.backfill.oldest_unmarked_seconds`：最老一条仍待自动回填的年龄 - * - `msgx.pipeline.watermark.lag`：水位落后信箱最新 ID 的距离 * - `msgx.pipeline.job.heartbeat_age_seconds`：距上一次作业 tick 完成的秒数（未跑过为 -1） * - `msgx.pipeline.job.last_failure_age_seconds`：距最近一次作业 tick 失败的秒数（从未失败为 -1） * - `msgx.pipeline.job.ticks.total` / `msgx.pipeline.job.failures.total`：作业 tick 完成/抛错次数 @@ -48,8 +44,6 @@ class PipelineMetrics( private val registry: MeterRegistry, private val backlogs: BacklogSnapshotProvider, private val deadEvents: DeliveryDeadSnapshotProvider, - private val cursor: BeanProvider, - private val mailbox: BeanProvider, private val activity: JobActivity, private val clock: Clock, private val counters: PipelineCounters, @@ -71,12 +65,6 @@ class PipelineMetrics( ?: -1.0 } - Gauge.builder("msgx.pipeline.watermark.lag", backlogs) { _ -> - val watermark = if (cursor.isPresent) cursor.get().load().committedUpTo else null - val maxId = if (mailbox.isPresent) runCatching { mailbox.get().maxId() }.getOrNull() else null - if (watermark != null && maxId != null) (maxId - watermark).toDouble() else -1.0 - }.strongReference(true).register(registry) - // 作业心跳与扫描积压：作业线程是回填的唯一驱动，停摆必须能被 /metrics 与 /health 看见。 Gauge.builder("msgx.pipeline.job.heartbeat_age_seconds", activity) { a -> a.snapshot().lastTickAt?.let { Duration.between(it, clock.instant()).seconds.toDouble() } ?: -1.0 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 49b634d..45930a4 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 @@ -308,52 +308,13 @@ interface ReqTrackRepository { fun expire(reqId: Long) } -/** - * 收报进度（水位 W）：记下"信箱里到哪个 ID 为止已经全部读进自有库"，全表只有一行。 - * - * 光记一个数字不够，还要记住缺口是什么时候出现的：如果 W 后面缺了一个 ID，就先停在 - * 缺口前面等（可能是上游还没提交完，随时会补上）。等的时间超过最大提交时延，就改判为 - * 永久缺失、跳过去继续推进——否则一次自增回滚留下的空位就能让水位永远卡住， - * 它后面的消息再也进不了队。 - * - * 水位推进与入队在同一个 PG 事务里提交：中途崩溃时水位没动，重启后重扫一遍即可补齐。 - */ -interface InboxCursorRepository { - /** - * @param committedUpTo 水位 W - * @param holeSince W 后面那个缺口最早被发现的时刻；当前没有缺口时为 null - * @param seededAt 非空 = 已按 `msgx.pipeline.cutover-watermark` 播种过。 - * **它为空不等于"从未消费"**：已有库新增该列后同样是 null，判断必须叠加"水位为 0 且无处理记录"。 - */ - data class Cursor( - val committedUpTo: Long = 0L, - val holeSince: Instant? = null, - val seededAt: Instant? = null, - ) - - fun load(): Cursor - - /** 只推进水位与空洞计时，不改动 [Cursor.seededAt]。 */ - fun save(cursor: Cursor) - - /** - * 切流播种：一次性写入水位并把空洞计时清空，同时记录播种事实。 - * 与 [save] 分开，避免"普通轮次推进水位"把播种标记抹掉。 - */ - fun markSeeded(committedUpTo: Long, now: Instant) -} - /** * 共享 MySQL 信箱 CMINMSGS 的读写入口。这个库是别人的，本系统只做约定的读写， * 不建表、不改结构。 * - * 这里把三件事分得很清楚，谁也不代替谁： - * - **发现**：按 ID 区间读有哪些新消息（[readRange]）； - * - **进度**：读到哪儿了记在自有库的水位里（见 [InboxCursorRepository]）； - * - **标记**：处理完了把"已处理"写回信箱（[markProcessedIfUnmarked]）。 - * - * 特别是发现，不能拿"有没有处理标记"当筛选条件：处理完但还没回填的行，以及永远不会 - * 回填的死信，会一直占着每一批的名额，攒够一批之后新消息就再也读不到了。 + * 发现与标记是两件事：发现只看处理时间是否为空（[readUnprocessed]）， + * 标记只在处理结束后补写（[markProcessedIfUnmarked]）。消费边界不落库， + * 重复读到同一行由 PROC_STATE 主键挡掉（`INV-1`）。 */ interface CminmsgInboxRepository { /** 兼容入口往信箱写一条报文，返回新的信箱 ID。 */ @@ -365,14 +326,13 @@ interface CminmsgInboxRepository { /** 读取信箱库记录的实际接收时间。 */ fun receivedAtOf(msgId: Long): Instant? - /** 按 ID 升序读一批 `ID > fromExclusive` 的行，不带别的过滤条件。 */ - fun readRange(fromExclusive: Long, limit: Int): List - - /** 信箱当前最大 ID，空表返回 null；只用来观测收报落后了多少。 */ - fun maxId(): Long? - - /** 信箱当前最小 ID，空表返回 null；切流播种用它推算 `W = MIN(ID) − 1`。 */ - fun minId(): Long? + /** + * 按 ID 升序读一批处理时间为空、且 `ID > afterId` 的行。 + * + * [afterId] 只是同一轮扫描里的续读起点：已登记但还没回填的行会被反复读到， + * 若每轮都从头读同一批，它们攒够一批之后新行就再也读不到。 + */ + fun readUnprocessed(afterId: Long, limit: Int): List /** * 把处理标记写回信箱，并且**只写还是空标记的行**：库里已有值时不覆盖、不回退， 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 105b901..7fd15b5 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 @@ -14,8 +14,8 @@ import java.time.Instant * 只做增删改查，不建表、不改结构——这个库属于别的系统。 * * 两处刻意为之： - * - 取新消息只看 ID（`ID > ?`），不看 `DATE_PROCESSED`。用处理标记当条件的话， - * 处理完但还没回填的行会长期占住每批名额，死信攒够一批就再也发现不了新消息。 + * - 取新消息只看 `DATE_PROCESSED IS NULL`（`INV-1`）；`ID > ?` 只是同一轮扫描的续读起点， + * 让已登记却还没回填的行不长期占住每批名额。 * - 写回处理标记带 `DATE_PROCESSED IS NULL` 条件，一行只会被标记一次，不会覆盖已有值。 */ @Singleton @@ -57,16 +57,16 @@ class JdbcCminmsgInboxRepository( { ps -> ps.setLong(1, msgId) }, ) { rs -> rs.getTimestamp("CMINMSGS_DATE_RECEIVED")?.toInstant() } - override fun readRange(fromExclusive: Long, limit: Int): List = + override fun readUnprocessed(afterId: Long, limit: Int): List = ds.query( """ SELECT CMINMSGS_ID, CMINMSGS_DATE_RECEIVED FROM cminmsgs - WHERE CMINMSGS_ID > ? + WHERE CMINMSGS_DATE_PROCESSED IS NULL AND CMINMSGS_ID > ? ORDER BY CMINMSGS_ID ASC LIMIT ? """.trimIndent(), { ps -> - ps.setLong(1, fromExclusive) + ps.setLong(1, afterId) ps.setInt(2, limit) }, ) { rs -> @@ -76,16 +76,6 @@ class JdbcCminmsgInboxRepository( ) } - override fun maxId(): Long? = - ds.queryOne("SELECT MAX(CMINMSGS_ID) AS max_id FROM cminmsgs", {}) { rs -> - rs.getLong("max_id").takeIf { !rs.wasNull() } - } - - override fun minId(): Long? = - ds.queryOne("SELECT MIN(CMINMSGS_ID) AS min_id FROM cminmsgs", {}) { rs -> - rs.getLong("min_id").takeIf { !rs.wasNull() } - } - /** * 只更新还是空标记的行，所以重复调用不会覆盖库里已有的值； * 影响 0 行时再查一次主键，区分“已有标记”和“信箱行缺失”；后者不能记为回填成功。 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 d11c125..4e968e2 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 @@ -18,7 +18,6 @@ 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 @@ -37,7 +36,6 @@ import java.time.Instant import java.time.LocalDate import java.time.LocalDateTime import java.time.ZoneId -import java.util.concurrent.atomic.AtomicBoolean import com.gzzn.omms.msgexchange.domain.OperationDayCalculator import java.time.format.DateTimeFormatter import java.util.Locale @@ -362,90 +360,6 @@ class JdbcProcStateRepository( } } -/** 收报水位游标（单行）的 JDBC 实现；水位推进与入队在同一个事务里提交。 */ -@Singleton -@Requires(property = "datasources.default.enabled", value = "true") -@Requires(missingProperty = "msgx.stubs") -class JdbcInboxCursorRepository( - private val ds: DataSource, - private val clock: Clock, -) : InboxCursorRepository { - override fun load(): InboxCursorRepository.Cursor = - ds.queryOne( - "SELECT committed_up_to, hole_since, seeded_at FROM inbox_cursor WHERE cursor_id = 1", - {}, - ) { rs -> - InboxCursorRepository.Cursor( - rs.getLong("committed_up_to"), - rs.getInstant("hole_since"), - rs.getInstant("seeded_at"), - ) - } - ?: InboxCursorRepository.Cursor().also { - // 缺行不是"水位为 0"的同义词：这里只报一次，随后 save() 会 upsert 自愈。 - if (missingCursorWarned.compareAndSet(false, true)) { - log.error("INBOX_CURSOR row (cursor_id=1) is missing; watermark will be re-created on the next save()") - } - } - - /** - * 水位推进必须是 upsert，不能是裸 UPDATE： - * 裸 UPDATE 在游标行缺失时影响 0 行且不报错，会让水位永远停在初值、每轮重扫同一批， - * 收报静默死锁在第一批。upsert 让缺行自愈，且与入队同事务提交。 - */ - override fun save(cursor: InboxCursorRepository.Cursor) { - // 先 UPDATE、缺行再 INSERT：比 ON CONFLICT 更可移植（H2 的 PostgreSQL 兼容模式不支持 ON CONFLICT）， - // 且 InboxPoller 始终把它放在同一个 PG 事务里提交，因此两条语句对外仍是原子的。 - // 关键点是**缺行必须能自愈**：裸 UPDATE 影响 0 行却不报错，会让水位永远停在初值、每轮重扫同一批。 - val updated = 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, clock.instant().toSqlTimestamp()) - }, - ) - if (updated > 0) return - ds.update( - "INSERT INTO inbox_cursor (cursor_id, committed_up_to, hole_since, updated_at) VALUES (1, ?, ?, ?)", - { ps -> - ps.setLong(1, cursor.committedUpTo) - ps.setTimestamp(2, cursor.holeSince?.toSqlTimestamp()) - ps.setTimestamp(3, clock.instant().toSqlTimestamp()) - }, - ) - } - - /** - * 切流播种：一次性写入水位 + 清空空洞计时 + 记录播种事实（同一条语句）。 - * 与 [save] 分开，普通轮次推进水位不会把播种标记抹掉。 - */ - override fun markSeeded(committedUpTo: Long, now: Instant) { - val updated = ds.update( - "UPDATE inbox_cursor SET committed_up_to = ?, hole_since = NULL, seeded_at = ?, updated_at = ? " + - "WHERE cursor_id = 1", - { ps -> - ps.setLong(1, committedUpTo) - ps.setTimestamp(2, now.toSqlTimestamp()) - ps.setTimestamp(3, now.toSqlTimestamp()) - }, - ) - if (updated > 0) return - ds.update( - "INSERT INTO inbox_cursor (cursor_id, committed_up_to, hole_since, seeded_at, updated_at) " + - "VALUES (1, ?, NULL, ?, ?)", - { ps -> - ps.setLong(1, committedUpTo) - ps.setTimestamp(2, now.toSqlTimestamp()) - ps.setTimestamp(3, now.toSqlTimestamp()) - }, - ) - } - - private val missingCursorWarned = AtomicBoolean(false) - private val log = org.slf4j.LoggerFactory.getLogger(JdbcInboxCursorRepository::class.java) -} - @Singleton @Requires(property = "datasources.default.enabled", value = "true") @Requires(missingProperty = "msgx.stubs") 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 5820c77..2a897c7 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 @@ -15,7 +15,6 @@ import com.gzzn.omms.msgexchange.domain.flight.HistoryRules 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.MailboxMarkResult import com.gzzn.omms.msgexchange.infra.persistence.FlightStateRepository @@ -512,12 +511,9 @@ class StubInbox(private val clock: Clock = Clock.systemUTC()) : CminmsgInboxRepo override fun receivedAtOf(msgId: Long): Instant? = received[msgId] - override fun readRange(fromExclusive: Long, limit: Int): List = - raws.keys.filter { it > fromExclusive }.sorted().take(limit).map { MailboxRow(it, received[it]) } - - override fun maxId(): Long? = raws.keys.maxOrNull() - - override fun minId(): Long? = raws.keys.minOrNull() + override fun readUnprocessed(afterId: Long, limit: Int): List = + raws.keys.filter { it > afterId && !marks.containsKey(it) } + .sorted().take(limit).map { MailboxRow(it, received[it]) } /** 只写还没有标记的行，并区分已有标记与行缺失。 */ override fun markProcessedIfUnmarked(msgId: Long, value: String): MailboxMarkResult { @@ -546,28 +542,6 @@ class StubInbox(private val clock: Clock = Clock.systemUTC()) : CminmsgInboxRepo } } -/** 内存版收报水位游标。 */ -@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) { - // 普通轮次只推进水位与空洞计时；播种标记由 markSeeded 负责，不能被这里抹掉。 - this.cursor = this.cursor.copy(committedUpTo = cursor.committedUpTo, holeSince = cursor.holeSince) - } - - override fun markSeeded(committedUpTo: Long, now: Instant) { - cursor = InboxCursorRepository.Cursor(committedUpTo = committedUpTo, holeSince = null, seededAt = now) - } -} - /** 判断是不是终态：处理已经结束、不会再重试的状态。 */ 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/InboxPoller.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt index c714dd5..30fd75f 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt @@ -2,32 +2,26 @@ 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.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.Clock import java.time.Duration import java.time.Instant -import java.util.concurrent.atomic.AtomicBoolean /** - * 收报：轮询共享信箱，把新消息登记到自有 PG 的 PROC_STATE，等着主泵处理。 + * 收报：轮询共享信箱，把处理时间为空的行登记到自有 PG 的 PROC_STATE，等着主泵处理。 * - * 每一轮只做三件事： - * 1. 从"水位"之后按 ID 升序读一批行（只看 ID，不看处理标记）。 - * 水位记的是"信箱里到哪个 ID 为止已经全部读进自有库"，存在自己的库里，重启不丢； - * 2. 把读到的行登记成 PENDING，并把水位往前推； - * 3. 登记和水位推进写在同一个事务里——中途崩溃时水位没动，下一轮重扫即可补齐。 + * 每一轮做两件事： + * 1. 按处理时间为空、编号升序读一批，每批 `msgx.pipeline.claim-batch` 条； + * 2. 逐条登记成 PENDING，主键冲突即已登记过，跳过（`INV-1`）。 * - * 水位一次只推到"连续"的位置，这是这里唯一需要理解的规则。如果下一个 ID 缺号， - * 说明可能有 ID 更小的消息还没提交上来。此时先停在缺口前，不把缺口后面的消息放进队列： - * 否则那条迟到的消息会排到它们后面，破坏"先来先处理"的约定，同一航班的报文可能被乱序应用。 + * 读满一批就接着往后读。处理完却还没回填的行（尤其永远不回填的死信）会被反复读到， + * 只读第一批的话，这些行攒够一批之后新消息就再也发现不了。续读起点只活在本轮的内存里， + * 不落库、也不是消费边界：重启后从头重扫，重复登记建不出第二行。 * - * 缺口等超过 `msgx.pipeline.max-commit-delay` 仍未出现，就认定它永远不会来了 - * （典型情况是自增回滚留下的空位），跳过它继续推进——否则水位会卡在第一个空位上 - * 再也不动。 + * 编号较小的行晚提交时，最坏只是被发现得晚（下一轮仍会读到），不会丢，但它不会排在 + * 已经处理完的较大编号之前。 * * 这一层不解析报文，也不写信箱处理标记：标记由处理完成后的 BackfillService 负责补。 */ @@ -35,7 +29,6 @@ import java.util.concurrent.atomic.AtomicBoolean class InboxPoller( private val mailbox: CminmsgInboxRepository, private val procState: ProcStateRepository, - private val cursor: InboxCursorRepository, private val txManager: PipelineTransactionManager, private val props: PipelineProps, private val clock: Clock, @@ -45,93 +38,35 @@ class InboxPoller( @Volatile private var running = false - /** @return 本轮新登记的消息条数（已登记过的行不计入，也不影响水位推进）。 */ + /** @return 本轮新登记的消息条数（已登记过的行不计入）。 */ fun pollOnce(now: Instant): Int { - seedCutoverWatermarkIfConfigured(now) val batch = props.pipeline.claimBatch.coerceAtLeast(1) - val watermark = cursor.load() - val rows = mailbox.readRange(watermark.committedUpTo, batch) - if (rows.isEmpty()) return 0 - - // 找到连续部分的末尾；如果这批里出现了缺口，缺口后面的行这一轮先不入队 - val contiguous = contiguousUpTo(watermark.committedUpTo, rows) ?: watermark.committedUpTo - var committedTo = contiguous - var holeSince: Instant? = null - if (rows.last().msgId > contiguous) { - // 连续上界已前移说明旧空洞已补齐；后面是新空洞，不得继承旧等待时间。 - val since = if (contiguous == watermark.committedUpTo) watermark.holeSince ?: now else now - if (Duration.between(since, now) < props.pipeline.maxCommitDelay) { - holeSince = since - } else { - // 缺口等太久了：当成永久缺失跳过，让水位继续往前走 - committedTo = rows.first { it.msgId > contiguous }.msgId - 1 - log.warn("hole after W={} aged out, watermark advanced to {}", contiguous, committedTo) - } - } - - val enqueued = txManager.inTransaction { - var n = 0 - rows.takeWhile { it.msgId <= committedTo }.forEach { row -> + var afterId = 0L + var enqueued = 0 + var scanned = 0 + while (true) { + val rows = mailbox.readUnprocessed(afterId, batch) + if (rows.isEmpty()) break + enqueued += txManager.inTransaction { // 入队时间用本轮注入的本地时钟，与超期判据 R 同源。 - if (procState.insertIfAbsent(row.msgId, row.receivedAt, enqueuedAt = now)) n++ + rows.count { procState.insertIfAbsent(it.msgId, it.receivedAt, enqueuedAt = now) } } - cursor.save(InboxCursorRepository.Cursor(committedTo, holeSince)) - n + afterId = rows.last().msgId + scanned += rows.size + if (rows.size < batch) break } if (enqueued > 0) { - log.info("polled {} new messages, W {} -> {}", enqueued, watermark.committedUpTo, committedTo) + log.info("polled {} new messages out of {} unprocessed rows", enqueued, scanned) } return enqueued } - /** - * 切流水位播种：**只在显式配置 `msgx.pipeline.cutover-watermark` 时动作，且至多一次**。 - * - * 安全约束（评审要求，逐条对应）： - * - 默认不配置即不动作，代码不做默认选择、也不会自动退化成 `max`； - * - `SEEDED_AT` 为 NULL **不等于**"从未消费"（已有库新增列后同样是 NULL）， - * 因此必须叠加"水位为 0 且没有任何处理记录"这条判据； - * - 升级实例（已消费过）**拒绝重新播种**；重新切流必须是显式操作； - * - 播种事实与水位在同一条语句里落库，失败后下一轮重试不会留下部分状态。 - * - 信箱为空时无从界定边界，保持未播种（等有行时再判断）。 - */ - private fun seedCutoverWatermarkIfConfigured(now: Instant) { - val mode = props.pipeline.cutoverWatermark ?: return - val current = cursor.load() - if (current.seededAt != null) return - - if (current.committedUpTo > 0L || procState.hasAny()) { - if (refusedReseed.compareAndSet(false, true)) { - log.error( - "cutover-watermark={} is configured but this instance has already consumed messages " + - "(W={}); refusing to re-seed. Re-cutover must be an explicit operation.", - mode, current.committedUpTo, - ) - } - return - } - - val min = mailbox.minId() ?: return - val target = when { - mode.equals("min", ignoreCase = true) -> min - 1 - mode.equals("zero", ignoreCase = true) -> 0L - mode.equals("max", ignoreCase = true) -> mailbox.maxId() ?: return - else -> mode.toLongOrNull() ?: return // 非法值已在启动 validate() 挡掉 - } - - txManager.inTransaction { cursor.markSeeded(target, now) } - log.warn("cutover watermark seeded: mode={} W={} (mailbox min={})", mode, target, min) - } - - /** "拒绝重新播种"的告警只打一次，避免每轮刷屏。 */ - private val refusedReseed = AtomicBoolean(false) - fun loop() { running = true log.info("inbox poller loop started") while (running) { try { - // 时间一律走注入的 Clock：空洞老化阈值是"等多久算永久缺失"的唯一判据，不能依赖系统时钟。 + // 时间一律走注入的 Clock：入队时间是超期判据 R 的比较对象，不能依赖系统时钟。 pollOnce(clock.instant()) sleepQuietly(props.pipeline.pollInterval) } catch (_: InterruptedException) { @@ -149,21 +84,6 @@ class InboxPoller( running = false } - /** - * 从 W+1 开始数，返回 ID 逐 1 相连的最后一个 ID；遇到第一个缺号就停。 - * 返回 null 表示 W+1 本身就不存在。 - */ - private fun contiguousUpTo(from: Long, rows: List): Long? { - var expected = from + 1 - var last: Long? = null - for ((msgId, _) in rows) { - if (msgId != expected) break - last = msgId - expected++ - } - return last - } - private fun sleepQuietly(d: Duration) { try { Thread.sleep(d.toMillis().coerceAtLeast(1)) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt index 399e87b..b797260 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt @@ -14,7 +14,6 @@ import com.gzzn.omms.msgexchange.domain.ProcStatus import com.gzzn.omms.msgexchange.infra.log.TraceLog import com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters 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 com.gzzn.omms.msgexchange.infra.retry.ProcFailure import jakarta.inject.Singleton @@ -38,7 +37,6 @@ import java.util.concurrent.atomic.AtomicLong @Singleton class Pump( private val procState: ProcStateRepository, - private val cursor: InboxCursorRepository, private val processor: MessageProcessor, private val props: PipelineProps, private val clock: Clock, @@ -80,17 +78,6 @@ class Pump( sleepQuietly(props.pipeline.pollInterval) return } - // 只领取"已被水位覆盖"的队头（`msgId <= W`）。 - // - // 水位以内的行都是收报按 ID 顺序发现并登记的；水位之外的行只可能来自兼容入口 - // 直接写 PROC_STATE（它不参与水位）。若允许领取，它就会越过那些尚未入队的较小 ID， - // 破坏 FIFO（不变量"只领取已发现的行"，`specification.md` `INV-4`）。这种行在空洞补齐、`W` 追平之后自然可领取。 - val watermark = cursor.load().committedUpTo - if (head.msgId > watermark) { - warnBeyondWatermark(head.msgId, watermark) - sleepQuietly(props.pipeline.pollInterval) - return - } val now = clock.instant() when { head.state == ProcStatus.FAILED && head.attempts >= props.pipeline.maxAttempts -> { @@ -111,19 +98,6 @@ class Pump( } } - /** 上一次"队头在水位之外"告警时的水位值：只在它变化时告警，避免每秒刷屏。 */ - private val warnedWatermark = AtomicLong(Long.MIN_VALUE) - - private fun warnBeyondWatermark(msgId: Long, watermark: Long) { - if (warnedWatermark.getAndSet(watermark) != watermark) { - log.warn( - "head msgId={} is beyond watermark W={}; waiting for discovery " + - "(row injected by the compat entry point?)", - msgId, watermark, - ) - } - } - private fun sleepQuietly(d: Duration) { if (!d.isNegative && !d.isZero) Thread.sleep(d.toMillis().coerceAtLeast(1)) } diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 80a3881..50beb20 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -18,12 +18,9 @@ msgx: max-attempts: 5 # 处理/投递同值 backoff-ms: [1000, 2000, 4000, 8000] # 指数退避；档位数必须 = max-attempts - 1（启动自检） backoff-cap-ms: 60000 - max-commit-delay: 5m # 空洞老化：W+1 空洞超过该时延判定为永久（Q7 最大提交时延） overdue-backfill: 30d # 超期补写期限 R：仅须 R ≤ R_keep；判据比较本地 ENQUEUED_AT（implementation.md「回填」） backfill-batch: 100 # 回填扫描单批条数 backfill-max-attempts: 100 # 单行重试的告警阈值；放弃判据是 R 超期，不是次数（implementation.md「回填」） - # 一次性切流播种：默认（注释掉）不播种。min=读现存全部 | zero=从 0 按空洞规则 | max=跳过可见存量 | - # cutover-watermark: min autostart: false # U07：启动即拉起 Pump/Dispatcher 循环；需真实仓储或 msgx.stubs=true 才开启（dev 见 application-dev.yml） schd: flush-period: 3s # KEEP 现役推送节律 @@ -46,7 +43,7 @@ micronaut: port: 8080 # 管理端点（U03/N32）：/env、/beans 默认 sensitive；仅开发/影子环境放开——见 application-dev.yml -# 自有 PostgreSQL：全部内部状态（PIPELINE_LOCK/PROC_STATE/MSG_EVENT/REQ_TRACK/INBOX_CURSOR/ +# 自有 PostgreSQL：全部内部状态（PIPELINE_LOCK/PROC_STATE/MSG_EVENT/REQ_TRACK/ # FLIGHT_SCHD 及明细表/SCHD_SNAP_LOG；迁移链：V1__flight_state_baseline.sql（原多版净结构合并） # + V2__flight_chute_class_type_rename.sql（chute 类字段 CCLS/CTYP）；回填事实与处理终态同表同行）。 # 数据层实装前 enabled=false（stub 模式不建连）。 diff --git a/src/main/resources/db/migration/V1__flight_state_baseline.sql b/src/main/resources/db/migration/V1__flight_state_baseline.sql index a6c8671..d8a099d 100644 --- a/src/main/resources/db/migration/V1__flight_state_baseline.sql +++ b/src/main/resources/db/migration/V1__flight_state_baseline.sql @@ -5,7 +5,6 @@ -- 系统库，只做契约内 DML，不建表、不改结构（C-2）。 -- -- 本文件原为 V1 基线；原 V2–V10 的净结构已全部合并进来，版本链收敛为一条： --- · INBOX_CURSOR 及 SEEDED_AT（原 V2/V5）； -- · PROC_STATE 的收信/入队时间与回填事实列，独立待办表 BACKFILL_TODO 不再存在 -- （原 V2/V3/V4/V6/V7：回填意图并入 PROC_STATE，处理开始时间列加后即删）； -- · REQ_TRACK 开放态部分唯一索引 uq_req_open（原 V8）； @@ -20,7 +19,7 @@ -- -- 分层（职责与写者见 docs/implementation.md「航班域」）： -- · 决策层：FLIGHT_SCHD + 8 张资源明细表 + FLIGHT_ROUTE_POINT —— 权威当前态（INV-11）； --- · 管道层：PIPELINE_LOCK / INBOX_CURSOR / PROC_STATE / MSG_EVENT / REQ_TRACK； +-- · 管道层：PIPELINE_LOCK / PROC_STATE / MSG_EVENT / REQ_TRACK； -- · 留痕层：SCHD_SNAP_LOG —— 只追加、可重建、不参与决策； -- · 证据层：报文原文归档 —— 冷路径，尚未交付（ARCHIVE_KEY 仅留引用位）。 -- 类型口径：TIMESTAMP(6) WITH TIME ZONE 统一 UTC 语义；BIGSERIAL 为 PostgreSQL 方言 @@ -65,22 +64,7 @@ CREATE INDEX idx_proc_head ON PROC_STATE (STATE, MSG_ID); -- 主泵队头查询 CREATE INDEX idx_proc_backfill_due ON PROC_STATE (BACKFILL_ATTEMPTS, MSG_ID) WHERE BACKFILL_AT IS NULL AND BACKFILL_ABANDONED_AT IS NULL; --- ③ 消费水位：全表只有一行。W 只随新 ID 成功入队推进、只增不减，遇空洞即停； --- 入队与水位推进同事务（INV-2），因此不存在「水位已推进、消息未入队」的持久化状态。 --- HOLE_SINCE 持久化空洞观测时刻，进程重启不丢计时；老化阈值取 --- PARAM:msgx.pipeline.max-commit-delay，超期只放行空洞本身、不越过任何已存在的行。 -CREATE TABLE INBOX_CURSOR ( - CURSOR_ID INT NOT NULL PRIMARY KEY, -- 固定为 1 - COMMITTED_UP_TO BIGINT NOT NULL, -- 水位 W：到哪个 ID 为止已全部读进自有库 - HOLE_SINCE TIMESTAMP(6) WITH TIME ZONE, -- 后续缺号最早被发现的时间；不缺号时为 NULL - UPDATED_AT TIMESTAMP(6) WITH TIME ZONE NOT NULL, - SEEDED_AT TIMESTAMP(6) WITH TIME ZONE -- 记录「已按 PARAM:msgx.pipeline.cutover-watermark 播种」这一事实； - -- 为 NULL **不等于**从未消费——已有库新增列后同样为 NULL -); --- 初值 W=0：首轮把信箱现存行全部重新读一遍，重复登记不会建出第二行。 -INSERT INTO INBOX_CURSOR (CURSOR_ID, COMMITTED_UP_TO, HOLE_SINCE, UPDATED_AT) VALUES (1, 0, NULL, now()); - --- ④ 航班当前态主行：一行一 FLID（决策层权威；展示视图是投影，不是权威，INV-5） +-- ③ 航班当前态主行：一行一 FLID（决策层权威；展示视图是投影，不是权威，INV-5） CREATE TABLE FLIGHT_SCHD ( FLID VARCHAR(32) NOT NULL PRIMARY KEY, -- AODB 实例 ID（Number(1-12)）；不得由航班号或资源号推断 OPERATION_DAY DATE NULL, -- 运营保障日：未由日计划收录时为 NULL；非空后不可改变（implementation「航班身份与运营日」） @@ -140,11 +124,11 @@ CREATE INDEX idx_flight_schd_state ON FLIGHT_SCHD (STATE); -- WHERE OPERATION_DAY IS NULL OR OPERATION_DAY = :day，行级条件更新即满足； -- 不使用触发器（Oracle/PG 双方言成本）。 --- ⑤ 资源明细表 ×8 + 路线点表：每 FLID 多行；每次完整状态写入按该 FLID 先删后插， +-- ④ 资源明细表 ×8 + 路线点表：每 FLID 多行；每次完整状态写入按该 FLID 先删后插， -- 以完整合并结果为准（implementation「字段与集合」）。ORDINAL 是持久化顺序（从 1 起），SOURCE_SEQ 是上游 -- 序号（允许为空或重复）；相同资源号不代表同一条分配，禁止按资源号去重。删除一律 -- 标记在主行 STATE，明细物理清除只发生在受控历史归档（US-14）。 --- ⑤-1 登机门 GTDT +-- ④-1 登机门 GTDT CREATE TABLE FLIGHT_GATE ( FLID VARCHAR(32) NOT NULL, ORDINAL INT NOT NULL, @@ -159,7 +143,7 @@ CREATE TABLE FLIGHT_GATE ( UPDATED_AT TIMESTAMP(6) WITH TIME ZONE NOT NULL, PRIMARY KEY (FLID, ORDINAL) ); --- ⑤-2 值机柜台 CKDT +-- ④-2 值机柜台 CKDT CREATE TABLE FLIGHT_CHECKIN ( FLID VARCHAR(32) NOT NULL, ORDINAL INT NOT NULL, @@ -175,7 +159,7 @@ CREATE TABLE FLIGHT_CHECKIN ( UPDATED_AT TIMESTAMP(6) WITH TIME ZONE NOT NULL, PRIMARY KEY (FLID, ORDINAL) ); --- ⑤-3 行李转盘 CLDT +-- ④-3 行李转盘 CLDT CREATE TABLE FLIGHT_BELT ( FLID VARCHAR(32) NOT NULL, ORDINAL INT NOT NULL, @@ -191,7 +175,7 @@ CREATE TABLE FLIGHT_BELT ( UPDATED_AT TIMESTAMP(6) WITH TIME ZONE NOT NULL, PRIMARY KEY (FLID, ORDINAL) ); --- ⑤-4 计划机位 PSDT +-- ④-4 计划机位 PSDT CREATE TABLE FLIGHT_STAND_PLAN ( FLID VARCHAR(32) NOT NULL, ORDINAL INT NOT NULL, @@ -203,7 +187,7 @@ CREATE TABLE FLIGHT_STAND_PLAN ( UPDATED_AT TIMESTAMP(6) WITH TIME ZONE NOT NULL, PRIMARY KEY (FLID, ORDINAL) ); --- ⑤-5 滑槽 CHDT +-- ④-5 滑槽 CHDT CREATE TABLE FLIGHT_CHUTE ( FLID VARCHAR(32) NOT NULL, ORDINAL INT NOT NULL, @@ -219,7 +203,7 @@ CREATE TABLE FLIGHT_CHUTE ( UPDATED_AT TIMESTAMP(6) WITH TIME ZONE NOT NULL, PRIMARY KEY (FLID, ORDINAL) ); --- ⑤-6 延误 DELY（业务上任意时刻仅 1 个有效延误，保留多行能力以无损承接） +-- ④-6 延误 DELY（业务上任意时刻仅 1 个有效延误，保留多行能力以无损承接） CREATE TABLE FLIGHT_DELAY ( FLID VARCHAR(32) NOT NULL, ORDINAL INT NOT NULL, @@ -232,7 +216,7 @@ CREATE TABLE FLIGHT_DELAY ( UPDATED_AT TIMESTAMP(6) WITH TIME ZONE NOT NULL, PRIMARY KEY (FLID, ORDINAL) ); --- ⑤-7 靠撤桥 ABTM（桥号 ABDG；A/D 各一条） +-- ④-7 靠撤桥 ABTM（桥号 ABDG；A/D 各一条） CREATE TABLE FLIGHT_BRIDGE_OP ( FLID VARCHAR(32) NOT NULL, ORDINAL INT NOT NULL, @@ -244,7 +228,7 @@ CREATE TABLE FLIGHT_BRIDGE_OP ( UPDATED_AT TIMESTAMP(6) WITH TIME ZONE NOT NULL, PRIMARY KEY (FLID, ORDINAL) ); --- ⑤-8 轮挡 CHOT（机位 CHID；ON/OFF 各一条） +-- ④-8 轮挡 CHOT（机位 CHID；ON/OFF 各一条） CREATE TABLE FLIGHT_CHOCK_OP ( FLID VARCHAR(32) NOT NULL, ORDINAL INT NOT NULL, @@ -256,7 +240,7 @@ CREATE TABLE FLIGHT_CHOCK_OP ( UPDATED_AT TIMESTAMP(6) WITH TIME ZONE NOT NULL, PRIMARY KEY (FLID, ORDINAL) ); --- ⑤-9 路线点 ROUT/ERUT 共用（ROUTE_KIND 区分两类；主键含该列，避免序号冲突） +-- ④-9 路线点 ROUT/ERUT 共用（ROUTE_KIND 区分两类；主键含该列，避免序号冲突） CREATE TABLE FLIGHT_ROUTE_POINT ( FLID VARCHAR(32) NOT NULL, ORDINAL INT NOT NULL, @@ -278,7 +262,7 @@ CREATE INDEX idx_flight_delay_flid ON FLIGHT_DELAY (FLID); CREATE INDEX idx_flight_bridge_flid ON FLIGHT_BRIDGE_OP (FLID); CREATE INDEX idx_flight_chock_flid ON FLIGHT_CHOCK_OP (FLID); --- ⑥ 统一投递事件 outbox（C-9、INV-3）：KAFKA:schd 发整态、KAFKA:msg 只通知变化； +-- ⑤ 统一投递事件 outbox（C-9、INV-3）：KAFKA:schd 发整态、KAFKA:msg 只通知变化； -- tombstone 仅在 ACTIVE→DELETED 时与删除同事务登记，投递失败按退避持续重试。 -- 目标级全序投递是当前实现，按 FLID 保序见 CLM-3。 CREATE TABLE MSG_EVENT ( @@ -302,7 +286,7 @@ CREATE UNIQUE INDEX uq_schd_event ON MSG_EVENT (TARGET, PARTITION_KEY) WHERE TARGET = 'KAFKA:schd'; --- ⑦ 请求状态机：只有 RESP 完成 RQFD 请求，按（运营日、发送方、请求类型）匹配最新一条 +-- ⑥ 请求状态机：只有 RESP 完成 RQFD 请求，按（运营日、发送方、请求类型）匹配最新一条 -- 开放请求。同类请求只留一条有效，新请求置旧请求为 EXPIRED。登记、超时与应答匹配 -- 尚未实现（G-REQ-TRACK、G-REQ-OPEN-UNIQUE）。 CREATE TABLE REQ_TRACK ( @@ -322,7 +306,7 @@ CREATE UNIQUE INDEX uq_req_open ON REQ_TRACK (REQ_TYPE, OPERATION_DAY, SENDER) WHERE STATE IN ('PENDING', 'SENT'); --- ⑧ SCHD 快照留痕：事务外追加，只追加留痕、不参与决策；一行=一次尝试，重放也记。 +-- ⑦ SCHD 快照留痕：事务外追加，只追加留痕、不参与决策；一行=一次尝试，重放也记。 -- RESULT 与 FLAGS 分列（可「成功且告警」）；写失败只记指标。 CREATE TABLE SCHD_SNAP_LOG ( LOG_ID BIGSERIAL PRIMARY KEY, diff --git a/src/main/resources/db/migration/oracle11g/README.md b/src/main/resources/db/migration/oracle11g/README.md index 486c877..6998b4d 100644 --- a/src/main/resources/db/migration/oracle11g/README.md +++ b/src/main/resources/db/migration/oracle11g/README.md @@ -16,7 +16,7 @@ Flyway 配置**；PG 路径使用 `classpath:db/migration`，两者互不混用 ## 计划内容 - 单基线：`V1__flight_state_baseline.sql` 的 11g 等价 DDL—— - PIPELINE_LOCK/INBOX_CURSOR/PROC_STATE/MSG_EVENT/REQ_TRACK/FLIGHT_SCHD + 9 张明细表/SCHD_SNAP_LOG； + PIPELINE_LOCK/PROC_STATE/MSG_EVENT/REQ_TRACK/FLIGHT_SCHD + 9 张明细表/SCHD_SNAP_LOG； `NUMBER`/序列替代 `BIGSERIAL`、`TIMESTAMP WITH TIME ZONE`、`VARCHAR2` BYTE/CHAR 语义钉死。 - `INSERT ... ON CONFLICT` 改 11g MERGE（OPERATION_DAY 不可变条件，implementation「航班身份与运营日」）。 - 空串按 NULL 的语义回归：显式清空的 presence 信息不得被 11g 空串语义吞掉 diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt index 60c87af..c78e964 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt @@ -52,7 +52,6 @@ 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() } @@ -98,8 +97,6 @@ class PipelineSmokeTest { val receipt = controller.send(UNSUPPORTED_XML) assertNotNull(receipt.body()) // 受理 ID val id = receipt.body()!!.toLong() - // 兼容入口只保证"已落信 + 已入队"，**不推进水位**；必须先被收报发现（W 追平）才可领取。 - ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce(Instant.now()) pump.tick() // 解码成功但无 XYZQ Handler → SKIPPED(unsupported) @@ -115,7 +112,6 @@ class PipelineSmokeTest { fun `replay reopens failed UNSUPPORTED row to PENDING`() { val receipt = controller.send(REFDATA_XML) val id = receipt.body()!!.toLong() - ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce(Instant.now()) pump.tick() val stub = ctx.getBean(StubProcState::class.java) assertEquals(ProcStatus.FAILED, stub.snapshotOf(id)!!.state) @@ -193,9 +189,6 @@ class PipelineSmokeTest { val proc = ctx.getBean(StubProcState::class.java) val props = ctx.getBean(com.gzzn.omms.msgexchange.config.PipelineProps::class.java) val backfill = ctx.getBean(com.gzzn.omms.msgexchange.processing.BackfillService::class.java) - // 队头必须在水位以内才可领取：直接播种 PROC_STATE 的用例要显式把水位推上去。 - ctx.getBean(com.gzzn.omms.msgexchange.infra.stub.StubInboxCursor::class.java) - .save(com.gzzn.omms.msgexchange.infra.persistence.InboxCursorRepository.Cursor(committedUpTo = 9001L)) proc.insertIfAbsent(9001L, Instant.now()) proc.update( 9001L, ProcStatus.FAILED, @@ -225,40 +218,32 @@ class PipelineSmokeTest { } /** - * FIFO 不变量"只领取已发现的行"端到端验收：兼容入口写入的高 ID **不会被提前领取**， - * 必须等水位追平（较小 ID 补齐并入队）之后才按顺序处理。 + * 扫描谓词端到端验收（`INV-1`）：处理标记为空的行都会被发现并登记，处理完并回填之后 + * 退出扫描谓词；兼容入口已登记的行不会被收报建成第二条记录。 */ @Test - fun `compat injected high id is not claimed until the watermark catches up`() { + fun `scan predicate discovers every unmarked row exactly once`() { val inbox = ctx.getBean(com.gzzn.omms.msgexchange.infra.stub.StubInbox::class.java) val proc = ctx.getBean(StubProcState::class.java) - val cursor = ctx.getBean(com.gzzn.omms.msgexchange.infra.stub.StubInboxCursor::class.java) val poller = ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java) - // 信箱：1 存在、2 是空洞（被删）、3 存在 → 水位停在 1 - inbox.simulateExternalWrite("") - val hole = inbox.simulateExternalWrite("") - inbox.simulateExternalWrite("") - inbox.removeRow(hole) + val external = inbox.simulateExternalWrite("") // 上游直接写信箱 + val compat = controller.send(UNSUPPORTED_XML).body()!!.toLong() // 兼容入口：已自行入队 + poller.pollOnce(Instant.now()) - assertEquals(1L, cursor.cursor.committedUpTo) + assertEquals(setOf(external, compat), proc.rows.keys.toSet()) - // 兼容入口写入高 ID：直接进 PG，但水位没追平（这正是原来的越序路径） - val high = controller.send(UNSUPPORTED_XML).body()!!.toLong() - assertTrue(high > 3L) + poller.pollOnce(Instant.now()) // 重扫不建第二条 + assertEquals(2, proc.rows.size) - pump.tick() // 先按 FIFO 处理 ID=1 - assertEquals(ProcStatus.DEAD, proc.find(1L)!!.state) - pump.tick() // 队头变成高 ID，但它在水位之外 → 不领取 - assertEquals(ProcStatus.PENDING, proc.find(high)!!.state) + repeat(2) { pump.tick() } + ctx.getBean(com.gzzn.omms.msgexchange.processing.BackfillService::class.java).sweep() + assertTrue(inbox.isMarked(external) && inbox.isMarked(compat)) - // 空洞补齐 → 水位追平 → 才允许继续按顺序处理 - inbox.restoreRow(hole, "", Instant.now()) + // 已回填的行退出扫描谓词：清掉处理记录后也不会被重新登记 + proc.clear() poller.pollOnce(Instant.now()) - assertTrue(cursor.cursor.committedUpTo >= high) - repeat(3) { pump.tick() } // 依次处理 2、3、高 ID - assertEquals(ProcStatus.SKIPPED, proc.find(high)!!.state) - assertEquals("unsupported:XYZQ-FOO", proc.find(high)!!.lastError) + assertTrue(proc.rows.isEmpty()) } @Test diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/config/PipelineConfigCheckTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/config/PipelineConfigCheckTest.kt index c2042b3..ead8b02 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/config/PipelineConfigCheckTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/config/PipelineConfigCheckTest.kt @@ -23,13 +23,4 @@ class PipelineConfigCheckTest { assertThrows(IllegalArgumentException::class.java) { PipelineConfigCheck(props) } } - - @Test - fun `illegal cutover watermark rejects startup even without autostart`() { - val props = PipelineProps().apply { - pipeline.cutoverWatermark = "yesterday" - } - - assertThrows(IllegalArgumentException::class.java) { PipelineConfigCheck(props) } - } } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/config/PipelinePropsTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/config/PipelinePropsTest.kt index c6977d0..214d721 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/config/PipelinePropsTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/config/PipelinePropsTest.kt @@ -33,19 +33,6 @@ class PipelinePropsTest { assertThrows(IllegalArgumentException::class.java) { pipeline.validate() } } - /** 切流播种只接受 min|zero|max|：非法值必须在启动时挡掉。 */ - @Test - fun `validate rejects an unknown cutover watermark mode`() { - pipeline.cutoverWatermark = "bogus" - assertThrows(IllegalArgumentException::class.java) { pipeline.validate() } - - pipeline.cutoverWatermark = "max" - pipeline.validate() - - pipeline.cutoverWatermark = "12345" - pipeline.validate() - } - @Test fun `non-positive attempt never throws and falls back to first slot`() { assertEquals(1000, pipeline.backoffFor(0)) 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 3708056..85d147a 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 @@ -3,9 +3,6 @@ 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 @@ -41,18 +38,14 @@ class HealthIndicatorsTest { /** 检查 /health 里那几个积压指标算得对不对。 */ @Test - fun `inbox lifecycle reports backlog, oldest age, unmarked terminals and watermark lag`() { + fun `inbox lifecycle reports backlog, oldest age and unmarked terminals`() { 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, now = MutableClock.BASE) - val newest = inbox.insertRaw("") - cursor.save(InboxCursorRepository.Cursor(committedUpTo = newest - 2)) - val result = lifecycleHealth(proc, cursor, inbox, now = MutableClock.BASE) + val result = lifecycleHealth(proc, now = MutableClock.BASE) assertEquals(HealthStatus.UP, result.status) @Suppress("UNCHECKED_CAST") @@ -60,12 +53,11 @@ class HealthIndicatorsTest { 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, now = MutableClock.BASE) + val result = lifecycleHealth(procState = null, now = MutableClock.BASE) assertEquals(HealthStatus.UP, result.status) // 可用性由依赖自身指示器承担 } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetricsTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetricsTest.kt index e16b2a2..f359d38 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetricsTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetricsTest.kt @@ -48,8 +48,6 @@ class PipelineMetricsTest { assertEquals(1.0, gauge("msgx.pipeline.backlog.unfinished"), 0.001) assertEquals(1.0, gauge("msgx.pipeline.backfill.unmarked_terminal"), 0.001) assertEquals(0.0, gauge("msgx.pipeline.backfill.abandoned"), 0.001) - // 信箱为空 → 没有"最新 ID"可比，滞后用 -1（无值）而不是伪造 0 - assertEquals(-1.0, gauge("msgx.pipeline.watermark.lag"), 0.001) } @Test 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 be8f116..203d234 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 @@ -11,8 +11,8 @@ import java.sql.DriverManager /** * 在真实 PostgreSQL 上跑一遍迁移链，确认结果符合预期：`V1__flight_state_baseline.sql` - * 基线加 `V2__flight_chute_class_type_rename.sql` 更名依序执行成功，该建的表和单行种子 - * （PIPELINE_LOCK、INBOX_CURSOR）都在，回填事实与收报水位都落在基线里，`BACKFILL_TODO`、 + * 基线加 `V2__flight_chute_class_type_rename.sql` 更名依序执行成功，该建的表和 PIPELINE_LOCK + * 单行种子都在，回填事实落在基线里，`INBOX_CURSOR`、`BACKFILL_TODO`、 * `idx_evt_flid`、`PROC_STATE` 的处理开始时间列都不复存在；FLIGHT_CHUTE 的类字段列 * 已由 V2 更名为 CCLS/CTYP（SIS 口径）。 * @@ -24,7 +24,7 @@ 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", "inbox_cursor", "schd_snap_log", + "msg_event", "req_track", "schd_snap_log", ) @Test @@ -78,6 +78,7 @@ class FlywayMigrationTest { val missing = expectedTables - tables assertTrue(missing.isEmpty(), "missing tables: $missing") assertTrue("backfill_todo" !in tables, "BACKFILL_TODO 已由 PROC_STATE 回填列取代") + assertTrue("inbox_cursor" !in tables, "扫描谓词按处理时间取行，不再有消费水位表") } // 身份不变量基础结构：FLIGHT_SCHD 主键 + STATE 列 @@ -200,17 +201,11 @@ class FlywayMigrationTest { assertEquals("NO", rs.getString("is_nullable"), "ENQUEUED_AT 必须非空") } - // 两张单行表（管道锁、收报水位）的种子数据都要在 + // 管道锁的单行种子要在 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 index 256211e..bc48afa 100644 --- 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 @@ -2,7 +2,6 @@ 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 com.gzzn.omms.msgexchange.infra.persistence.MailboxMarkResult import org.h2.jdbcx.JdbcDataSource import org.junit.jupiter.api.Assertions.assertEquals @@ -24,7 +23,7 @@ import javax.sql.DataSource * * 库用 H2 的 PostgreSQL 兼容模式，表结构照抄 V1 + V2，因此不需要 docker 或外接数据库 * 就能跑。覆盖两边的真实语句：自有 PG 侧（写终态时一并写下回填待办、挑选待回填记录、 - * 积压统计、水位游标读写）和共享信箱侧（按 ID 区间读、标记只写一次）。 + * 积压统计）和共享信箱侧（按处理时间为空扫描、标记只写一次）。 * * 有一处覆盖不到：H2 不支持 `INSERT ... ON CONFLICT DO NOTHING`，所以入队幂等没在这里验证， * 由 `InboxPollerTest`（重复轮询不再登记）和 PostgreSQL 本身的语义来保证。 @@ -34,7 +33,6 @@ 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)) @@ -49,18 +47,12 @@ class InboxLifecycleJdbcSqlTest { ds.connection.use { conn -> conn.createStatement().use { st -> st.execute(PROC_STATE_DDL) - st.execute(CURSOR_DDL) st.execute(MAILBOX_DDL) st.execute(EVENT_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, java.time.Clock.systemUTC()) mailbox = JdbcCminmsgInboxRepository(ds) - cursor = JdbcInboxCursorRepository(ds, java.time.Clock.systemUTC()) } @Test @@ -133,15 +125,6 @@ class InboxLifecycleJdbcSqlTest { 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 `replay resets attempts, backoff and the previous error reason`() { seed(12L, t0) @@ -185,22 +168,22 @@ class InboxLifecycleJdbcSqlTest { } @Test - fun `mailbox range read ignores processing marks and marking never overwrites`() { + fun `mailbox scan takes unprocessed rows in id order 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()) + // 扫描谓词只看处理时间为空；afterId 是同一轮的续读起点 + assertEquals(listOf(first, second, third), mailbox.readUnprocessed(0L, 50).map { it.msgId }) + assertNotNull(mailbox.readUnprocessed(0L, 50).first().receivedAt) + assertEquals(listOf(second, third), mailbox.readUnprocessed(first, 50).map { it.msgId }) + assertEquals(listOf(first, second), mailbox.readUnprocessed(0L, 2).map { it.msgId }) assertEquals(MailboxMarkResult.MARKED, mailbox.markProcessedIfUnmarked(second, "PROCESSED")) assertEquals(MailboxMarkResult.ALREADY_MARKED, mailbox.markProcessedIfUnmarked(second, "OTHER")) assertEquals("PROCESSED", statusOf(second)) - // 已标记行仍出现在区间读结果中（发现与标记彻底解耦） - assertEquals(listOf(first, second, third), mailbox.readRange(0L, 50).map { it.msgId }) + // 打过标记的行退出扫描谓词 + assertEquals(listOf(first, third), mailbox.readUnprocessed(0L, 50).map { it.msgId }) assertEquals(MailboxMarkResult.MISSING, mailbox.markProcessedIfUnmarked(999L, "PROCESSED")) } @@ -252,16 +235,6 @@ class InboxLifecycleJdbcSqlTest { ) """ - 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, - seeded_at 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, diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/CutoverSeedTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/CutoverSeedTest.kt deleted file mode 100644 index 9a94120..0000000 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/CutoverSeedTest.kt +++ /dev/null @@ -1,164 +0,0 @@ -package com.gzzn.omms.msgexchange.ingress - -import com.gzzn.omms.msgexchange.config.PipelineProps -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.StubPipelineTx -import com.gzzn.omms.msgexchange.infra.stub.StubProcState -import org.junit.jupiter.api.Assertions.assertEquals -import org.junit.jupiter.api.Assertions.assertNotEquals -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.time.Clock -import java.time.Instant -import java.time.ZoneOffset - -/** - * 切流水位播种（ACM2-35）：**显式、一次性、升级安全**。 - * - * 覆盖评审提出的四条硬要求： - * 1. 默认（未配置）不动作，代码不做默认选择； - * 2. 四种模式严格区分（`min` 读现存全部 / `zero` 从 0 按空洞规则 / `max` 跳过可见存量 / 显式 ID）； - * 3. 升级实例（已有水位或已有处理记录）**拒绝重新播种**——`SEEDED_AT` 为 NULL 不等于"从未消费"； - * 4. 播种至多一次，且不会被普通的水位推进抹掉。 - */ -class CutoverSeedTest { - - 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 setUp() { - inbox = StubInbox().apply { clear() } - proc = StubProcState().apply { clear() } - cursor = StubInboxCursor().apply { clear() } - props.pipeline.cutoverWatermark = null - poller = InboxPoller( - inbox, proc, cursor, StubPipelineTx(), props, - Clock.fixed(t0, ZoneOffset.UTC), - ) - } - - /** 造一个 MIN(ID)=5 的信箱：1..4 已被库方清除（典型"最老分区已 DROP"）。 */ - private fun mailboxWithMinId5(): List { - val ids = (1..8).map { inbox.insertRaw("") } - ids.take(4).forEach { inbox.removeRow(it) } - return ids.drop(4) - } - - @Test - fun `default does nothing - no seeding without explicit configuration`() { - val kept = mailboxWithMinId5() - - assertEquals(0, poller.pollOnce(t0)) // W=0 → ID=1 判为空洞，不推进 - - assertNull(cursor.cursor.seededAt) - assertEquals(0L, cursor.cursor.committedUpTo) - assertNotNull(cursor.cursor.holeSince) - assertTrue(kept.all { proc.find(it) == null }) - } - - @Test - fun `min mode reads all currently existing rows`() { - val kept = mailboxWithMinId5() - props.pipeline.cutoverWatermark = "min" - - assertEquals(4, poller.pollOnce(t0)) // 播种 W=4，同一轮把 5..8 全部读入 - - assertNotNull(cursor.cursor.seededAt) - assertEquals(8L, cursor.cursor.committedUpTo) - assertTrue(kept.all { proc.find(it) != null }) - } - - @Test - fun `max mode skips the currently visible backlog`() { - mailboxWithMinId5() - props.pipeline.cutoverWatermark = "max" - - assertEquals(0, poller.pollOnce(t0)) - - assertNotNull(cursor.cursor.seededAt) - assertEquals(8L, cursor.cursor.committedUpTo) // 直接跳到 MAX(ID) - assertEquals(0, proc.rows.size) - } - - @Test - fun `zero mode scans from zero and stops at the first hole`() { - mailboxWithMinId5() - props.pipeline.cutoverWatermark = "zero" - - assertEquals(0, poller.pollOnce(t0)) - - assertNotNull(cursor.cursor.seededAt) - assertEquals(0L, cursor.cursor.committedUpTo) // 1..4 是空洞 → 按空洞规则停住 - assertNotNull(cursor.cursor.holeSince) - assertEquals(0, proc.rows.size) - } - - @Test - fun `explicit boundary mode is accepted`() { - mailboxWithMinId5() - props.pipeline.cutoverWatermark = "6" - - poller.pollOnce(t0) - - assertNotNull(cursor.cursor.seededAt) - assertEquals(8L, cursor.cursor.committedUpTo) // 从 6 起读 7..8（6 本身已在界内） - } - - @Test - fun `upgraded instance with existing watermark refuses to re-seed`() { - mailboxWithMinId5() - cursor.save(InboxCursorRepository.Cursor(committedUpTo = 3L, holeSince = t0)) - props.pipeline.cutoverWatermark = "max" - - poller.pollOnce(t0) - - assertNull(cursor.cursor.seededAt) // 没有重新播种 - assertNotEquals(8L, cursor.cursor.committedUpTo) - } - - @Test - fun `instance that already consumed messages refuses to seed even at W zero`() { - mailboxWithMinId5() - proc.insertIfAbsent(1L, t0) // 已有处理记录（即使水位还是 0） - props.pipeline.cutoverWatermark = "max" - - poller.pollOnce(t0) - - assertNull(cursor.cursor.seededAt) - } - - @Test - fun `seeding happens at most once and survives normal watermark saves`() { - mailboxWithMinId5() - props.pipeline.cutoverWatermark = "max" - poller.pollOnce(t0) - val seededAt = cursor.cursor.seededAt - assertNotNull(seededAt) - - cursor.save(InboxCursorRepository.Cursor(committedUpTo = 2L, holeSince = null)) - poller.pollOnce(t0.plusSeconds(1)) - - assertEquals(seededAt, cursor.cursor.seededAt) // 标记仍在：普通 save 不抹播种事实 - assertEquals(2L, cursor.cursor.committedUpTo) - } - - @Test - fun `empty mailbox is left unseeded`() { - props.pipeline.cutoverWatermark = "max" - - assertEquals(0, poller.pollOnce(t0)) - - assertNull(cursor.cursor.seededAt) - } -} 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 d2c0c51..543f0b0 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPollerTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPollerTest.kt @@ -4,49 +4,46 @@ 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 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.time.Clock import java.time.Instant +import java.time.ZoneOffset /** - * 收报环节最要紧的几条规矩： - * - 取新消息只看 ID，不看处理标记。处理完却没能回填的行（尤其是永远不回填的死信） - * 不允许占住批次，也不允许挡住后面的新消息——这是曾经的线上隐患； - * - 水位只在成功登记后才推进，而且和登记写在同一个事务里，中断后重扫就能补齐； - * - 遇到 ID 缺口先停下来（可能有更小的消息还没到），缺口等太久则跳过（否则水位永远卡住）； + * 收报环节最要紧的几条规矩（`INV-1`）： + * - 取新消息只看处理时间是否为空，按编号升序、每批 `claim-batch` 条； + * - 同一编号只建一条处理记录：重扫、重启、与兼容入口并发都不会多建； + * - 已回填的行不再被扫到；终态但未回填的行会重复扫到，但不许占住批次挡住后面的新行； * - 这一层不碰信箱的处理标记，标记留给回填环节写。 */ class InboxPollerTest { private val t0: Instant = Instant.parse("2026-09-08T03:00:00Z") private val props = PipelineProps() + private val clock: Clock = Clock.fixed(t0, ZoneOffset.UTC) private lateinit var inbox: StubInbox private lateinit var proc: StubProcState - private lateinit var cursor: StubInboxCursor private lateinit var poller: InboxPoller @BeforeEach fun setUp() { inbox = StubInbox().apply { clear() } proc = StubProcState().apply { clear() } - cursor = StubInboxCursor().apply { clear() } - poller = InboxPoller( - inbox, proc, cursor, StubPipelineTx(), props, - java.time.Clock.fixed(t0, java.time.ZoneOffset.UTC), - ) + poller = newPoller() } + private fun newPoller() = InboxPoller(inbox, proc, StubPipelineTx(), props, clock) + @Test - fun `external rows are enqueued in id order and advance the watermark without marking the mailbox`() { + fun `unprocessed rows are enqueued in id order without marking the mailbox`() { val first = inbox.simulateExternalWrite("") val second = inbox.simulateExternalWrite("") @@ -54,16 +51,29 @@ class InboxPollerTest { assertEquals(ProcStatus.PENDING, proc.find(first)!!.state) assertEquals(ProcStatus.PENDING, proc.find(second)!!.state) - assertEquals(second, cursor.cursor.committedUpTo) - assertNull(cursor.cursor.holeSince) + assertEquals(listOf(first, second), proc.rows.keys.sorted()) // 收报只写自有库，不碰信箱的处理标记 assertFalse(inbox.isMarked(first)) - assertEquals(0, poller.pollOnce(t0)) // 重复扫描幂等 + assertEquals(0, poller.pollOnce(t0)) // 重扫幂等：同编号不再建第二行 + assertEquals(2, proc.rows.size) + } + + /** 已写回处理时间的行退出扫描谓词，不再被读到。 */ + @Test + fun `rows already backfilled leave the scan predicate`() { + val done = inbox.simulateExternalWrite("") + assertEquals(1, poller.pollOnce(t0)) + proc.markTerminal(done, ProcStatus.SUCCEEDED, now = t0) + inbox.markProcessedIfUnmarked(done, "PROCESSED") + proc.rows.clear() // 模拟到期清理后处理记录已删除 + + assertEquals(0, poller.pollOnce(t0)) // 标记已写回 → 不再重建记录 + assertNull(proc.find(done)) } /** - * 回归用例：处理完却永远不会回填的行（典型是解码失败的死信）曾经占满每一批的名额， - * 导致收报整体停摆。取新消息这件事必须和"有没有处理标记"彻底分开。 + * 回归用例：处理完却还没回填的行（典型是等待回填的死信）会被反复扫到， + * 但不得占满首批窗口——否则攒够一批之后新消息就再也发现不了（曾经的线上隐患）。 */ @Test fun `terminal rows without a mailbox mark do not block discovery of later messages`() { @@ -78,105 +88,60 @@ class InboxPollerTest { assertEquals(1, poller.pollOnce(t0)) assertEquals(ProcStatus.PENDING, proc.find(fresh)!!.state) - assertEquals(fresh, cursor.cursor.committedUpTo) } + /** 未回填的存量远多于一批时，续读要一直翻到最后一批，才能发现末尾的新行。 */ @Test - 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) + fun `multi batch scan reaches new rows behind several full batches`() { + props.pipeline.claimBatch = 2 + val stale = (1..5).map { inbox.simulateExternalWrite("") } + assertEquals(5, poller.pollOnce(t0)) + stale.forEach { proc.markTerminal(it, ProcStatus.SUCCEEDED, now = t0) } + + val fresh = inbox.simulateExternalWrite("") 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) + assertNotNull(proc.find(fresh)) } + /** 重启不带任何消费边界：从头重扫，已登记的行不会被建成第二行。 */ @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) // 记录空洞观测时刻 + fun `a restarted poller rescans from the start without duplicating rows`() { + val ids = (1..3).map { inbox.simulateExternalWrite("") } + assertEquals(3, poller.pollOnce(t0)) - val agedOut = t0.plus(props.pipeline.maxCommitDelay) - assertEquals(0, poller.pollOnce(agedOut)) // 空洞判永久：推进水位但不越过入队 + val restarted = newPoller() - assertNull(cursor.cursor.holeSince) - assertEquals(afterHole - 1, cursor.cursor.committedUpTo) - assertEquals(1, poller.pollOnce(agedOut)) // 下一轮恢复发现 - assertEquals(afterHole, cursor.cursor.committedUpTo) - assertNotNull(proc.find(afterHole)) + assertEquals(0, restarted.pollOnce(t0.plusSeconds(60))) + assertEquals(ids.size, proc.rows.size) + assertEquals(ids, proc.rows.keys.sorted()) + } + + /** 编号较小的行晚提交：本轮读不到不算丢，下一轮照常发现。 */ + @Test + fun `a late committed smaller id is discovered on the next round`() { + val first = inbox.simulateExternalWrite("") + val late = inbox.simulateExternalWrite("") + val third = inbox.simulateExternalWrite("") + inbox.removeRow(late) + + assertEquals(2, poller.pollOnce(t0)) + assertNotNull(proc.find(first)) + assertNotNull(proc.find(third)) + assertNull(proc.find(late)) + + inbox.restoreRow(late, "", t0.plusSeconds(30)) + + assertEquals(1, poller.pollOnce(t0.plusSeconds(30))) + assertEquals(ProcStatus.PENDING, proc.find(late)!!.state) } @Test fun `compat http path and poller do not double enqueue the same message`() { - val receipt = InboxService(inbox, proc, java.time.Clock.fixed(t0, java.time.ZoneOffset.UTC)).accept("") + val receipt = InboxService(inbox, proc, clock).accept("") assertEquals(0, poller.pollOnce(t0)) - assertEquals(receipt.msgId, cursor.cursor.committedUpTo) // 已在 PG：读取进度照常推进 + assertEquals(1, proc.rows.size) assertNotNull(proc.find(receipt.msgId)) } - - @Test - fun `a newly exposed hole does not inherit the age of the previous hole`() { - val first = inbox.simulateExternalWrite("") - val oldHole = inbox.simulateExternalWrite("") - val third = inbox.simulateExternalWrite("") - val newHole = inbox.simulateExternalWrite("") - val fifth = inbox.simulateExternalWrite("") - inbox.removeRow(oldHole) - inbox.removeRow(newHole) - - assertEquals(1, poller.pollOnce(t0)) - val almostAged = t0.plus(props.pipeline.maxCommitDelay).minusSeconds(1) - inbox.restoreRow(oldHole, "", almostAged) - - assertEquals(2, poller.pollOnce(almostAged)) - assertEquals(third, cursor.cursor.committedUpTo) - assertEquals(almostAged, cursor.cursor.holeSince) - assertNull(proc.find(fifth)) - - // 旧空洞的期限已到，但新空洞必须获得完整等待窗口。 - assertEquals(0, poller.pollOnce(t0.plus(props.pipeline.maxCommitDelay))) - assertEquals(third, cursor.cursor.committedUpTo) - assertNull(proc.find(fifth)) - assertEquals(first + 2, third) - } - - /** - * 兼容入口会把消息登记在水位**之外**（`msgId > W`），因此它天然成为"最小未完成行"。 - * 领取侧的守卫在主泵：只领 `msgId <= W`（端到端验收见 `PipelineSmokeTest` 的 - * "compat injected high id is not claimed until the watermark catches up"）。 - * 本用例只固定收报侧的事实：**登记发生，但水位不动**。 - */ - @Test - fun `compat accept registers a row above the watermark without advancing it`() { - props.pipeline.claimBatch = 10 - val one = inbox.simulateExternalWrite("") // 1 - val hole = inbox.simulateExternalWrite("") // 2（空洞） - val third = inbox.simulateExternalWrite("") // 3 - inbox.removeRow(hole) - - assertEquals(1, poller.pollOnce(t0)) // W=1，遇空洞即停 - assertEquals(one, cursor.cursor.committedUpTo) - assertNull(proc.find(third)) // 3 还没入队 - - val receipt = InboxService(inbox, proc, java.time.Clock.fixed(t0, java.time.ZoneOffset.UTC)) - .accept("") // 兼容入口：高 ID 直接进 PG - val high = receipt.msgId - assertTrue(high > third) - - assertEquals(one, cursor.cursor.committedUpTo) // 水位不动（不参与水位） - // 主泵处理掉队头 1 之后，最小未完成行就是兼容入口写进来的高 ID…… - proc.markTerminal(one, ProcStatus.SUCCEEDED, now = t0) - assertEquals(high, proc.headUnfinished()!!.msgId) - assertTrue(high > cursor.cursor.committedUpTo) // ……但它超出水位，主泵不会领取 - assertNull(proc.find(third)) - } } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxServiceTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxServiceTest.kt index 4e5b0fa..10e8871 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxServiceTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxServiceTest.kt @@ -7,7 +7,6 @@ import com.gzzn.omms.msgexchange.infra.persistence.MailboxMarkResult import com.gzzn.omms.msgexchange.infra.persistence.MailboxRow import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository 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 org.junit.jupiter.api.Assertions.assertEquals @@ -43,11 +42,7 @@ class InboxServiceTest { else -> Instant.parse("2026-09-06T01:59:55Z") } - override fun readRange(fromExclusive: Long, limit: Int): List = emptyList() - - override fun maxId(): Long? = null - - override fun minId(): Long? = null + override fun readUnprocessed(afterId: Long, limit: Int): List = emptyList() override fun markProcessedIfUnmarked(msgId: Long, value: String): MailboxMarkResult = MailboxMarkResult.MISSING } @@ -100,10 +95,9 @@ class InboxServiceTest { fun `the poller backfills a compat receipt whose pg enqueue failed, exactly once`() { val inbox = StubInbox().apply { clear() } val proc = StubProcState().apply { clear() } - val cursor = StubInboxCursor().apply { clear() } val clock = MutableClock(t0) val service = InboxService(inbox, FlakyProcState(proc), clock) - val poller = InboxPoller(inbox, proc, cursor, StubPipelineTx(), PipelineProps(), clock) + val poller = InboxPoller(inbox, proc, StubPipelineTx(), PipelineProps(), clock) val receipt = service.accept("") // 落信成功；PG 入队失败但不抛 assertEquals(1L, service.pgEnqueueFailures.get()) @@ -112,7 +106,6 @@ class InboxServiceTest { assertEquals(1, poller.pollOnce(t0)) // 轮询补建 assertNotNull(proc.find(receipt.msgId)) assertEquals(1, proc.rows.size) - assertEquals(receipt.msgId, cursor.cursor.committedUpTo) assertEquals(0, poller.pollOnce(t0)) // 已存在：不重复建行，也不重复计数 assertEquals(1, proc.rows.size) diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/BackfillServiceTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/BackfillServiceTest.kt index 266d6ab..0c4a9e7 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/BackfillServiceTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/BackfillServiceTest.kt @@ -46,9 +46,7 @@ class BackfillServiceTest { override fun insertRaw(rawXml: String): Long = 1L override fun rawOf(msgId: Long): String? = null override fun receivedAtOf(msgId: Long): Instant? = null - override fun readRange(fromExclusive: Long, limit: Int): List = emptyList() - override fun maxId(): Long? = null - override fun minId(): Long? = null + override fun readUnprocessed(afterId: Long, limit: Int): List = emptyList() override fun markProcessedIfUnmarked(msgId: Long, value: String): MailboxMarkResult { if (fail) throw IllegalStateException("mysql-down") if (missing) return MailboxMarkResult.MISSING @@ -237,9 +235,7 @@ class BackfillServiceTest { override fun insertRaw(rawXml: String) = 1L override fun rawOf(msgId: Long): String? = "" override fun receivedAtOf(msgId: Long) = t0 - override fun readRange(fromExclusive: Long, limit: Int) = emptyList() - override fun maxId(): Long? = 1L - override fun minId(): Long? = 1L + override fun readUnprocessed(afterId: Long, limit: Int) = emptyList() override fun markProcessedIfUnmarked(msgId: Long, value: String): MailboxMarkResult { entered.countDown() release.await() diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt index 6938785..b684f93 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt @@ -11,14 +11,12 @@ import com.gzzn.omms.msgexchange.domain.MsgKind import com.gzzn.omms.msgexchange.domain.ProcState import com.gzzn.omms.msgexchange.domain.ProcStatus import com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters -import com.gzzn.omms.msgexchange.infra.persistence.InboxCursorRepository import com.gzzn.omms.msgexchange.infra.persistence.PipelineLockRepository import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager import com.gzzn.omms.msgexchange.infra.retry.ProcFailure import com.gzzn.omms.msgexchange.infra.retry.FailureScheduler import com.gzzn.omms.msgexchange.infra.stub.StubFlightState import com.gzzn.omms.msgexchange.infra.stub.StubInbox -import com.gzzn.omms.msgexchange.infra.stub.StubInboxCursor import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents import com.gzzn.omms.msgexchange.infra.stub.StubProcState import com.gzzn.omms.msgexchange.infra.stub.StubSnapshotLog @@ -63,8 +61,6 @@ class IgnoreBranchTest { counters: PipelineCounters = PipelineCounters(), codec: XmlCodec, ): MessageProcessor { - val cursor = StubInboxCursor() - cursor.cursor = InboxCursorRepository.Cursor(committedUpTo = Long.MAX_VALUE) val clock = Clock.systemUTC() val props = PipelineProps() val opDay = OperationDayProps().apply { zone = "Asia/Shanghai"; cutoffHour = 0 }