feat(ingress): 按处理时间为空扫描信箱，移除水位消费（ACM2-78）

readUnprocessed 替代 ID 水位；去掉 INBOX_CURSOR/cutover-watermark/max-commit-delay 及相关健康与指标。 Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-21 14:49:40 +08:00
parent 23b74ed554
commit 891088235c
29 changed files with 166 additions and 805 deletions
@@ -27,15 +27,6 @@ class PipelineProps {
        var backoffMs: List<Long> = 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|<id>, got '$cutover'"
-            }
        }
    }

@@ -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<ProcStateRepository>,
-    private val cursor: BeanProvider<InboxCursorRepository>,
-    private val mailbox: BeanProvider<CminmsgInboxRepository>,
    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<String, Any>(
            "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()
 }
@@ -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<InboxCursorRepository>,
-    private val mailbox: BeanProvider<CminmsgInboxRepository>,
    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
@@ -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<MailboxRow>
-
-    /** 信箱当前最大 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<MailboxRow>

    /**
     * 把处理标记写回信箱，并且**只写还是空标记的行**：库里已有值时不覆盖、不回退，
@@ -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<MailboxRow> =
+    override fun readUnprocessed(afterId: Long, limit: Int): List<MailboxRow> =
        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 行时再查一次主键，区分“已有标记”和“信箱行缺失”；后者不能记为回填成功。
@@ -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")
@@ -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<MailboxRow> =
-        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<MailboxRow> =
+        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
@@ -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<MailboxRow>): 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))
@@ -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))
    }