From 817236ca26dcd1cda0e67dedc816d573252f9851 Mon Sep 17 00:00:00 2001 From: windyboy Date: Fri, 11 Sep 2026 20:44:30 +0800 Subject: [PATCH] =?UTF-8?q?fix(processing):=20=E8=90=BD=E5=9C=B0=20D1/D5/D?= =?UTF-8?q?6=20=E4=B8=89=E9=A1=B9=E8=A3=81=E5=86=B3=EF=BC=8C=E5=88=A0?= =?UTF-8?q?=E9=99=A4=20head-deadline=20=E5=8F=82=E6=95=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit D5（终态判据只保留尝试上限）： - Pump.tick 内联 attempts 判定，删除 head-deadline 相关的毒丸分支与滞留告警代码 - 删除配置项 head-deadline（PipelineProps / application.yml）与 PumpDeadlineTest - PROCESSING_STARTED_AT 变为只写，注释如实说明当前无判据消费它 D1（回填放弃判据改为时间）： - 暂时性故障在 R 之前只退避重试，不再按尝试次数放弃；到 R 才放弃并记 TRANSIENT_DEADLINE - backfill-max-attempts 降级为单行重试的告警阈值 D6（超期判据改用本地入队时间）： - 新增 V6 迁移：PROC_STATE 加 ENQUEUED_AT（回填存量后置为非空 + 默认） - findBackfillDue 的谓词与 overdue 标记改比较 enqueued_at，不再用库方时钟的 received_at - BackfillDue 增加 overdue；收报与兼容入口显式写入本地入队时间 文档同步： - 清理 4 处 message-lifecycle.md 章节号死链（Pump/InboxService/PipelineProps/application.yml） - 关闭 G-HEAD-DEADLINE、G-BACKFILL-ABANDON-BYTIME、G-ENQUEUED-AT 三条缺口登记 - reference/user-stories/README 与实现对齐 验证：./gradlew test ⇒ 122 tests, 0 failures, 1 skipped Refs: ACM2-45 --- docs/README.md | 2 +- docs/design.md | 12 ++--- docs/invariants.md | 5 +- docs/reference.md | 9 ++-- docs/user-stories.md | 4 +- .../omms/msgexchange/config/PipelineProps.kt | 3 +- .../gzzn/omms/msgexchange/domain/ProcState.kt | 20 ++++++-- .../infra/persistence/Repositories.kt | 20 +++++--- .../persistence/jdbc/JdbcPgRepositories.kt | 36 +++++++++----- .../infra/stub/StubRepositories.kt | 23 ++++++--- .../omms/msgexchange/ingress/InboxPoller.kt | 3 +- .../omms/msgexchange/ingress/InboxService.kt | 8 +-- .../msgexchange/processing/BackfillService.kt | 42 ++++++++++------ .../gzzn/omms/msgexchange/processing/Pump.kt | 25 ++++------ src/main/resources/application.yml | 5 +- .../db/migration/V6__enqueued_at.sql | 22 +++++++++ .../persistence/jdbc/FlywayMigrationTest.kt | 18 +++++-- .../jdbc/InboxLifecycleJdbcSqlTest.kt | 17 +++++-- .../infra/retry/ReplayServiceTest.kt | 6 ++- .../processing/BackfillServiceTest.kt | 49 +++++++++++++------ .../processing/PumpDeadlineTest.kt | 40 --------------- 21 files changed, 217 insertions(+), 152 deletions(-) create mode 100644 src/main/resources/db/migration/V6__enqueued_at.sql delete mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/processing/PumpDeadlineTest.kt diff --git a/docs/README.md b/docs/README.md index 6345bd6..c3ed212 100644 --- a/docs/README.md +++ b/docs/README.md @@ -34,7 +34,7 @@ | 保留期下界 `R_keep`、清除前置条件 | contracts「保留与清除」 | design 只引 `C-x`；执行步骤在上线前另立 | | 处理标记值集与写权限 | contracts `C-5` | design 只写行为约束「只写空标记、不回撤、不覆盖」（`INV-6`） | | 回填四结果、放弃语义、`R` 的作用 | design「回填」 | invariants 记结论与可声明性 | -| `head-deadline` / 退避 / `claim-batch` 等取值 | reference「参数」 | design 只引 `PARAM:x` | +| 退避 / `claim-batch` / 回填期限等取值 | reference「参数」 | design 只引 `PARAM:x` | | 消费权排他、ID 不复位、报文不可变、时钟、单实例 | invariants「前提」 | 其他文档只引 `PRE-x` | | 航班身份、合并语义、`STATE_VERSION`、`OPERATION_DAY` | flight-state.md | design 只引域规则 | | 对外术语（落信 / 入站 / 库方 / 处理标记） | contracts「术语」 | — | diff --git a/docs/design.md b/docs/design.md index 94b11ae..8342763 100644 --- a/docs/design.md +++ b/docs/design.md @@ -23,7 +23,7 @@ | 记录 | 用途 | 关键约束 | |---|---|---| -| `PROC_STATE` | 入站消息的处理状态、身份、尝试次数、错误原因与回填事实 | `MSG_ID = CMINMSGS_ID` 主键防重复入队；`IDENTITY_KEY` 唯一约束防业务重复；按最小未完成 `MSG_ID` 取队头；`BACKFILL_NEXT_AT` 非空 = 还欠一次回填，`BACKFILL_AT` 非空 = 标记已确认，`BACKFILL_ABANDONED_AT/REASON` 非空 = 已停止自动重试（**不等于**标记已确认）；`RECEIVED_AT` 复制自信箱接收时间、**可能为 NULL**、仅用于对账与展示；`ENQUEUED_AT` 是本地入队时间、非空、是超期判据的唯一依据 `[G-ENQUEUED-AT]`。 | +| `PROC_STATE` | 入站消息的处理状态、身份、尝试次数、错误原因与回填事实 | `MSG_ID = CMINMSGS_ID` 主键防重复入队；`IDENTITY_KEY` 唯一约束防业务重复；按最小未完成 `MSG_ID` 取队头；`BACKFILL_NEXT_AT` 非空 = 还欠一次回填，`BACKFILL_AT` 非空 = 标记已确认，`BACKFILL_ABANDONED_AT/REASON` 非空 = 已停止自动重试（**不等于**标记已确认）；`RECEIVED_AT` 复制自信箱接收时间、**可能为 NULL**、仅用于对账与展示；`ENQUEUED_AT` 是本地入队时间、非空、是超期判据的唯一依据。 | | `MSG_EVENT` | 等待投递的事件（outbox） | `EVENT_ID` 决定投递顺序（全局串行分配，见投递）；`TARGET` 区分 `KAFKA:msg` / `KAFKA:schd`；`PARTITION_KEY` 恒为 `FLID`；`EVENT_TYPE` 区分 UPSERT 与 TOMBSTONE。`KAFKA:schd` 按 `FLID` 单行 upsert，只保留最新 `STATE_VERSION`。 | | `REQ_TRACK` | 上游请求及应答关联 | 状态 `PENDING / SENT / DONE / EXPIRED`；保存请求类型、覆盖运营日、发送方、出站信箱 ID 与发送/完成时间；**「同类只允许一个开放请求」的唯一键 = `(请求类型, 覆盖运营日, 发送方)`，且仅对开放状态生效**。登记、超时与应答匹配尚未实现 `[G-REQ-TRACK]`。 | | `REF_MASTER` | 静态参考数据（目标表） | `(RTYPE, RKEY)` 唯一；尚未建表，客户端与刷新流程见 user-stories US-13/US-14，US-14 两类映射的存储落点未定。 | @@ -126,7 +126,7 @@ 4. 队头为 `FAILED` 且未到 `next_attempt_at`：休眠到可重试时刻，不处理后续消息。 5. 其余（新消息或退避到期的重试）：调用 `MessageProcessor.processOne`，失败迁移在该边界内完成。 -**队头滞留只告警、不迁移状态**：`PARAM:msgx.pipeline.head-deadline` 用于 `msgloop` 健康告警（「队头滞留超阈值」），**不作为终态判据**。理由：按尝试上限与退避表，正常重试包络远小于该阈值，它只在单次处理长时间卡住时触发；而处理卡死应由外部调用的有界超时兜底，用超时把消息直接推入 `DEAD` 会绕过人工复核，并制造与人工重放并发的旁路写入者。 +**终态判据只有尝试上限，没有按时间的毒丸**：处理卡死应由外部调用的有界超时兜底；用一个时间阈值把消息直接推入 `DEAD` 会绕过人工复核，并制造与人工重放并发的旁路写入者。队头年龄由 `msgx.pipeline.backlog.oldest_unprocessed_seconds` 观测，主泵不做这项判定。 **所有取时统一经注入 `Clock`**（收报空洞老化、主泵调度、处理器落库时间、回填重试、作业切日），不使用系统时钟。 @@ -177,7 +177,7 @@ processOne(head)： - 顺序由 `MSG_ID` 决定，不由执行方式决定。入队与处理由不同线程驱动、可以并发，「先入队后处理」只是可选的运维规程，不是正确性前提；系统不提供「只入队」模式。 - 不加速、不分流、不走旁路：不允许并行队头，也不允许实时消息跳过积压。 -- 尝试上限、退避与队头滞留告警对积压同样生效，不因积压而放宽。 +- 尝试上限与退避对积压同样生效，不因积压而放宽。 - 经确认不再处理的行置 `SKIPPED` 并记录原因，到达终态后走回填通道；不存在「整段 DELETE」的快速通道（授权与留痕见 `C-27`／`Q12`）。 - 消化期间的可观测项与完成时限口径见 reference 与 CLM-9：**扫描周期不是完成时限**。 @@ -198,7 +198,7 @@ ORDER BY BACKFILL_ATTEMPTS ASC, MSG_ID ASC -- 公平轮转，永久失败 LIMIT PARAM:msgx.pipeline.backfill-batch ``` -超期判据使用**本地入队时间**（`ENQUEUED_AT`），不使用信箱的 `RECEIVED_AT`：后者来自外部时钟，前偏会在「打标即清除」语义下造成提前清除（`PRE-4`）。`ENQUEUED_AT` 落地前，该分支沿用 `RECEIVED_AT` 并在声明边界标注时钟偏斜风险 `[G-ENQUEUED-AT]`。 +超期判据使用**本地入队时间**（`ENQUEUED_AT`），不使用信箱的 `RECEIVED_AT`：后者来自外部时钟，前偏会在「打标即清除」语义下造成提前清除（`PRE-4`）。 ### 6.2 四种结果与放弃 @@ -218,7 +218,7 @@ LIMIT PARAM:msgx.pipeline.backfill-batch 1. **取消退避**：已终态但超期未打标的行，每轮扫描都被尝试，不再等退避到期； 2. **暂时性故障的放弃期限**：超时、连接失败这类暂时性故障**在 `R` 之前只退避重试、不放弃**；到 `R` 仍未打标才停止自动重试、记入放弃清单并告警。 -放弃判据用**时间**而不是**尝试次数**：按退避表（档位 ≤ 8 秒）与扫描周期，固定次数的实际上限只有几十分钟，一次小时级的共享库故障会把全部待回填行一次性判死，随后必须成批人工恢复——这是必须避免的失败模式。`PARAM:msgx.pipeline.backfill-max-attempts` 因此不再是放弃判据，只保留为单行重试的告警阈值 `[G-BACKFILL-ABANDON-BYTIME]`。 +放弃判据用**时间**而不是**尝试次数**：按退避表（档位 ≤ 8 秒）与扫描周期，固定次数的实际上限只有几十分钟，一次小时级的共享库故障会把全部待回填行一次性判死，随后必须成批人工恢复——这是必须避免的失败模式。`PARAM:msgx.pipeline.backfill-max-attempts` 因此不再是放弃判据，只保留为单行重试的告警阈值。 关于「最终一定打标」，准确表述是三段，缺一不可： @@ -314,7 +314,7 @@ PENDING → SENT → DONE | 标记写入中途 | 标记仍为空 | 重新写入；重复写入同一值无副作用 | | 兼容入口已入队、水位未追平 | PG 已有该 ID 的记录 | 轮询读到该行时主键幂等，水位照常推进 | | 回填时信箱行已不存在 | 写入 0 行且信箱行不存在 | 立即放弃自动重试（`MISSING_ROW`）并告警；放弃不等于标记已确认，仍需人工对账 | -| `RECEIVED_AT` 为 NULL | 超期分支以 `ENQUEUED_AT` 判定，不再失效 | `[G-ENQUEUED-AT]` 落地前靠退避重试保证 | +| `RECEIVED_AT` 为 NULL | 超期分支以本地 `ENQUEUED_AT` 判定，不受库方时钟与 NULL 影响 | 按 `R` 超期强补写；未超期则按退避重试 | | 投递目标已接受、`SENT` 未置 | 事件仍 `PENDING` | 允许重发，消费方按事件身份去重 | | PG 从备份恢复 | 终态与事件回退到备份点 | 按至少一次接受重复；不重放业务、不据此改写航班 | diff --git a/docs/invariants.md b/docs/invariants.md index 1b92496..51ca40b 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -79,7 +79,7 @@ | INV-5 | 终态未回填不阻断发现 | 缺口（补齐后应断言发现谓词不引用处理状态） | | INV-6 | 投递失败后终态不变 | 缺口 | | INV-7 | 回填四种结果 | 写入成功 / 早已标记（不覆盖、记成功）/ 信箱行不存在（立即放弃并告警，不得视为已标记）/ 暂时故障持续到 `R` 仍未打标（停止自动重试，可人工恢复） | -| INV-7 | `RECEIVED_AT` 为 NULL | 超期兜底不生效的行为被显式验证，且不导致标记提前写入（`G-ENQUEUED-AT` 落地后改为覆盖该分支） | +| INV-7 | `RECEIVED_AT` 为 NULL | 超期分支仍成立且不导致标记提前写入——判据是本地 `ENQUEUED_AT`，与库方时钟及 NULL 无关 | | INV-8 | PG 提交失败、信箱回填失败 | 事件、终态与回填意图一起回滚；已提交结果只补写标记，不重放业务；中间态永不补写 | | INV-8 | 非业务型终态 | 不触碰航班表 / `MSG_EVENT`，只写 `PROC_STATE`，且终态与回填意图同语句生效 | | INV-9 | 同身份多条记录、失败后重试、归档后重复 | 只产生一次有效业务处理，不把自身重试判为重复 | @@ -107,12 +107,9 @@ | `G-RESP-GUARD` | `RESP` 应答守卫未实现，当前与 `DNLD` 无差别进入快照写入 | 请求匹配闭环；`C-23` | | `G-REQ-TRACK` | `REQ_TRACK` 无运行时协调器：出站适配、请求编码、超时与应答匹配未实现 | US-08；`C-24` | | `G-PROC-HST` | `PROC_STATE_HST` 未建表，终态归档未落地 | US-11；归档能力 | -| `G-ENQUEUED-AT` | `PROC_STATE` 尚无 `ENQUEUED_AT` 列，超期判据暂用 `RECEIVED_AT` | `INV-7` / CLM-4；跨系统时钟偏斜与 `RECEIVED_AT` 为 NULL | | `G-FLOP-IDEMPOTENT` | 29 类 FLOP 幂等矩阵未补全 | `INV-20`、CLM-3 | -| `G-BACKFILL-ABANDON-BYTIME` | 回填放弃判据由「尝试次数」改为「`R` 超期」尚未落地 | `INV-7`、CLM-4 | | `G-EVENT-RETENTION` | `MSG_EVENT` 已发送行的保留期与清理作业未实现 | outbox 有界性 | | `G-BACKFILL-BACKOFF` | 回填独立退避键（`backfill-backoff-ms` / `-cap-ms`）未实现，暂沿用处理退避表 | 回填重试节奏 | -| `G-HEAD-DEADLINE` | `head-deadline` 已定为「仅告警」，代码注释与判据仍写「毒丸升级」 | design「主泵调度」的表述一致性 | | `G-KAFKA-D3` | `kafka.producers.default.max-in-flight` 默认 5，与架构决策 D3 要求的 1 不一致 | 投递幂等前提 | | `G-JOB-HEARTBEAT` | 作业心跳、扫描积压、实际回填延迟指标未实现 | CLM-9；回填可观测性 | | `G-REPLAY-CHANNEL` | 「打标即清除」语义下的独立原文保留通道未设计 | CLM-5 | diff --git a/docs/reference.md b/docs/reference.md index 232731f..c286a92 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -17,13 +17,12 @@ | `msgx.pipeline.max-attempts` | `5` | 次 | 假定 | 处理与投递共用；达到即转 `DEAD(EXHAUSTED)` | | `msgx.pipeline.backoff-ms` | `[1000,2000,4000,8000]` | ms / 档 | 假定 | **档位数必须 = `max-attempts − 1`**，启动自检拦截错位 | | `msgx.pipeline.backoff-cap-ms` | `60000` | ms | 假定 | 单档封顶；默认表内无档触及 | -| `msgx.pipeline.head-deadline` | `10m` | Duration | 假定 | **仅用于队头滞留告警**，不作终态判据（判据只由 `max-attempts`）。代码注释仍写「毒丸升级」，属待收敛的表述差 `[G-HEAD-DEADLINE]` | | `msgx.pipeline.max-commit-delay` | `5m` | Duration | **假定（无依据）** | 空洞老化阈值；由 `C-2` 决定，**不可由 SIS `Expiry` 推导**（`Q2`） | | `msgx.pipeline.overdue-backfill`（`R`） | `30d` | Duration | 契约（`R ≤ R_keep`） | 进入强补写窗口、**取消退避**的阈值；**不是兜底保证**，不保护重放窗口（`Q6`） | | `msgx.pipeline.backfill-batch` | `100` | 条 | 假定 | 回填扫描单批条数 | -| `msgx.pipeline.backfill-max-attempts` | `100` | 次 | **改作告警阈值** | **不再是放弃判据**：暂时性故障按 `R` 超期放弃（见 design「回填」）。该键保留为单行重试的告警阈值 `[G-BACKFILL-ABANDON-BYTIME]` | -| `msgx.pipeline.backfill-backoff-ms` | 目标参数（未实现） | ms / 档 | 假定 | 回填独立退避表；**当前不存在，回填沿用 `backoff-ms`** `[G-BACKFILL-BACKOFF]` | -| `msgx.pipeline.backfill-backoff-cap-ms` | 目标参数（未实现） | ms | 假定 | 回填退避封顶；旧文档「封顶 15 分钟」无对应配置键，已作废 `[G-BACKFILL-BACKOFF]` | +| `msgx.pipeline.backfill-max-attempts` | `100` | 次 | 假定 | 单行重试的**告警阈值**；放弃判据是 `R` 超期，不是次数（见 design「回填」） | +| `msgx.pipeline.backfill-backoff-ms` | 目标参数（未实现） | ms / 档 | 假定 | 回填独立退避表；**当前不存在**，`BackfillService` 内硬编码 30 秒起步、封顶 15 分钟 `[G-BACKFILL-BACKOFF]` | +| `msgx.pipeline.backfill-backoff-cap-ms` | 目标参数（未实现） | ms | 假定 | 回填退避封顶；对应实现是代码内常量，尚无配置键 `[G-BACKFILL-BACKOFF]` | | `msgx.pipeline.cutover-watermark` | 不设置 | `min\|zero\|max\|` | 一次性运维决策 | 显式播种水位；非法值由启动自检挡下；升级实例拒绝重新播种 | | `msgx.pipeline.late-detect-period` | `60s` | Duration | 假定 | 只读迟到检测周期；`≤0` 关闭；机制为临时观测（见 design 扫描路径） | | `msgx.pipeline.late-detect-batch` | `200` | 条 | 假定 | 每轮复查的空洞 ID 上限 | @@ -98,7 +97,7 @@ | 持久化与恢复 | `infra/persistence/`、`infra/retry/`（`ProcFailure` / `ReplayService` / `FailureScheduler`） | | 启停与配置 | `PipelineLifecycle.kt`、`config/PipelineProps.kt`、`config/HistoryProps.kt` | | 指标与健康 | `infra/metrics/PipelineMetrics.kt`、`infra/health/BacklogSnapshotProvider.kt` | -| 迁移 | `src/main/resources/db/migration/`（V1 基线、V2 生命周期、V3 HOL 起点、V4 回填闭环、V5 切流播种；`oracle11g/` 为占位） | +| 迁移 | `src/main/resources/db/migration/`（V1 基线、V2 生命周期、V3 稳定处理起点、V4 回填闭环、V5 切流播种、V6 本地入队时间；`oracle11g/` 为占位） | ## 4. 错误分类与重放白名单 diff --git a/docs/user-stories.md b/docs/user-stories.md index 4215b1d..efcb573 100644 --- a/docs/user-stories.md +++ b/docs/user-stories.md @@ -80,12 +80,12 @@ 2. 安全解码 XML，至少覆盖 META、SCHD、FLOP、参考应答与忽略类路由；合法但能力未支持是 `UNSUPPORTED`，不能一律归为非法报文。保留原文以支持诊断和回放。 3. 解码后首次绑定 `SNDR|TYPE|STYP|SEQN`；冲突转 `SKIPPED` 并记录原 ID；自身重试保留绑定。生产 `include-day-boundary=false`，更改算法须另行评审上游序号规则。 4. `MALFORMED` 直接 `DEAD`；`CODEC_ERROR / UNSUPPORTED / INFRA` 按次数和退避处理，耗尽转 `DEAD(EXHAUSTED)`。不能无限重试未实现类型，也不能立即当非法报文丢弃。 -5. HOL deadline 必须基于稳定起始时刻（`PROC_STATE.PROCESSING_STARTED_AT`），不能用每次重试刷新的 `updatedAt` 代替；调度判断注入 `Clock`。默认 5 次重试、10 分钟滞留限制；积压与人工重放的 deadline 边界按 Q6 验证。 +5. 终态判据只有尝试上限（默认 5 次），**没有按时间的毒丸**；调度判断注入 `Clock`。积压与人工重放的边界按 Q6 验证。 6. 主泵在同一 PG 事务提交航班主表/明细、事件与处理结果；终态回填意图通过 US-09 同事务保存。任一步失败整体回滚；提交后只重试外部回填，不重复生成业务事件。 7. 领域决策逻辑只读取当前完整态与已解码报文，返回下一完整态和待发事件，不执行 I/O；Processor 作为事务协调器，将状态写入、事件、处理终态与回填意图收敛在同一事务边界内，不直接触碰 Kafka。失败只在持有消息上下文的边界落账，中断向上传递，不作为普通失败吞掉。 8. 权威存储不可用或未完成恢复时停止业务处理；不能把“整个状态丢失”误判为“单航班不存在”而批量成功结束增量报文。 -**当前基础与落点**：`processing/Pump.kt`（含 `MessageProcessor`）、`DynamicProcessors.kt`、`Identity.kt`、`codec/JacksonXmlCodec.kt`、`infra/retry/`。严格 FIFO 主泵、SCHD（DNLD/RESP/ADFT）/FLOP/FDEL 处理器、PG 单事务（含回填意图预登记）、身份绑定与重试已实现；调度取时由可注入 `Clock` 提供，HOL deadline 以 `PROCESSING_STARTED_AT` 为稳定起点。尚未完成：忽略规则分支（US-04），以及逐类矩阵与 golden 样例（US-05）。长期积压和人工重放如何计算 deadline 仍由 Q6 定案。 +**当前基础与落点**：`processing/Pump.kt`（含 `MessageProcessor`）、`DynamicProcessors.kt`、`Identity.kt`、`codec/JacksonXmlCodec.kt`、`infra/retry/`。严格 FIFO 主泵、SCHD（DNLD/RESP/ADFT）/FLOP/FDEL 处理器、PG 单事务（含回填意图预登记）、身份绑定与重试已实现；调度取时由可注入 `Clock` 提供。尚未完成：忽略规则分支（US-04），以及逐类矩阵与 golden 样例（US-05）。长期积压和人工重放如何计算边界仍由 Q6 定案。 **前置**：US-01；Q1 已定单库方向，Q6 决定 deadline 边界。数据库迁移只落自有库。 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 d97d167..4111ffc 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt @@ -26,7 +26,6 @@ class PipelineProps { // 因此最多只用得到 max-attempts − 1 个档位。多出来的档位永远走不到（会被 validate() 拦下）。 var backoffMs: List = listOf(1000, 2000, 4000, 8000) var backoffCapMs: Long = 60_000 - var headDeadline: Duration = Duration.ofMinutes(10) // 最坏 HOL 上界（毒丸升级） /** * 缺口等待时长：水位后面缺了一个 ID 时，等这么久还没出现就认定它永远不会来了， @@ -42,7 +41,7 @@ class PipelineProps { * （比如回填一直失败），扫描谓词的超期分支成立，无视退避强制补写（只会提前、从不推迟打标）。 * * 唯一约束是 R ≤ R_keep，**不保护重放窗口**（打标时刻与 R 解耦，完整论证见 - * message-lifecycle.md §5.2）。取值待 Q6 定案，确认之前不要为提速下调。 + * design「回填」）。取值待 Q6 定案，确认之前不要为提速下调。 */ var overdueBackfill: Duration = Duration.ofDays(30) 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 ffbf1cd..84ff96d 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/ProcState.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/ProcState.kt @@ -41,7 +41,7 @@ enum class ErrorClass { /** 解码逻辑的问题；修好 codec 之后可以重放。 */ CODEC_ERROR, - /** 重试次数用尽或队头滞留超时；人工复核后可以重放。 */ + /** 重试次数用尽（尝试上限）；人工复核后可以重放。 */ EXHAUSTED, /** 数据库、网络等基础设施抖动，重试通常就能过。 */ @@ -64,8 +64,13 @@ data class ProcState( val errorClass: ErrorClass? = null, /** 最近一次失败的原因（截断后落库，供排查）。 */ val lastError: String? = null, - /** 信箱里的接收时间：用来判断"超期仍未回填"，也是最老未处理信龄的计算依据。 */ + /** 信箱里的接收时间，来自**库方时钟**、**可能为 NULL**；仅用于对账与展示，不作任何判据。 */ val receivedAt: Instant? = null, + /** + * 本地入队时间（本系统写入，非空）。超期补写期限 `R` **只比较它**：与判据用的本地 + * `NOW` 同源，不受库方时钟偏斜影响（`PRE-4`）。 + */ + val enqueuedAt: Instant? = null, /** 非空表示已确认信箱行带上了处理标记。 */ val backfillAt: Instant? = null, /** 非空表示还欠一次回填：写终态时置为当前时间，失败后退避推后。 */ @@ -73,15 +78,20 @@ data class ProcState( val backfillAttempts: Int = 0, val backfillError: String? = null, /** - * 非空表示已判定"不必再回填"（信箱行不存在，或达到尝试上限）。 + * 非空表示已判定"不必再回填"：信箱行不存在，或暂时性故障持续到 `R` 仍未打标。 * * **它不等于标记已确认**：`backfillAt` 仍为空，所以不满足"边界内全部行已打标"的清除条件。 * 停止重试与"已满足清除前提"是两件事，不能互相替代。 */ val backfillAbandonedAt: Instant? = null, - /** 放弃原因（`MISSING_ROW` / `MAX_ATTEMPTS`），供人工对账与恢复判断。 */ + /** 放弃原因（`MISSING_ROW` / `TRANSIENT_DEADLINE`），供人工对账与恢复判断。 */ val backfillAbandonedReason: String? = null, - /** 首次被主泵取得的时刻；重试不刷新，用作 HOL deadline 的稳定起点。 */ + /** + * 首次被主泵取得的时刻；重试不刷新。 + * + * 当前没有判据消费它（终态只看尝试上限），保留作排障与后续扩展；写入方是 + * [com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository.markProcessingStartedIfAbsent]。 + */ val processingStartedAt: Instant? = null, val updatedAt: Instant = Instant.now(), ) 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 2398e34..2619745 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 @@ -45,10 +45,11 @@ interface ProcStateRepository { * * 幂等：同一个消息 ID 重复登记既不报错、也不会建第二行（收报重扫和兼容入口并发调用都安全）。 * - * @param receivedAt 信箱里的接收时间，用于判断超期未回填和统计最老信龄 + * @param receivedAt 信箱里的接收时间，来自库方时钟、可能为 NULL；只作对账与展示 + * @param enqueuedAt 本地入队时间；不传时由实现用自己注入的时钟填（列非空） * @return true 表示这次真的新建了一行 */ - fun insertIfAbsent(msgId: Long, receivedAt: Instant?): Boolean + fun insertIfAbsent(msgId: Long, receivedAt: Instant?, enqueuedAt: Instant? = null): Boolean fun find(msgId: Long): ProcState? @@ -99,11 +100,13 @@ interface ProcStateRepository { fun recordBackfillFailure(msgId: Long, error: String?, attempts: Int, nextAttemptAt: Instant, now: Instant) /** - * 放弃回填：判定该行不必再自动尝试（信箱行不存在，或达到尝试上限）。 + * 放弃回填：判定该行不必再自动尝试（信箱行不存在，或暂时性故障持续到 `R` 仍未打标）。 * * **abandoned ≠ 标记已确认**：`BACKFILL_AT` 仍为空，因此**不**满足"边界内全部行已打标"的 * 清除前提；放弃只是停止自动重试并把事实留痕，供人工对账。 * + * 放弃判据是**时间**（`R` 超期）而不是尝试次数：一次小时级的共享库故障不该把待回填行成批判死。 + * * @return false 表示该行不存在 */ fun markBackfillAbandoned(msgId: Long, reason: String, now: Instant): Boolean @@ -117,8 +120,9 @@ interface ProcStateRepository { /** * 找出现在该回填的记录：已经到终态、还没确认回填，并且退避时间已到。 * - * [overdueBefore] 是兜底：消息接收时间早于它的（已经等了很久）无视退避直接补写。 + * [overdueBefore] 是兜底：入队时间早于它的（已经等了超过 `R`）无视退避直接补写。 * 没有这条兜底，退避一直失败的话这些行就永远打不上标记，库方也没法清理信箱。 + * 判据比较的是**本地** `ENQUEUED_AT`，不是库方时钟的 `RECEIVED_AT`（`PRE-4`）。 */ fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List @@ -132,8 +136,12 @@ interface ProcStateRepository { fun hasAny(): Boolean } -/** 扫描到的待回填记录。 */ -data class BackfillDue(val msgId: Long, val attempts: Int) +/** + * 扫描到的待回填记录。 + * @param overdue 入队时间早于 `NOW − R`：本轮已经进入强补写窗口 + * （调用方据此决定"暂时性故障是否到放弃期限"） + */ +data class BackfillDue(val msgId: Long, val attempts: Int, val overdue: Boolean = false) /** * 处理侧积压快照。 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 090c85e..6319def 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 @@ -76,16 +76,20 @@ class JdbcProcStateRepository( private val clock: Clock, ) : ProcStateRepository { /** 入队（幂等）：主键冲突时什么都不做，所以重复扫描和兼容入口并发调用都安全。 */ - override fun insertIfAbsent(msgId: Long, receivedAt: Instant?): Boolean = - ds.update( - "INSERT INTO proc_state (msg_id, state, received_at, updated_at) VALUES (?, 'PENDING', ?, ?) " + - "ON CONFLICT (msg_id) DO NOTHING", + override fun insertIfAbsent(msgId: Long, receivedAt: Instant?, enqueuedAt: Instant?): Boolean { + val now = clock.instant() + return ds.update( + "INSERT INTO proc_state (msg_id, state, received_at, enqueued_at, updated_at) " + + "VALUES (?, 'PENDING', ?, ?, ?) ON CONFLICT (msg_id) DO NOTHING", { ps -> ps.setLong(1, msgId) ps.setTimestamp(2, receivedAt?.toSqlTimestamp()) - ps.setTimestamp(3, clock.instant().toSqlTimestamp()) + // 入队时间必须由本系统时钟给出：它同时是超期判据 R 的比较对象（PRE-4）。 + ps.setTimestamp(3, (enqueuedAt ?: now).toSqlTimestamp()) + ps.setTimestamp(4, now.toSqlTimestamp()) }, ) == 1 + } override fun find(msgId: Long): ProcState? = ds.queryOne("$SELECT_PROC WHERE msg_id = ?", { ps -> ps.setLong(1, msgId) }, ::mapProcState) @@ -254,22 +258,29 @@ class JdbcProcStateRepository( * 排序用**公平轮转**：先按已尝试次数升序，再按 msg_id。若只按 msg_id 升序， * 最旧的一批永久失败行会持续占满批次，后面的记录永远轮不到（全局回填饥饿）。 */ + /** + * 超期判据用 `enqueued_at`（本地时钟、非空），不用 `received_at`： + * 后者来自库方时钟，前偏会让 `NOW − R` 提前成立（`PRE-4`）。 + */ override fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List = ds.query( """ - SELECT msg_id, backfill_attempts FROM proc_state + SELECT msg_id, backfill_attempts, (enqueued_at < ?) AS overdue FROM proc_state WHERE backfill_at IS NULL AND backfill_abandoned_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 < ?)) + AND (backfill_next_at IS NULL OR backfill_next_at <= ? OR enqueued_at < ?) ORDER BY backfill_attempts ASC, msg_id ASC LIMIT ? """.trimIndent(), { ps -> - ps.setTimestamp(1, now.toSqlTimestamp()) - ps.setTimestamp(2, overdueBefore.toSqlTimestamp()) - ps.setInt(3, limit) + ps.setTimestamp(1, overdueBefore.toSqlTimestamp()) + ps.setTimestamp(2, now.toSqlTimestamp()) + ps.setTimestamp(3, overdueBefore.toSqlTimestamp()) + ps.setInt(4, limit) }, - ) { rs -> BackfillDue(rs.getLong("msg_id"), rs.getInt("backfill_attempts")) } + ) { rs -> + BackfillDue(rs.getLong("msg_id"), rs.getInt("backfill_attempts"), rs.getBoolean("overdue")) + } override fun requeueByErrorClasses(errorClasses: List): Int { if (errorClasses.isEmpty()) return 0 @@ -321,6 +332,7 @@ class JdbcProcStateRepository( errorClass = rs.getString("error_class")?.let(ErrorClass::valueOf), lastError = rs.getString("last_error"), receivedAt = rs.getInstant("received_at"), + enqueuedAt = rs.getInstant("enqueued_at"), backfillAt = rs.getInstant("backfill_at"), backfillNextAt = rs.getInstant("backfill_next_at"), backfillAttempts = rs.getInt("backfill_attempts"), @@ -334,7 +346,7 @@ class JdbcProcStateRepository( 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, " + + "received_at, enqueued_at, backfill_at, backfill_next_at, backfill_attempts, backfill_error, " + "backfill_abandoned_at, backfill_abandoned_reason, processing_started_at, updated_at FROM proc_state" } } 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 7f4e9c5..3a84931 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 @@ -63,9 +63,17 @@ class StubProcState : ProcStateRepository { rows.clear(); bound.clear() } - override fun insertIfAbsent(msgId: Long, receivedAt: Instant?): Boolean { + /** + * 内存版没有独立时钟，入队时间默认沿用 `receivedAt`（测试可显式传入以区分两者）； + * 生产实现由注入的 Clock 填，`ENQUEUED_AT` 列非空。 + */ + override fun insertIfAbsent(msgId: Long, receivedAt: Instant?, enqueuedAt: Instant?): Boolean { if (rows.containsKey(msgId)) return false - rows[msgId] = ProcState(msgId, ProcStatus.PENDING, receivedAt = receivedAt) + rows[msgId] = ProcState( + msgId, ProcStatus.PENDING, + receivedAt = receivedAt, + enqueuedAt = enqueuedAt ?: receivedAt, + ) return true } @@ -182,14 +190,13 @@ class StubProcState : ProcStateRepository { override fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List = rows.values .filter { it.state.isTerminal() && it.backfillAt == null && it.backfillAbandonedAt == null } - .filter { - it.backfillNextAt == null || it.backfillNextAt <= now || - (it.receivedAt != null && it.receivedAt < overdueBefore) - } + // 超期判据是本地入队时间，与 JDBC 实现保持一致（不用库方时钟的 receivedAt）。 + .map { it to (it.enqueuedAt?.let { e -> e < overdueBefore } ?: false) } + .filter { (row, overdue) -> overdue || row.backfillNextAt == null || row.backfillNextAt <= now } // 公平轮转：先按已尝试次数，再按 msg_id。只按 msg_id 会让最旧的一批永久失败行占满批次。 - .sortedWith(compareBy({ it.backfillAttempts }, { it.msgId })) + .sortedWith(compareBy({ it.first.backfillAttempts }, { it.first.msgId })) .take(limit) - .map { BackfillDue(it.msgId, it.backfillAttempts) } + .map { (row, overdue) -> BackfillDue(row.msgId, row.backfillAttempts, overdue) } override fun hasAny(): Boolean = rows.isNotEmpty() 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 a7dd384..4acb102 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt @@ -77,7 +77,8 @@ class InboxPoller( val enqueued = txManager.inTransaction { var n = 0 rows.takeWhile { it.msgId <= committedTo }.forEach { row -> - if (procState.insertIfAbsent(row.msgId, row.receivedAt)) n++ + // 入队时间用本轮注入的本地时钟，与超期判据 R 同源（PRE-4）。 + if (procState.insertIfAbsent(row.msgId, row.receivedAt, enqueuedAt = now)) n++ } cursor.save(InboxCursorRepository.Cursor(committedTo, holeSince)) n 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 1ec9457..e132113 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxService.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxService.kt @@ -14,7 +14,7 @@ import java.time.Instant * 收报轮询会按 ID 把它补进来，所以不会丢消息。 * * 该入口**不参与水位**：它直接写 PROC_STATE，登记的行可能超出水位；主泵只领 `msgId ≤ W`， - * 因此不破坏 FIFO——这类行等水位追平后按序自然领取（message-lifecycle.md §5.1）。 + * 因此不破坏 FIFO——这类行等水位追平后按序自然领取（`invariants.md` INV-4）。 */ @Singleton class InboxService( @@ -35,8 +35,10 @@ class InboxService( if (mailboxReceivedAt == null) { log.warn("mailbox receive time missing msgId={}, falling back to enqueue time", id) } - val receivedAt = mailboxReceivedAt ?: clock.instant() - procState.insertIfAbsent(id, receivedAt) + val now = clock.instant() + val receivedAt = mailboxReceivedAt ?: now + // 入队时间单独传本地时钟：它是超期判据 R 的比较对象，不能借用库方时间（PRE-4）。 + procState.insertIfAbsent(id, receivedAt, enqueuedAt = now) log.info("compat-accepted msgId={}", id) return Receipt(id, receivedAt) } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/BackfillService.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/BackfillService.kt index 2da61a0..6a9ae2b 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/BackfillService.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/BackfillService.kt @@ -19,9 +19,12 @@ import java.time.Instant * 这个类负责第二件： * * - [attempt]：单行写标记，只由 [sweep] 逐行调用——主泵与处理器不直接调它（回填一律扫描驱动）。 - * - [sweep]：定时把还欠回填的记录挑出来重试。失败就按 30 秒起步、最长 15 分钟的 - * 退避往后推；如果一条消息从收到现在已经超过超期期限，则无视退避强制补写—— - * 否则退避可能一直失败下去，这些行永远打不上标记，库方就没法清理信箱。 + * - [sweep]：定时把还欠回填的记录挑出来重试。失败按 30 秒起步、最长 15 分钟的退避往后推； + * 入队时间早于 `NOW − R` 的行无视退避、每轮都试。 + * + * 放弃判据是**时间**（`R` 超期）而不是尝试次数：一次小时级的共享库故障不该把待回填行成批 + * 判死、再要求人工成批恢复。`R` 之前只退避重试；到 `R` 仍未打标才停止自动重试并进放弃清单 + * （保留 [reopen] 人工恢复）。`backfill-max-attempts` 因此降级为单行重试的告警阈值。 * * 两条底线：回填失败不会把终态改回去，也不会重新执行业务逻辑；写标记只写还是空标记的 * 行，重复执行没有副作用。 @@ -46,8 +49,8 @@ class BackfillService( /** 放弃原因：运行时查询确认信箱行不存在（确定性结论，重试不会改变结果）。 */ const val ABANDON_MISSING_ROW = "MISSING_ROW" - /** 放弃原因：暂时性故障达到尝试上限；停止自动重试，但保留人工恢复能力。 */ - const val ABANDON_MAX_ATTEMPTS = "MAX_ATTEMPTS" + /** 放弃原因：暂时性故障持续到 `R` 仍未打标；停止自动重试，但保留人工恢复能力。 */ + const val ABANDON_TRANSIENT_DEADLINE = "TRANSIENT_DEADLINE" fun backoffDelayFor(attempts: Int): Duration { val shift = (attempts - 1).coerceIn(0, 20) @@ -59,13 +62,17 @@ class BackfillService( * 处理完立刻试一次。失败只记一笔退避信息就返回，不抛异常—— * 调用方是主泵的处理路径，不能被回填问题拖住。 */ - fun attempt(msgId: Long, now: Instant = clock.instant()) { + /** + * @param overdue 该行入队时间早于 `NOW − R`（已进入强补写窗口）。通常来自 [sweep] 的扫描结果； + * 直接调用默认 false，即按普通退避处理。 + */ + fun attempt(msgId: Long, overdue: Boolean = false, now: Instant = clock.instant()) { lifecycleGate.exclusive { val row = procState.find(msgId) ?: return@exclusive if (row.state !in TERMINAL_STATES) return@exclusive if (row.backfillAt != null) return@exclusive if (row.backfillAbandonedAt != null) return@exclusive - record(msgId, attempts = row.backfillAttempts, now = now)?.let { + record(msgId, attempts = row.backfillAttempts, overdue = overdue, now = now)?.let { log.warn("backfill failed msgId={} error={} (sweep will retry)", msgId, it) } } @@ -89,12 +96,12 @@ class BackfillService( */ fun sweep(now: Instant = clock.instant()): Int { val due = procState.findBackfillDue(now, now.minus(props.pipeline.overdueBackfill), props.pipeline.backfillBatch) - due.forEach { attempt(it.msgId, now) } + due.forEach { attempt(it.msgId, it.overdue, now) } return due.size } /** 回填一条。@return 失败原因；返回 null 表示已处理完（写成功／早已标记／已放弃）。 */ - private fun record(msgId: Long, attempts: Int, now: Instant): String? = + private fun record(msgId: Long, attempts: Int, overdue: Boolean, now: Instant): String? = try { when (mailbox.markProcessedIfUnmarked(msgId, mailboxProps.processedValue)) { MailboxMarkResult.MARKED, MailboxMarkResult.ALREADY_MARKED -> { @@ -111,18 +118,25 @@ class BackfillService( } } } catch (e: Exception) { - // 超时/连接失败是暂时性的，**不能**当作缺行证据：按退避重试。 - // 达到上限后停止自动重试（保留人工恢复能力），避免永久占满扫描批次造成饥饿。 + // 超时/连接失败是暂时性的，**不能**当作缺行证据，也不能按次数放弃： + // 按退避重试，直到入队时间超过 R 才停止自动重试（保留人工恢复能力）。 val reason = e.message ?: e.javaClass.simpleName val nextAttempts = attempts + 1 - if (nextAttempts >= props.pipeline.backfillMaxAttempts) { - runCatching { procState.markBackfillAbandoned(msgId, ABANDON_MAX_ATTEMPTS, now) } + if (overdue) { + runCatching { procState.markBackfillAbandoned(msgId, ABANDON_TRANSIENT_DEADLINE, now) } .onFailure { log.error("abandon backfill failed msgId={}", msgId, it) } - log.error("backfill abandoned after {} attempts msgId={} error={}", nextAttempts, msgId, reason) + log.error("backfill abandoned at R msgId={} attempts={} error={}", msgId, nextAttempts, reason) } else { runCatching { procState.recordBackfillFailure(msgId, reason, nextAttempts, now.plus(backoffDelayFor(nextAttempts)), now) }.onFailure { log.error("record backfill failure failed msgId={}", msgId, it) } + if (nextAttempts >= props.pipeline.backfillMaxAttempts) { + log.warn( + "backfill retries for msgId={} reached the warning threshold ({}); " + + "still retrying until R elapses — 放弃判据是 R 超期，不是次数", + msgId, props.pipeline.backfillMaxAttempts, + ) + } } reason } 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 f7ca06f..f6d5ccd 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt @@ -25,8 +25,7 @@ import java.util.concurrent.atomic.AtomicLong * * 每次 tick 只看当前最小的未完成消息（"队头"）： * - 没有待处理消息就睡一个轮询间隔； - * - 队头失败了还在退避期，就等到能重试的时刻；如果重试次数用尽或滞留太久， - * 直接转死信，不放任它一直堵着； + * - 队头失败了还在退避期，就等到能重试的时刻；重试次数用尽才转死信，不放任它一直堵着； * - 其余情况交给 [MessageProcessor] 处理。 * * 一次只处理一条是刻意的。后面的消息不能越过卡住的队头，否则同一条航班的报文 @@ -81,38 +80,36 @@ class Pump( // // 水位以内的行都是收报按 ID 顺序发现并登记的；水位之外的行只可能来自兼容入口 // 直接写 PROC_STATE（它不参与水位）。若允许领取，它就会越过那些尚未入队的较小 ID， - // 破坏 FIFO（不变量"只领取已发现的行"，message-lifecycle.md §11）。这种行在空洞补齐、`W` 追平之后自然可领取。 + // 破坏 FIFO（不变量"只领取已发现的行"，`invariants.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 && poisoned(head) -> { + head.state == ProcStatus.FAILED && head.attempts >= props.pipeline.maxAttempts -> { log.error("poison -> DEAD msgId={} attempts={} lastError={}", head.msgId, head.attempts, head.lastError) // markTerminal 在同一条 UPDATE 里登记回填意图；回填由扫描补写，不在这里做跨库写。 procState.markTerminal( head.msgId, ProcStatus.DEAD, errorClass = ErrorClass.EXHAUSTED, - lastError = head.lastError ?: "head-deadline-exceeded", + lastError = head.lastError ?: "attempts-exhausted", attempts = head.attempts, - now = clock.instant(), + now = now, ) } - head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > clock.instant() -> - sleepQuietly(Duration.between(clock.instant(), head.nextAttemptAt)) + head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > now -> + sleepQuietly(Duration.between(now, head.nextAttemptAt)) // 其余情况（新消息，或退避到期的重试）交给处理入口 else -> { - procState.markProcessingStartedIfAbsent(head.msgId, clock.instant()) + procState.markProcessingStartedIfAbsent(head.msgId, now) processor.processOne(head) } } } - private fun poisoned(head: ProcState): Boolean = - isHeadPoisoned(head, clock.instant(), props) - /** 上一次"队头在水位之外"告警时的水位值：只在它变化时告警，避免每秒刷屏。 */ private val warnedWatermark = AtomicLong(Long.MIN_VALUE) @@ -131,10 +128,6 @@ class Pump( } } -internal fun isHeadPoisoned(head: ProcState, now: Instant, props: PipelineProps): Boolean = - head.attempts >= props.pipeline.maxAttempts || - Duration.between(head.processingStartedAt ?: head.updatedAt, now) >= props.pipeline.headDeadline - /** * 处理一条消息：读原文 → 解码 → 绑定业务身份 → 分派给对应处理器。 * diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 41f142e..4995db8 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -18,11 +18,10 @@ msgx: max-attempts: 5 # 处理/投递同值 backoff-ms: [1000, 2000, 4000, 8000] # 指数退避；档位数必须 = max-attempts - 1（启动自检） backoff-cap-ms: 60000 - head-deadline: 10m # 队头滞留上界 = 最坏 HOL 时长（毒丸升级） max-commit-delay: 5m # §5.1 空洞老化：W+1 空洞超过该时延判定为永久（Q2 最大提交时延） - overdue-backfill: 30d # §5.2 超期补写期限 R（Q6）：仅须 R ≤ R_keep，不保护重放窗口（message-lifecycle §5.2） + overdue-backfill: 30d # 超期补写期限 R（Q6）：仅须 R ≤ R_keep，不保护重放窗口；判据比较本地 ENQUEUED_AT（design「回填」） backfill-batch: 100 # 回填扫描单批条数 - backfill-max-attempts: 100 # 回填自动重试上限；达上限停止自动重试（可人工恢复），避免占满批次 + backfill-max-attempts: 100 # 单行重试的告警阈值；放弃判据是 R 超期，不是次数（design「回填」） # 一次性切流播种：默认（注释掉）不播种。min=读现存全部 | zero=从 0 按空洞规则 | max=跳过可见存量 | # cutover-watermark: min late-detect-period: 60s # 迟到到达检测（只读，阶段 0）：复查被放行的空洞 ID 是否后来真的出现；0=关闭 diff --git a/src/main/resources/db/migration/V6__enqueued_at.sql b/src/main/resources/db/migration/V6__enqueued_at.sql new file mode 100644 index 0000000..ca95338 --- /dev/null +++ b/src/main/resources/db/migration/V6__enqueued_at.sql @@ -0,0 +1,22 @@ +-- ===================================================================== +-- V6：PROC_STATE 增加本地入队时间 ENQUEUED_AT +-- --------------------------------------------------------------------- +-- 只动自有 PostgreSQL；共享 MySQL 不建表、不改结构。 +-- +-- 背景（门禁裁决 D6）：超期补写期限 `R` 原先比较 `RECEIVED_AT`，那是**库方写入的异地时钟**。 +-- 库方时钟前偏时 `RECEIVED_AT < NOW − R` 会提前成立，在「打标即可清除」语义下正好制造 +-- 原文被提前清除的丢失窗口（`PRE-4`）。改为比较本系统自己写入的 `ENQUEUED_AT`： +-- 本地、非空、与判据用的 `NOW` 同源。`RECEIVED_AT` 退回「仅用于对账与展示」，并允许为 NULL。 +-- +-- 存量行没有历史入队时间，只能一次性尽力回填：优先信箱接收时间，其次最后一次更新时间。 +-- 这是一次推断，不回写业务、不影响任何已有终态，也不改已发布的迁移历史。 +-- ===================================================================== + +ALTER TABLE PROC_STATE ADD COLUMN ENQUEUED_AT TIMESTAMP(6) WITH TIME ZONE; + +UPDATE PROC_STATE + SET ENQUEUED_AT = COALESCE(RECEIVED_AT, UPDATED_AT, now()) + WHERE ENQUEUED_AT IS NULL; + +ALTER TABLE PROC_STATE ALTER COLUMN ENQUEUED_AT SET NOT NULL; +ALTER TABLE PROC_STATE ALTER COLUMN ENQUEUED_AT SET DEFAULT now(); 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 791a3e3..7d67a35 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 @@ -83,10 +83,10 @@ class FlywayMigrationTest { assertEquals(setOf("flid", "operation_day", "state", "state_version", "last_msg_id"), cols) } - // 回填相关的列都落在 PROC_STATE 上（收信时间判断超期；abandoned 记录"停止自动重试"） + // 回填相关的列都落在 PROC_STATE 上（本地入队时间判断超期；abandoned 记录"停止自动重试"） 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', " + + "AND column_name IN ('received_at', 'enqueued_at', 'backfill_at', 'backfill_next_at', " + "'backfill_attempts', 'backfill_error', 'backfill_abandoned_at', " + "'backfill_abandoned_reason', 'processing_started_at')", ).use { rs -> @@ -94,13 +94,23 @@ class FlywayMigrationTest { while (rs.next()) cols.add(rs.getString("column_name")) assertEquals( setOf( - "received_at", "backfill_at", "backfill_next_at", "backfill_attempts", "backfill_error", - "backfill_abandoned_at", "backfill_abandoned_reason", "processing_started_at", + "received_at", "enqueued_at", "backfill_at", "backfill_next_at", "backfill_attempts", + "backfill_error", "backfill_abandoned_at", "backfill_abandoned_reason", + "processing_started_at", ), cols, ) } + // V6：入队时间是超期判据 R 的比较对象，必须非空（received_at 则允许为 NULL） + stmt.executeQuery( + "SELECT is_nullable FROM information_schema.columns " + + "WHERE table_name = 'proc_state' AND column_name = 'enqueued_at'", + ).use { rs -> + assertTrue(rs.next(), "V6 必须已加上 ENQUEUED_AT 列") + assertEquals("NO", rs.getString("is_nullable"), "ENQUEUED_AT 必须非空") + } + // 两张单行表（管道锁、收报水位）的种子数据都要在 stmt.executeQuery("SELECT count(*) FROM pipeline_lock WHERE lock_id = 1").use { rs -> assertTrue(rs.next()) 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 c508213..afd663e 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 @@ -100,7 +100,7 @@ class InboxLifecycleJdbcSqlTest { seed(2L, t0) proc.markTerminal(2L, ProcStatus.SUCCEEDED, now = t0) proc.recordBackfillFailure(2L, "mysql-down", 1, t0.plus(Duration.ofMinutes(15)), t0) - // ③ 退避还没到，但收信时间已经很久了：应当无视退避直接补写 + // ③ 退避还没到，但本地入队时间已经超过 R：应当无视退避直接补写 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) @@ -111,10 +111,13 @@ class InboxLifecycleJdbcSqlTest { 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 } + val due = proc.findBackfillDue(t0, t0.minus(Duration.ofDays(30)), limit = 100) + val ids = due.map { it.msgId } - assertEquals(listOf(1L, 3L), due) + assertEquals(listOf(1L, 3L), ids) assertEquals(listOf(1L), proc.findBackfillDue(t0, t0.minus(Duration.ofDays(30)), limit = 1).map { it.msgId }) + // overdue 标记只在入队时间早于 NOW − R 时为真；调用方据此决定暂时性故障是否到放弃期限。 + assertEquals(listOf(1L to false, 3L to true), due.map { it.msgId to it.overdue }) } @Test @@ -186,15 +189,18 @@ class InboxLifecycleJdbcSqlTest { assertEquals(MailboxMarkResult.MISSING, mailbox.markProcessedIfUnmarked(999L, "PROCESSED")) } + /** 播种一行：`enqueued_at` 与 `received_at` 取同一时刻（超期判据只看前者）。 */ 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, ?)", + "INSERT INTO proc_state (msg_id, state, received_at, enqueued_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.setTimestamp(4, receivedAt?.let { Timestamp.from(it) } ?: Timestamp.from(t0)) + ps.setTimestamp(5, Timestamp.from(t0)) ps.executeUpdate() } } @@ -219,6 +225,7 @@ class InboxLifecycleJdbcSqlTest { error_class VARCHAR(20), last_error VARCHAR(1000), received_at TIMESTAMP WITH TIME ZONE, + enqueued_at TIMESTAMP WITH TIME ZONE NOT NULL, backfill_at TIMESTAMP WITH TIME ZONE, backfill_next_at TIMESTAMP WITH TIME ZONE, backfill_attempts INT NOT NULL DEFAULT 0, 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 e64747e..1db0766 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 @@ -26,7 +26,11 @@ class ReplayServiceTest { ) } - override fun insertIfAbsent(msgId: Long, receivedAt: Instant?): Boolean = rows.putIfAbsent(msgId, ProcState(msgId, ProcStatus.PENDING, receivedAt = receivedAt)) == null + override fun insertIfAbsent(msgId: Long, receivedAt: Instant?, enqueuedAt: Instant?): Boolean = + rows.putIfAbsent( + msgId, + ProcState(msgId, ProcStatus.PENDING, receivedAt = receivedAt, enqueuedAt = enqueuedAt ?: 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 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 d44252b..3b61e10 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/BackfillServiceTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/BackfillServiceTest.kt @@ -301,25 +301,44 @@ class BackfillServiceTest { assertNotNull(proc.find(911L)!!.backfillNextAt) } + /** + * 门禁裁决 D1：暂时性故障**不按次数放弃**。即使超过 `backfill-max-attempts` 这个 + * 告警阈值，也继续退避重试——一次小时级的共享库故障不该把待回填行成批判死。 + */ @Test - fun `transient failures keep retrying until the attempt cap and stay recoverable`() { + fun `transient failures keep retrying past the warning threshold`() { props.pipeline.backfillMaxAttempts = 3 val proc = StubProcState() val mailbox = FakeMailbox(fail = true) succeeded(proc, 912L) val svc = service(proc, mailbox) - svc.attempt(912L) // 1 次：暂时性故障，只退避 - svc.attempt(912L) // 2 次 - assertNull(proc.find(912L)!!.backfillAbandonedAt) + repeat(5) { svc.attempt(912L) } - svc.attempt(912L) // 3 次：达到上限 → 停止自动重试 val row = proc.find(912L)!! - assertEquals(BackfillService.ABANDON_MAX_ATTEMPTS, row.backfillAbandonedReason) + assertEquals(5, row.backfillAttempts) + assertNull(row.backfillAbandonedAt) // 次数不是放弃判据 assertNull(row.backfillAt) + assertNotNull(row.backfillNextAt) // 仍在退避重试 + } - assertTrue(svc.reopen(912L)) // 人工恢复入口存在且有效 - assertNull(proc.find(912L)!!.backfillAbandonedAt) + /** 到 `R` 仍未打标（overdue）时，暂时性故障才停止自动重试并进放弃清单。 */ + @Test + fun `overdue transient failure is abandoned with the deadline reason and stays recoverable`() { + val proc = StubProcState() + val mailbox = FakeMailbox(fail = true) + succeeded(proc, 913L) + val svc = service(proc, mailbox) + + svc.attempt(913L, overdue = true) + + val row = proc.find(913L)!! + assertEquals(BackfillService.ABANDON_TRANSIENT_DEADLINE, row.backfillAbandonedReason) + assertNotNull(row.backfillAbandonedAt) + assertNull(row.backfillAt) // 放弃 ≠ 标记已确认 + + assertTrue(svc.reopen(913L)) // 人工恢复入口存在且有效 + assertNull(proc.find(913L)!!.backfillAbandonedAt) } @Test @@ -346,18 +365,20 @@ class BackfillServiceTest { } /** - * 语义固定（G4）：`RECEIVED_AT` 为 NULL 时"超期"分支不成立，`R` 兜底**不生效**， - * 该行只能靠退避重试。把这条钉住，避免以后误以为 `R` 一定能兜底。 + * 语义固定（门禁裁决 D6）：超期判据是**本地入队时间** `ENQUEUED_AT`，与 `RECEIVED_AT` 无关。 + * 上游没给接收时间（NULL）不再让 `R` 兜底失效——这正是引入 `ENQUEUED_AT` 要消除的窗口。 */ @Test - fun `null received time disables the overdue shortcut so only backoff applies`() { + fun `null received time no longer disables the overdue shortcut`() { val proc = StubProcState() - proc.insertIfAbsent(921L, null) // 上游未提供接收时间 + val old = t0.minus(props.pipeline.overdueBackfill).minusSeconds(60) + proc.insertIfAbsent(921L, null, enqueuedAt = old) // 上游未提供接收时间 proc.markTerminal(921L, ProcStatus.SUCCEEDED, now = t0) proc.recordBackfillFailure(921L, "mysql-down", attempts = 1, nextAttemptAt = t0.plusSeconds(3600), now = t0) - val due = proc.findBackfillDue(now = t0, overdueBefore = t0.plusSeconds(10_000), limit = 10) + val due = proc.findBackfillDue(now = t0, overdueBefore = t0.minus(props.pipeline.overdueBackfill), limit = 10) - assertTrue(due.isEmpty()) // 超期分支无效 + 退避未到期 + assertEquals(listOf(921L), due.map { it.msgId }) // 退避未到期，但入队时间已超 R → 仍被扫描 + assertTrue(due.single().overdue) } } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/PumpDeadlineTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/PumpDeadlineTest.kt deleted file mode 100644 index 9169249..0000000 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/PumpDeadlineTest.kt +++ /dev/null @@ -1,40 +0,0 @@ -package com.gzzn.omms.msgexchange.processing - -import com.gzzn.omms.msgexchange.config.PipelineProps -import com.gzzn.omms.msgexchange.domain.ProcState -import com.gzzn.omms.msgexchange.domain.ProcStatus -import org.junit.jupiter.api.Assertions.assertFalse -import org.junit.jupiter.api.Assertions.assertTrue -import org.junit.jupiter.api.Test -import java.time.Instant - -class PumpDeadlineTest { - private val props = PipelineProps() - private val started = Instant.parse("2026-09-08T03:00:00Z") - - @Test - fun `deadline uses stable processing start rather than refreshed update time`() { - val row = ProcState( - msgId = 1, - state = ProcStatus.FAILED, - attempts = 1, - processingStartedAt = started, - updatedAt = started.plusSeconds(590), - ) - - assertTrue(isHeadPoisoned(row, started.plusSeconds(600), props)) - } - - @Test - fun `head below attempt and time limits remains retryable`() { - val row = ProcState( - msgId = 1, - state = ProcStatus.FAILED, - attempts = 1, - processingStartedAt = started, - updatedAt = started.plusSeconds(590), - ) - - assertFalse(isHeadPoisoned(row, started.plusSeconds(599), props)) - } -}