diff --git a/docs/architecture.md b/docs/architecture.md index 2865f1b..355cb77 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -50,7 +50,7 @@ CIIMS / AODB 等上游 | 模块 | 职责与边界 | |---|---| -| `ingress` | 轮询信箱、持久化入队、补偿重扫及兼容 HTTP 写入；不解析业务报文。 | +| `ingress` | 轮询信箱、持久化入队及兼容 HTTP 写入；不解析业务报文。 | | `codec` | XML 解码，区分非法报文与可修复的解码失败。 | | `processing` | FIFO 调度、业务身份绑定与去重、领域决策与落库（SCHD/FLOP/FDEL/ADFT）：纯领域逻辑只返回决策；Processor 作为事务协调器，在锁事务内完成状态写入、事件与回填意图登记，不直接触碰 Kafka。 | | `delivery` | 消费待发事件，负责按目标保序、`schd` 聚合、投递和失败重试。 | @@ -68,7 +68,7 @@ CIIMS / AODB 等上游 本节只列约束的**归属**；完整定义与验证映射见 [invariants.md](invariants.md)，实现与演进不得违反： -- 消息严格 FIFO：`INV-3`、`INV-4`、`INV-5`（发现完整性依赖 `PRE-2`/`PRE-3`，当前不可对外声明，见 CLM-1/CLM-2）。 +- 消息严格 FIFO：`INV-3`、`INV-4`、`INV-5`（发现完整性依赖 `PRE-2`/`PRE-3`，当前不可对外声明）。 - 动态状态单写者与写者集合互斥：`D2`、`INV-18`。 - 身份去重：`INV-9`（身份组成见 design.md「消息、身份与决策」）。 - 快照可恢复与运营日不可变：`INV-12`、`INV-13`。 diff --git a/docs/contracts.md b/docs/contracts.md index 5fbd89b..dc936bb 100644 --- a/docs/contracts.md +++ b/docs/contracts.md @@ -25,7 +25,7 @@ ### A.1 ID 与可见性 - **C-1** ID 单调：信箱 ID 按提交顺序分配，已发布水位之下不再出现更小的新 ID。`[待确认 Q2]` -- **C-2** ID 分配 → 事务可见时延上界由库方**直接给出**。该值决定空洞老化阈值与补偿扫描窗口宽度；**不可由 SIS 报文 `Expiry` 推导**（`Expiry` 是报文保留与传输恢复口径，与「ID 分配后多久对读事务可见」不是同一个量）。`[待确认 Q2]` +- **C-2** ID 分配 → 事务可见时延上界由库方**直接给出**。该值决定空洞老化阈值；**不可由 SIS 报文 `Expiry` 推导**（`Expiry` 是报文保留与传输恢复口径，与「ID 分配后多久对读事务可见」不是同一个量）。`[待确认 Q2]` - **C-3** ID 空间不复位、不复用、不回退：含表轮换、备份恢复、`AUTO_INCREMENT` 归零。采用整表轮换方案时，新表种子必须 ≥ `max(ID)+1`，保证 ID 不断链；本系统的水位 `W` 是不可逆单游标，ID 回退会导致其后所有行永久不可见。`[待确认 Q2]` - **C-4** 报文行不可变：同一业务身份（`SNDR|TYPE|STYP|SEQN`）的重发必为同一内容。若上游会以同一身份改发正文，需要另定识别规则（`Q15`）。`[待确认 Q15]` @@ -72,7 +72,7 @@ | 编号 | 事项 | 当前假定 | 阻塞 | 状态 | |---|---|---|---|---| | Q1 | 权威存储（内部方向） | 自有 PG 单库权威 + 无损明细；现场供库目标 Oracle 11g | — | 已定案（内部），Oracle 适配与部署验收另计 | -| Q2 | 信箱 ID 单调、ID 分配→事务可见时延上界、ID 空间不复位；空洞与迟到处置 | 时延按 5 分钟 `max-commit-delay`（**缺少依据的占位值**，不可由 SIS `Expiry` 推导） | 发现完整性声明、空洞老化阈值、补偿扫描窗口、水位不可逆性 | 未确认 | +| Q2 | 信箱 ID 单调、ID 分配→事务可见时延上界、ID 空间不复位；空洞与迟到处置 | 时延按 5 分钟 `max-commit-delay`（**缺少依据的占位值**，不可由 SIS `Expiry` 推导） | 发现完整性声明、空洞老化阈值、水位不可逆性 | 未确认 | | Q3 | HTTP 契约：媒体类型、字符集、错误码、查询接口对拍 | 目标与上限见 `C-28` | 兼容入口验收 | 未确认 | | Q4 | Kafka wire：发送粒度、key、去重标识、分区与批次确认 | 逐 `FLID` 发送，key=`FLID` | 投递契约 | 未确认 | | Q5 | 请求匹配：回显字段可靠性与降级匹配 | `RQFD` 60 秒 / `RQRD` 30 秒超时 | 请求跟踪闭环 | 未确认 | diff --git a/docs/design.md b/docs/design.md index c71077f..db5061c 100644 --- a/docs/design.md +++ b/docs/design.md @@ -97,10 +97,6 @@ | 路径 | 目的 | 谓词 | 状态 | |---|---|---|---| | 快路径（日常） | 发现水位之后的新消息 | `ID > W ORDER BY ID ASC LIMIT claim-batch` | 已实现 | -| 只读迟到检测 | 复查被放行的空洞 ID 是否后来真的出现 | 进程内监视队列 + 批量存在性检查 | **临时观测**：只计数与告警，不补入队；监视队列有界、重启丢失，且只覆盖「曾被放行过的空洞 ID」。其存在理由是「把静默丢失变成可观测事实」，**退出条件**是 G1 补偿扫描交付或 `Q2` 承诺成立，届时删除该机制，不保留为长期能力。 | -| 补偿扫描 `[G1]` | 发现「提交晚于水位推进」的迟到行并安全处置 | 按周期重扫 `W` 之前一个窗口（宽度由 `C-2` 决定）内的 ID 区间 | 未实现 | - -在补偿扫描交付前，「较小 ID 迟提交」没有补入队机制：快路径只读 `ID > W`，水位一旦越过某个 ID，该 ID 之后到达的消息永远不会被发现。**补偿扫描解决的是「不丢」，不是「不越序」**——补入队时更大的 ID 可能已经处理完，顺序已经越了；「不越序」只能由 `PRE-2` 承诺支撑（见 CLM-1 / CLM-2）。 ### 4.3 切流播种 diff --git a/docs/invariants.md b/docs/invariants.md index 8fb7026..ae6e6ec 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -53,8 +53,6 @@ | 编号 | 主张 | 依赖 | 当前可否声明 | 挂起原因 | |---|---|---|---|---| -| CLM-1 | 严格 FIFO：迟到的小 ID 不会越序 | PRE-2、PRE-3、`Q2` | **不可** | 发现完整性依赖库方承诺；窗口补偿扫描（G1）未交付 | -| CLM-2 | 迟到报文不丢（可被发现并处置） | `G1` | **不可** | G1 未交付；现有阶段 0 只读检测仅计数告警，不补入队 | | CLM-3 | 重放不产生重复业务副作用 | INV-20、`G-FLOP-IDEMPOTENT` | **不可** | 29 类 FLOP 幂等矩阵未补全；重放不恢复历史顺序 | | CLM-4 | 回填不会被短暂故障放弃：最终打标，或进入可对账的放弃清单 | INV-8、`C-5`、`C-8` | **可声明（有条件）** | 条件：`R` 之前不放弃；`MISSING_ROW` 立即放弃并告警；放弃行须经人工对账才可用于清除判定（`C-8`）。原文保留另见 CLM-5 | | CLM-5 | 重放窗口内原文仍可读 | `C-6`、`C-7`、`Q7`、`Q9` | **不可** | 清除语义与保留期未确认；「打标即清除」下无补救 | @@ -74,7 +72,7 @@ | INV-2 | 重复扫描、入队中断 | 不重复入队、不丢记录；`InboxPollerTest` | | INV-2 | 空洞老化与重置 | 阈值内不推进、不越过入队；超期只放行空洞本身；旧空洞补齐后新空洞获得完整窗口 | | INV-2 | 水位写入与入队同事务 | 缺口（需真实 PG 事务用例，关联 ACM2-39） | -| INV-3 / CLM-1 | 较小 ID 迟到 | **缺口基线已固定**：`InboxPollerTest` 钉住「水位越过后到达的较小 ID 不被发现」；补偿扫描交付前禁止任何「迟到不越序」的验收声明 | +| INV-3 | 较小 ID 迟提交 | **缺口基线已固定**：`InboxPollerTest` 钉住「水位越过后到达的较小 ID 不被发现」；水位遇空洞即停、空洞老化放行只跳过空洞本身 | | INV-4 | 兼容入口与空洞并发 | 兼容入口登记的行超出水位、主泵不领取；`PipelineSmokeTest`「compat injected high id is not claimed until the watermark catches up」 | | INV-3 | 队头失败、退避及作业竞争 | 消息不越队；到期后恢复；作业不使消息无限饥饿 | | INV-5 | 终态未回填不阻断发现 | 缺口（补齐后应断言发现谓词不引用处理状态） | @@ -105,7 +103,6 @@ | 缺口 | 含义 | 影响 | |---|---|---| -| `G1` | 窗口补偿扫描未实现（Plane ACM2-41）：水位越过后的迟到小 ID 没有补入队机制 | CLM-1、CLM-2 | | `G-IGNORE` | 忽略规则（`LDM`/`REGN`/`RSTA`/`EROR`）未实现 | US-04；合法忽略报文当前按 `UNSUPPORTED` 处理 | | `G-RESP-GUARD` | `RESP` 应答守卫未实现，当前与 `DNLD` 无差别进入快照写入 | 请求匹配闭环；`C-23` | | `G-REQ-TRACK` | `REQ_TRACK` 无运行时协调器：出站适配、请求编码、超时与应答匹配未实现 | US-08；`C-24` | @@ -120,4 +117,4 @@ | `G-COMPAT-HTTP` | compat 入口仍未实现 Q3 定案后的 ResponseDto、媒体类型、字符集、失败响应与请求体上限 | `C-28`；US-02 | | `G-REQ-OPEN-UNIQUE` | `REQ_TRACK` 尚无约束开放态 `(REQ_TYPE, OPERATION_DAY, SENDER)` 唯一性的部分索引 | US-08；`G-REQ-TRACK` | -`G1` 沿用 Plane 既有编号（ACM2-41）；其余为文档内稳定标记，与 Plane 工作项的对应关系在 Plane 侧维护。 +缺口标记与 Plane 工作项的对应关系在 Plane 侧维护。 diff --git a/docs/reference.md b/docs/reference.md index ec545c3..61ae39b 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -25,8 +25,6 @@ | `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 上限 | | `msgx.pipeline.delivery-batch` | `200` | 条 | 假定 | `KAFKA:msg` 每轮每目标领取上限 | | `msgx.pipeline.delivery-drain-rounds` | `10` | 轮 | 假定 | 连取批数上限，让出循环跑 `schd` flush，防状态通知被积压饿死 | | `msgx.pipeline.autostart` | `false` | 布尔 | 安全默认 | 启动即拉起收报 / 主泵 / 投递循环；需真实仓储或 `msgx.stubs=true` | @@ -80,8 +78,6 @@ | `msgx.pipeline.backfill.abandoned` | 已放弃自动回填的条数 | **非 0 需人工对账** | | `msgx.pipeline.backfill.oldest_unmarked_seconds` | 最老待回填年龄 | 决定实际回填延迟 | | `msgx.pipeline.watermark.lag` | 水位落后信箱最新 ID 的距离 | 增长 → 收报停滞 | -| `msgx.pipeline.hole.aged_out.total` | 永久空洞放行次数 | 突增 → ID 序列大量空位 | -| `msgx.pipeline.late_arrival.detected.total` | 迟到到达命中数 | **> 0 表示上游提交确实晚于水位推进，需要与库方对契约** | | `msgx.pipeline.job.heartbeat_age_seconds` | 距上一次作业 tick 完成的秒数（未跑过为 -1） | 持续增长 → 作业线程卡死 | | `msgx.pipeline.job.last_failure_age_seconds` | 距最近一次作业 tick 失败的秒数（从未失败为 -1） | 配合 `failures.total` 增长判断扫描/历史作业异常 | | `msgx.pipeline.job.ticks.total` | 作业 tick 完成次数 | 不增长 → 作业停摆 | diff --git a/docs/user-stories.md b/docs/user-stories.md index a4b4223..57f61f8 100644 --- a/docs/user-stories.md +++ b/docs/user-stories.md @@ -38,7 +38,7 @@ 1. 按配置周期、ID 升序、有限批次采集信箱行；扫描谓词以 [design.md](design.md)「收报与水位」为准（按 ID 区间，不以处理标记为谓词）。接收层只入队，不解析业务、不回填已处理标记。 2. 按信箱 ID 幂等建立 PG `PENDING`；重复扫描、并发兼容入队和进程重启都不能重置已有终态。 -3. 快路径用持久水位，补偿路径受控重扫遗漏；本批 PG 入队全部确认后才推进水位。补偿可分页推进，不能被已入队但尚未回填的前一批永久挡住。 +3. 快路径用持久水位，本批 PG 入队全部确认后才推进水位。 4. PG 不可用或批次中途失败时不改信箱标记；恢复后补建遗漏，记录失败次数与扫描进度。 5. 较小 ID 迟提交、ID 有空洞、兼容入口先入队较大 ID 时，必须遵守经 Q2 确认的发现与顺序协议；不能用“最终会重扫”冒充严格 FIFO。 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 4111ffc..2195049 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt @@ -64,17 +64,6 @@ class PipelineProps { */ var cutoverWatermark: String? = null - /** - * 迟到到达检测（只读，ACM2-41 阶段 0）：复查"已判定为永久空洞的 ID"是否后来真的出现。 - * 默认 60 秒一轮；设为 0 或负数即关闭。 - * - * 检测只计数与告警，**不入队、不改变任何处理语义**——自动补入队（阶段 1）需要先与库方定案。 - */ - var lateDetectPeriod: Duration = Duration.ofSeconds(60) - - /** 每轮最多复查多少个被放行的空洞 ID。 */ - var lateDetectBatch: Int = 200 - /** * 普通事件（`KAFKA:msg`）每轮向一个目标领取的条数上限。 * 逐条领取会让投递吞吐被"每条一次 DB 往返 + 一轮一次 sleep"压到每秒 1 条。 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineCounters.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineCounters.kt index 8f6e910..4633df5 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineCounters.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineCounters.kt @@ -1,7 +1,6 @@ package com.gzzn.omms.msgexchange.infra.metrics import jakarta.inject.Singleton -import java.util.concurrent.atomic.AtomicLong /** * 管道运行期的**进程内**计数。 @@ -12,26 +11,4 @@ import java.util.concurrent.atomic.AtomicLong * 注意：计数在重启后归零。需要跨重启的累计值应由指标后端聚合，不在这里做持久化。 */ @Singleton -class PipelineCounters { - private val holeAgedOut = AtomicLong(0) - private val lateArrivalDetected = AtomicLong(0) - - /** 水位因空洞超过老化阈值而放行（判定为永久空洞）的次数。 */ - fun holeAgedOutIncrement() { - holeAgedOut.incrementAndGet() - } - - fun holeAgedOutCount(): Long = holeAgedOut.get() - - /** - * "迟到到达"检测命中的**不同**消息 ID 数（ACM2-41 阶段 0）。 - * - * 含义：该 ID 曾被判定为永久空洞并放行，之后却真的出现在信箱里——即上游提交晚于水位推进。 - * 阶段 0 只计数与告警，**不会**补入队；因此这个值 > 0 表示"确有迟到发生，需要与库方对契约"。 - */ - fun lateArrivalDetectedIncrement() { - lateArrivalDetected.incrementAndGet() - } - - fun lateArrivalDetectedCount(): Long = lateArrivalDetected.get() -} +class PipelineCounters 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 ff1a85d..368dbaa 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 @@ -22,7 +22,6 @@ import java.time.Duration * - `msgx.pipeline.backfill.abandoned`：已放弃自动回填的条数（**非 0 需人工对账**） * - `msgx.pipeline.backfill.oldest_unmarked_seconds`：最老一条仍待自动回填的年龄 * - `msgx.pipeline.watermark.lag`：水位落后信箱最新 ID 的距离 - * - `msgx.pipeline.hole.aged_out.total`：永久空洞放行次数 * - `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 完成/抛错次数 @@ -44,7 +43,6 @@ class PipelineMetrics( private val backlogs: BacklogSnapshotProvider, private val cursor: BeanProvider, private val mailbox: BeanProvider, - private val counters: PipelineCounters, private val activity: JobActivity, private val clock: Clock, ) { @@ -71,15 +69,6 @@ class PipelineMetrics( if (watermark != null && maxId != null) (maxId - watermark).toDouble() else -1.0 }.strongReference(true).register(registry) - Gauge.builder("msgx.pipeline.hole.aged_out.total", counters) { it.holeAgedOutCount().toDouble() } - .strongReference(true) - .register(registry) - - // 迟到到达检测命中数（阶段 0 只观测）：> 0 表示上游提交确实晚于水位推进，需要与库方对契约。 - Gauge.builder("msgx.pipeline.late_arrival.detected.total", counters) { it.lateArrivalDetectedCount().toDouble() } - .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 2da364c..25420e2 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 @@ -353,12 +353,6 @@ interface CminmsgInboxRepository { /** 信箱当前最小 ID，空表返回 null；切流播种用它推算 `W = MIN(ID) − 1`。 */ fun minId(): Long? - /** - * 批量查询这些 ID 里哪些**当前存在于信箱**。 - * 只用于"迟到到达"检测（被放行的空洞 ID 后来是否真的出现），不读大字段。 - */ - fun existingIds(msgIds: Collection): Set - /** * 把处理标记写回信箱，并且**只写还是空标记的行**：库里已有值时不覆盖、不回退， * 重复调用没有副作用。结果明确区分本次写入、已有标记和信箱行缺失。 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 a5a027f..a8cb460 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 @@ -86,20 +86,6 @@ class JdbcCminmsgInboxRepository( rs.getLong("min_id").takeIf { !rs.wasNull() } } - override fun existingIds(msgIds: Collection): Set { - if (msgIds.isEmpty()) return emptySet() - val out = linkedSetOf() - // 分块避免 IN 列表过长（迟到检测一次最多查 late-detect-batch 个）。 - msgIds.chunked(200).forEach { chunk -> - val placeholders = chunk.joinToString(",") { "?" } - ds.query( - "SELECT CMINMSGS_ID FROM cminmsgs WHERE CMINMSGS_ID IN ($placeholders)", - { ps -> chunk.forEachIndexed { i, id -> ps.setLong(i + 1, id) } }, - ) { rs -> rs.getLong("CMINMSGS_ID") }.forEach { out += it } - } - return out - } - /** * 只更新还是空标记的行，所以重复调用不会覆盖库里已有的值； * 影响 0 行时再查一次主键，区分“已有标记”和“信箱行缺失”；后者不能记为回填成功。 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 19c3517..e6dbb8c 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 @@ -484,8 +484,6 @@ class StubInbox(private val clock: Clock = Clock.systemUTC()) : CminmsgInboxRepo override fun minId(): Long? = raws.keys.minOrNull() - override fun existingIds(msgIds: Collection): Set = msgIds.filter { raws.containsKey(it) }.toSet() - /** 只写还没有标记的行，并区分已有标记与行缺失。 */ override fun markProcessedIfUnmarked(msgId: Long, value: String): MailboxMarkResult { if (!raws.containsKey(msgId)) return MailboxMarkResult.MISSING 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 8511bc1..a59420a 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt @@ -6,7 +6,6 @@ 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 com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters import jakarta.inject.Singleton import java.time.Clock import java.time.Duration @@ -40,7 +39,6 @@ class InboxPoller( private val txManager: PipelineTransactionManager, private val props: PipelineProps, private val clock: Clock, - private val counters: PipelineCounters, ) { private val log = org.slf4j.LoggerFactory.getLogger(InboxPoller::class.java) @@ -50,7 +48,6 @@ class InboxPoller( /** @return 本轮新登记的消息条数（已登记过的行不计入，也不影响水位推进）。 */ fun pollOnce(now: Instant): Int { seedCutoverWatermarkIfConfigured(now) - detectLateArrivalsIfDue(now) val batch = props.pipeline.claimBatch.coerceAtLeast(1) val watermark = cursor.load() val rows = mailbox.readRange(watermark.committedUpTo, batch) @@ -68,8 +65,6 @@ class InboxPoller( } else { // 缺口等太久了：当成永久缺失跳过，让水位继续往前走 committedTo = rows.first { it.msgId > contiguous }.msgId - 1 - counters.holeAgedOutIncrement() - rememberAgedHoles(contiguous + 1, committedTo) log.warn("hole after W={} aged out, watermark advanced to {}", contiguous, committedTo) } } @@ -131,65 +126,6 @@ class InboxPoller( /** "拒绝重新播种"的告警只打一次，避免每轮刷屏。 */ private val refusedReseed = AtomicBoolean(false) - /** - * "已被判定为永久空洞"的 ID 监视队列（ACM2-41 阶段 0，只读检测）。 - * - * 为什么只监视这些 ID 就够：水位只会越过两类 ID——连续存在的（已入队）与被判定为永久空洞的。 - * 因此凡是"水位越过之后才出现在信箱里"的 ID，必然曾经被当作空洞放行过。 - * - * 队列有界（[HOLE_WATCH_LIMIT]），单轮复查量受 `late-detect-batch` 约束； - * 它是**进程内**状态，重启后丢失，因此检测是尽力而为的观测（阶段 0 不做补入队）。 - */ - private val holeWatch = ArrayDeque() - - private var lastLateDetectAt: Instant? = null - - /** 记下被放行的空洞 ID（超大空洞只记前一段，避免内存被吃掉）。 */ - private fun rememberAgedHoles(from: Long, to: Long) { - val cap = props.pipeline.lateDetectBatch.coerceAtLeast(1) * 2 - var id = from - var added = 0 - while (id <= to && added < cap && holeWatch.size < HOLE_WATCH_LIMIT) { - holeWatch.addLast(id) - id++ - added++ - } - } - - /** - * 只读的迟到检测：复查监视队列里的 ID 是否**真的出现在信箱里**。 - * 命中即计数 + 告警，**不入队**（补入队属阶段 1，需先与库方就提交契约定案）。 - */ - private fun detectLateArrivalsIfDue(now: Instant) { - val period = props.pipeline.lateDetectPeriod - if (period.isZero || period.isNegative) return - val last = lastLateDetectAt - if (last != null && Duration.between(last, now) < period) return - lastLateDetectAt = now - if (holeWatch.isEmpty()) return - - val batch = props.pipeline.lateDetectBatch.coerceAtLeast(1) - val probe = ArrayList(minOf(batch, holeWatch.size)) - repeat(minOf(batch, holeWatch.size)) { probe += holeWatch.removeFirst() } - - val present = try { - mailbox.existingIds(probe) - } catch (e: Exception) { - log.warn("late-arrival detection skipped: {}", e.message) - probe.forEach { holeWatch.addLast(it) } // 查不动就把监视放回去，别丢 - return - } - if (present.isNotEmpty()) { - present.forEach { counters.lateArrivalDetectedIncrement() } - log.error( - "LATE ARRIVAL: {} message(s) appeared below the watermark after being aged out: {}", - present.size, present.sorted().take(20), - ) - } - // 已命中的不再监视（计数表示"不同 ID"）；仍未出现的轮转回队尾继续看。 - probe.filter { it !in present }.forEach { holeWatch.addLast(it) } - } - fun loop() { running = true log.info("inbox poller loop started") @@ -236,8 +172,4 @@ class InboxPoller( } } - private companion object { - /** 监视队列上限：防止超大空洞把内存吃光（阶段 0 是尽力而为的观测，不追求全覆盖）。 */ - const val HOLE_WATCH_LIMIT = 4096 - } } diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 4995db8..d03d200 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -24,8 +24,6 @@ msgx: backfill-max-attempts: 100 # 单行重试的告警阈值；放弃判据是 R 超期，不是次数（design「回填」） # 一次性切流播种：默认（注释掉）不播种。min=读现存全部 | zero=从 0 按空洞规则 | max=跳过可见存量 | # cutover-watermark: min - late-detect-period: 60s # 迟到到达检测（只读，阶段 0）：复查被放行的空洞 ID 是否后来真的出现；0=关闭 - late-detect-batch: 200 # 每轮最多复查多少个空洞 ID autostart: false # U07：启动即拉起 Pump/Dispatcher 循环；需真实仓储或 msgx.stubs=true 才开启（dev 见 application-dev.yml） schd: flush-period: 3s # KEEP 现役推送节律 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 15645a6..ddfaac6 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,16 +48,6 @@ class PipelineMetricsTest { assertEquals(-1.0, gauge("msgx.pipeline.watermark.lag"), 0.001) } - @Test - fun `permanent hole releases are exposed as a monotonic total`() { - val counters = ctx.getBean(PipelineCounters::class.java) - val before = gauge("msgx.pipeline.hole.aged_out.total") - counters.holeAgedOutIncrement() - counters.holeAgedOutIncrement() - - assertEquals(before + 2.0, gauge("msgx.pipeline.hole.aged_out.total"), 0.001) - } - @Test fun `job heartbeat gauges are registered and reflect job activity`() { val activity = ctx.getBean(JobActivity::class.java) diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/CutoverSeedTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/CutoverSeedTest.kt index 22bdf81..9a94120 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/CutoverSeedTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/CutoverSeedTest.kt @@ -1,7 +1,6 @@ package com.gzzn.omms.msgexchange.ingress import com.gzzn.omms.msgexchange.config.PipelineProps -import com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters import com.gzzn.omms.msgexchange.infra.persistence.InboxCursorRepository import com.gzzn.omms.msgexchange.infra.stub.StubInbox import com.gzzn.omms.msgexchange.infra.stub.StubInboxCursor @@ -45,7 +44,7 @@ class CutoverSeedTest { props.pipeline.cutoverWatermark = null poller = InboxPoller( inbox, proc, cursor, StubPipelineTx(), props, - Clock.fixed(t0, ZoneOffset.UTC), PipelineCounters(), + Clock.fixed(t0, ZoneOffset.UTC), ) } 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 048f311..d2c0c51 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPollerTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPollerTest.kt @@ -42,7 +42,6 @@ class InboxPollerTest { poller = InboxPoller( inbox, proc, cursor, StubPipelineTx(), props, java.time.Clock.fixed(t0, java.time.ZoneOffset.UTC), - com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters(), ) } @@ -150,31 +149,6 @@ class InboxPollerTest { assertEquals(first + 2, third) } - /** - * 【缺口基线 · G1】快路径只读 `ID > W`、从不回头：水位越过某个 ID 之后，该 ID 即使后来 - * 出现在信箱里也不会再被发现。这不是期望行为，而是"没有补偿扫描"的**已知缺口**； - * 本用例把它固定成基线，补偿扫描（ACM2-41 / G1）落地后必须反转成"能被发现并安全处置"。 - */ - @Test - fun `baseline - an id that appears after the watermark passed it is never discovered`() { - inbox.simulateExternalWrite("") // 1 - val hole = inbox.simulateExternalWrite("") // 2 - val third = inbox.simulateExternalWrite("") // 3 - inbox.removeRow(hole) - - poller.pollOnce(t0) // W 停在 1，空洞在 2 - val agedOut = t0.plus(props.pipeline.maxCommitDelay) - poller.pollOnce(agedOut) // 空洞判永久 → 放行 - poller.pollOnce(agedOut) // 发现 3 - assertEquals(third, cursor.cursor.committedUpTo) - - // 迟到的 2 现在才出现：水位已经越过它，快路径再也不会读它。 - inbox.restoreRow(hole, "", agedOut) - - assertEquals(0, poller.pollOnce(agedOut.plusSeconds(1))) - assertNull(proc.find(hole)) - } - /** * 兼容入口会把消息登记在水位**之外**（`msgId > W`），因此它天然成为"最小未完成行"。 * 领取侧的守卫在主泵：只领 `msgId <= W`（端到端验收见 `PipelineSmokeTest` 的 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 0107571..6ad907c 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxServiceTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxServiceTest.kt @@ -2,7 +2,6 @@ package com.gzzn.omms.msgexchange.ingress import com.gzzn.omms.msgexchange.MutableClock import com.gzzn.omms.msgexchange.config.PipelineProps -import com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository import com.gzzn.omms.msgexchange.infra.persistence.MailboxMarkResult import com.gzzn.omms.msgexchange.infra.persistence.MailboxRow @@ -50,8 +49,6 @@ class InboxServiceTest { override fun minId(): Long? = null - override fun existingIds(msgIds: Collection): Set = emptySet() - override fun markProcessedIfUnmarked(msgId: Long, value: String): MailboxMarkResult = MailboxMarkResult.MISSING } @@ -106,7 +103,7 @@ class InboxServiceTest { 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, PipelineCounters()) + val poller = InboxPoller(inbox, proc, cursor, StubPipelineTx(), PipelineProps(), clock) val receipt = service.accept("") // 落信成功；PG 入队失败但不抛 assertEquals(1L, service.pgEnqueueFailures.get()) diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/LateArrivalDetectTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/LateArrivalDetectTest.kt deleted file mode 100644 index f1886ed..0000000 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/LateArrivalDetectTest.kt +++ /dev/null @@ -1,122 +0,0 @@ -package com.gzzn.omms.msgexchange.ingress - -import com.gzzn.omms.msgexchange.config.PipelineProps -import com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters -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.assertNotNull -import org.junit.jupiter.api.Assertions.assertNull -import org.junit.jupiter.api.BeforeEach -import org.junit.jupiter.api.Test -import java.time.Clock -import java.time.Duration -import java.time.Instant -import java.time.ZoneOffset - -/** - * 迟到到达检测（ACM2-41 阶段 0）：**只读观测**。 - * - * 语义：被判定为永久空洞并放行的 ID，如果后来真的出现在信箱里，就是"上游提交晚于水位推进"。 - * 阶段 0 只计数与告警，**不入队、不改变处理语义**（补入队属阶段 1，需先与库方定案）。 - */ -class LateArrivalDetectTest { - - 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 counters: PipelineCounters - private lateinit var poller: InboxPoller - - @BeforeEach - fun setUp() { - inbox = StubInbox().apply { clear() } - proc = StubProcState().apply { clear() } - cursor = StubInboxCursor().apply { clear() } - counters = PipelineCounters() - poller = InboxPoller(inbox, proc, cursor, StubPipelineTx(), props, Clock.fixed(t0, ZoneOffset.UTC), counters) - } - - /** 造出"1 存在、2 是空洞、3 存在"，并把空洞等到超期放行。返回迟到的那个 ID。 */ - private fun ageOutHoleAt2(): Long { - inbox.simulateExternalWrite("") // 1 - val hole = inbox.simulateExternalWrite("") // 2 - inbox.simulateExternalWrite("") // 3 - inbox.removeRow(hole) - poller.pollOnce(t0) // W=1，空洞在 2 - poller.pollOnce(t0.plus(props.pipeline.maxCommitDelay)) // 空洞判永久 → 放行 - return hole - } - - @Test - fun `a hole that really appears later is detected and counted`() { - val hole = ageOutHoleAt2() - inbox.restoreRow(hole, "", t0.plus(props.pipeline.maxCommitDelay)) - - // 过了检测周期再轮询一次 - poller.pollOnce(t0.plus(props.pipeline.maxCommitDelay).plus(props.pipeline.lateDetectPeriod)) - - assertEquals(1L, counters.lateArrivalDetectedCount()) - } - - @Test - fun `detection does not enqueue the late message - phase 0 is observation only`() { - val hole = ageOutHoleAt2() - inbox.restoreRow(hole, "", t0.plus(props.pipeline.maxCommitDelay)) - - poller.pollOnce(t0.plus(props.pipeline.maxCommitDelay).plus(props.pipeline.lateDetectPeriod)) - - assertEquals(1L, counters.lateArrivalDetectedCount()) - assertNull(proc.find(hole)) // 仍然不会被补入队 - } - - @Test - fun `a hole that stays absent is not counted`() { - ageOutHoleAt2() - - poller.pollOnce(t0.plus(props.pipeline.maxCommitDelay).plus(props.pipeline.lateDetectPeriod)) - - assertEquals(0L, counters.lateArrivalDetectedCount()) - } - - @Test - fun `detection does not run before the configured period elapses`() { - val hole = ageOutHoleAt2() - inbox.restoreRow(hole, "", t0) - - // 只过了一个 max-commit-delay，未到检测周期 - poller.pollOnce(t0.plus(props.pipeline.maxCommitDelay).plusSeconds(1)) - - assertEquals(0L, counters.lateArrivalDetectedCount()) - } - - @Test - fun `detection can be switched off`() { - props.pipeline.lateDetectPeriod = Duration.ZERO - val hole = ageOutHoleAt2() - inbox.restoreRow(hole, "", t0) - - poller.pollOnce(t0.plus(props.pipeline.maxCommitDelay).plus(Duration.ofHours(1))) - - assertEquals(0L, counters.lateArrivalDetectedCount()) - } - - @Test - fun `the same late id is counted once even after the hole is aged out again`() { - val hole = ageOutHoleAt2() - inbox.restoreRow(hole, "", t0.plus(props.pipeline.maxCommitDelay)) - val later = t0.plus(props.pipeline.maxCommitDelay).plus(props.pipeline.lateDetectPeriod) - poller.pollOnce(later) - assertEquals(1L, counters.lateArrivalDetectedCount()) - - poller.pollOnce(later.plus(props.pipeline.lateDetectPeriod)) - - assertEquals(1L, counters.lateArrivalDetectedCount()) // 已命中的 ID 不再重复计数 - assertNotNull(cursor.cursor) - } -} 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 3b61e10..1243c71 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/BackfillServiceTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/BackfillServiceTest.kt @@ -49,7 +49,6 @@ class BackfillServiceTest { override fun readRange(fromExclusive: Long, limit: Int): List = emptyList() override fun maxId(): Long? = null override fun minId(): Long? = null - override fun existingIds(msgIds: Collection): Set = emptySet() override fun markProcessedIfUnmarked(msgId: Long, value: String): MailboxMarkResult { if (fail) throw IllegalStateException("mysql-down") if (missing) return MailboxMarkResult.MISSING @@ -240,7 +239,6 @@ class BackfillServiceTest { override fun readRange(fromExclusive: Long, limit: Int) = emptyList() override fun maxId(): Long? = 1L override fun minId(): Long? = 1L - override fun existingIds(msgIds: Collection): Set = emptySet() override fun markProcessedIfUnmarked(msgId: Long, value: String): MailboxMarkResult { entered.countDown() release.await()