refactor(ingress): 移除 G1 窗口补偿扫描与迟到到达检测

实际负载不足 10 条/秒，G1 属过度防御。删除迟到到达检测机制、 existingIds 接口、PipelineCounters 字段、lateDetect 配置，以及文档中 CLM-1/CLM-2 声明与 G1 缺口索引。
2026-09-13 08:08:43 +08:00
parent a6b2120836
commit b8738554e6
20 changed files with 10 additions and 322 deletions
@@ -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 条。
@@ -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
@@ -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<InboxCursorRepository>,
    private val mailbox: BeanProvider<CminmsgInboxRepository>,
-    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
@@ -353,12 +353,6 @@ interface CminmsgInboxRepository {
    /** 信箱当前最小 ID，空表返回 null；切流播种用它推算 `W = MIN(ID) − 1`。 */
    fun minId(): Long?

-    /**
-     * 批量查询这些 ID 里哪些**当前存在于信箱**。
-     * 只用于"迟到到达"检测（被放行的空洞 ID 后来是否真的出现），不读大字段。
-     */
-    fun existingIds(msgIds: Collection<Long>): Set<Long>
-
    /**
     * 把处理标记写回信箱，并且**只写还是空标记的行**：库里已有值时不覆盖、不回退，
     * 重复调用没有副作用。结果明确区分本次写入、已有标记和信箱行缺失。
@@ -86,20 +86,6 @@ class JdbcCminmsgInboxRepository(
            rs.getLong("min_id").takeIf { !rs.wasNull() }
        }

-    override fun existingIds(msgIds: Collection<Long>): Set<Long> {
-        if (msgIds.isEmpty()) return emptySet()
-        val out = linkedSetOf<Long>()
-        // 分块避免 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 行时再查一次主键，区分“已有标记”和“信箱行缺失”；后者不能记为回填成功。
@@ -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<Long>): Set<Long> = msgIds.filter { raws.containsKey(it) }.toSet()
-
    /** 只写还没有标记的行，并区分已有标记与行缺失。 */
    override fun markProcessedIfUnmarked(msgId: Long, value: String): MailboxMarkResult {
        if (!raws.containsKey(msgId)) return MailboxMarkResult.MISSING
@@ -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<Long>()
-
-    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<Long>(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
-    }
 }
@@ -24,8 +24,6 @@ msgx:
    backfill-max-attempts: 100      # 单行重试的告警阈值；放弃判据是 R 超期，不是次数（design「回填」）
    # 一次性切流播种：默认（注释掉）不播种。min=读现存全部 | zero=从 0 按空洞规则 | max=跳过可见存量 | <id>
    # 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 现役推送节律