refactor(ingress): 信箱边界层契约重构——发现与处理标记解耦、回填事实并入 PROC_STATE

收报扫描不再以 DATE_PROCESSED 为谓词：终态而未回填的行（解码失败死信等）会永久占据有限批次，累积到 claim-batch 后收报整体停摆（US-01 条目 3 / message-lifecycle §5.3）。 ingress/InboxPoller.kt：按 ID 区间升序有界读取（ID > W），水位落 INBOX_CURSOR 并与入队同一 PG 事务推进（中断后重扫补建）；遇空洞即停，空洞超过 pipeline.max-commit-delay 判定为永久并放行——否则水位永久停摆于一次自增回滚留下的空位。删除 InboxEnqueue（改由 insertIfAbsent 幂等入队）与 poller 内的回填扫描（消除 ingress→jobs 反向依赖）。 infra/persistence：端口按事实重画为 readRange/maxId/markProcessedIfUnmarked，标记 UPDATE 带 DATE_PROCESSED IS NULL 守卫，只把空标写为已处理（§11 单调，重复执行无副作用）。回填事实并入 PROC_STATE（RECEIVED_AT/BACKFILL_AT/NEXT_AT/ATTEMPTS/ERROR），BACKFILL_TODO 随 V2 迁移下线；终态与回填意图是同一条 UPDATE，由处理器在自己的业务事务内落库， message-lifecycle §4 登记的两个崩溃窗口（提交后回填前崩溃、待办二次落账失败）不再是缺口。 processing/BackfillService.kt（取代 BackfillSweepJob）：终态提交后立即尝试一次，失败按 30s→15min 指数退避重试；扫描条件「终态 + 未确认标记 +（已到期或接收时间早于 NOW − R）」使 §5.2 的超期期限 R 覆盖退避，中间态永不补写。死信同样可补写——回填只需消息 ID，不再依赖 META。Pump 改用可注入 Clock。 infra/health：InboxLifecycleHealthIndicator 输出积压条数、最老未处理信龄、未回填终态数与水位滞后（OPS-2 / §5.3 验收）。预计消化时长需吞吐采样，留待接入指标注册表时补。配置：pipeline.max-commit-delay / overdue-backfill / backfill-batch、mailbox.processed-value （Q2/Q6/Q7 未书面确认前取保守初值，不得为提速下调）。不变量回归测试：死信不阻断后续发现、水位遇空洞即停与老化放行、终态+意图同事务、超期 R 覆盖退避、中间态不补写、标记单调；InboxLifecycleJdbcSqlTest 以 H2 的 PostgreSQL 兼容模式直连验证上述 SQL 语义（不依赖 docker）。libs.h2 由 testRuntimeOnly 提为 testImplementation 以支持该用例。验证：gradle clean test --offline → 78 tests / 0 failures / 1 skipped （PG Testcontainers 集成用例在本机无 docker 时按既有约定 assumeTrue 跳过）。
2026-09-10 10:59:58 +08:00
parent ffd3abd655
commit 09b53c77bb
32 changed files with 1402 additions and 548 deletions
@@ -0,0 +1,74 @@
+package com.gzzn.omms.msgexchange.infra.health
+
+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
+import io.micronaut.health.HealthStatus
+import io.micronaut.management.health.indicator.HealthIndicator
+import io.micronaut.management.health.indicator.HealthResult
+import jakarta.inject.Singleton
+import org.reactivestreams.Publisher
+import java.time.Duration
+import java.time.Instant
+
+/**
+ * 信箱生命周期观测（docs/message-lifecycle.md §5.3 验收 / user-stories.md OPS-2）：
+ * 输出剩余积压、最老未处理信龄、未回填终态条数与水位滞后，供积压消化期间持续观察。
+ *
+ * 端口缺省（未接通共享信箱或自有 PG）时报告未绑定而不判 DOWN——可用性由各依赖自身的
+ * 健康指示器承担，本指示器只反映生命周期状态；端口查询失败判 DOWN。
+ */
+@Singleton
+class InboxLifecycleHealthIndicator(
+    private val procState: BeanProvider<ProcStateRepository>,
+    private val cursor: BeanProvider<InboxCursorRepository>,
+    private val mailbox: BeanProvider<CminmsgInboxRepository>,
+) : HealthIndicator {
+
+    override fun getResult(): Publisher<HealthResult> =
+        Publishers.just(
+            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,
+                )
+            }.getOrElse { down(it) },
+        )
+}
+
+internal fun lifecycleHealth(
+    procState: ProcStateRepository?,
+    cursor: InboxCursorRepository?,
+    mailbox: CminmsgInboxRepository?,
+    now: Instant = Instant.now(),
+): HealthResult {
+    if (procState == null) {
+        return HealthResult.builder(NAME).status(HealthStatus.UP)
+            .details(mapOf("message" to "proc_state repository not bound (stub off, impl pending)"))
+            .build()
+    }
+    val 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 backlog.unfinished,
+            "oldestUnprocessedSeconds" to (
+                backlog.oldestReceivedAt?.let { Duration.between(it, now).seconds } ?: -1L
+                ),
+            "unmarkedTerminal" to backlog.unmarkedTerminal,
+            "watermark" to (watermark ?: -1L),
+            "watermarkLag" to if (watermark != null && maxId != null) maxId - watermark else -1L,
+        ),
+    ).build()
+}
+
+private fun down(error: Throwable): HealthResult =
+    HealthResult.builder(NAME).status(HealthStatus.DOWN)
+        .details(mapOf("message" to (error.message ?: error.javaClass.simpleName)))
+        .build()
+
+private const val NAME = "inbox-lifecycle"
@@ -29,12 +29,17 @@ interface PipelineLockRepository {
    fun lock()
 }

-/** PROC_STATE：每消息一行（design.md §2.1）；SUCCEEDED 终态兼作日计划重放判定。 */
+/**
+ * PROC_STATE：每消息一行（design.md §2.1）；SUCCEEDED 终态兼作日计划重放判定。
+ * 回填事实（BACKFILL_AT/NEXT_AT/ATTEMPTS/ERROR）与处理事实同行，取代独立的回填待办表。
+ */
 interface ProcStateRepository {
-    fun insert(msgId: Long, state: ProcStatus = ProcStatus.PENDING)
-
-    /** 主路径轮询/compat 入队前判重（PG 已有行则跳过）。 */
-    fun exists(msgId: Long): Boolean
+    /**
+     * 入队：MSG_ID 主键幂等（重复扫描与 compat 入口并发都不会重复建行）。
+     * @param receivedAt 信箱 DATE_RECEIVED，用于 §5.2 超期判据与信龄观测。
+     * @return true = 本次实际新建
+     */
+    fun insertIfAbsent(msgId: Long, receivedAt: Instant?): Boolean

    fun find(msgId: Long): ProcState?

@@ -48,6 +53,7 @@ interface ProcStateRepository {

    fun ownerOfIdentity(identityKey: String): Long?

+    /** 非终态迁移（PENDING / FAILED 及退避），不触碰回填列。 */
    fun update(
        msgId: Long,
        state: ProcStatus,
@@ -57,10 +63,44 @@ interface ProcStateRepository {
        lastError: String? = null,
    )

+    /**
+     * 终态 + 回填意图同一条 UPDATE（message-lifecycle §2/§4），由处理器在自己的业务
+     * 事务内调用：航班变更、事件、终态、回填意图同提交同回滚。
+     */
+    fun markTerminal(
+        msgId: Long,
+        state: ProcStatus,
+        errorClass: ErrorClass? = null,
+        lastError: String? = null,
+        attempts: Int? = null,
+        now: Instant = Instant.now(),
+    )
+
+    /** 信箱行已确认持有处理标记。 */
+    fun markBackfilled(msgId: Long, now: Instant = Instant.now())
+
+    /** 回填失败：次数 +1、按退避推后、留错误；终态不得回改（§11）。 */
+    fun recordBackfillFailure(msgId: Long, error: String?, attempts: Int, nextAttemptAt: Instant, now: Instant)
+
+    /**
+     * 待回填待办（message-lifecycle §3/§5.2）：终态 + 未确认标记，
+     * 且（已到期 或 接收时间已达超期期限 overdueBefore）。
+     */
+    fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List<BackfillDue>
+
    /** 显式重放入口：仅把给定 errorClass 集合中的行从 FAILED/DEAD 置回 PENDING（ATTEMPTS=0）。 */
    fun requeueByErrorClasses(errorClasses: List<ErrorClass>): Int
+
+    /** OPS-2 观测口径（message-lifecycle §5.3 验收）：积压、最老信龄锚点、未回填终态。 */
+    fun backlog(): Backlog
 }

+/** 某条待回填记录（扫描输入）。 */
+data class BackfillDue(val msgId: Long, val attempts: Int)
+
+/** 处理侧积压快照。 */
+data class Backlog(val unfinished: Int, val oldestReceivedAt: Instant?, val unmarkedTerminal: Int)
+
 /** MSG_EVENT outbox（flight-state.md §5）。KAFKA_SCHD 合并同 FLID 未发事件按最新 STATE_VERSION 输出。 */
 interface MsgEventRepository {
    fun insertAll(events: List<MsgEvent>): List<Long>
@@ -168,41 +208,39 @@ interface ReqTrackRepository {
 }

 /**
- * 共享信箱回填补偿待办（design.md §3.3/§6.1）：业务事务内预登记，
- * 提交后由 BackfillSweepJob 重试；回填失败不得把 SUCCEEDED 改回 FAILED。
+ * 消费水位 W（message-lifecycle §5.1）：单行游标，只随新 ID 成功入队推进、遇空洞即停，
+ * 与入队在同一 PG 事务提交——中断后 W 未前进，重扫即补建（§4 第一行）。
+ *
+ * `holeSince` 记录 W+1 处空洞首次被观测到的时刻：超过最大提交时延（Q2 承诺）即判定为
+ * 永久空洞并放行，否则水位会永久停摆于一次自增回滚留下的空位，后续 ID 再无入队机会。
 */
-interface BackfillTodoRepository {
-    data class BackfillTask(
-        val msgId: Long,
-        val sndr: String,
-        val type: String,
-        val styp: String,
-        val seqn: Long,
-        val attempts: Int = 0,
-    )
+interface InboxCursorRepository {
+    data class Cursor(val committedUpTo: Long = 0L, val holeSince: Instant? = null)

-    /** 失败即落库（幂等 upsert，同 ID 重复失败只刷新错误与重试时间）。 */
-    fun record(task: BackfillTask, lastError: String?, now: Instant = Instant.now())
+    fun load(): Cursor

-    fun findDue(now: Instant = Instant.now(), limit: Int = 50): List<BackfillTask>
-
-    fun markFailed(msgId: Long, lastError: String?, nextAttemptAt: Instant, now: Instant = Instant.now())
-
-    fun delete(msgId: Long)
-
-    fun count(): Int
+    fun save(cursor: Cursor)
 }

 /**
- * 共享 MySQL 信箱 CMINMSGS 访问（他人系统库，本系统不建表）。
- * 主路径 JDBC 轮询读 + 处理回填；compat HTTP 写；出站写 COUTMSGS 由出站适配层承担。
+ * 共享 MySQL 信箱 CMINMSGS 访问（他人系统库，本系统不建表，只做 DML）。
+ * 三个事实互不替代（message-lifecycle §5.1/§11）：**发现**按 ID 区间读、**水位**只表示
+ * 读取进度、**处理标记**只用于回填与库方清除——标记不得作为扫描谓词。
 */
 interface CminmsgInboxRepository {
    fun insertRaw(rawXml: String): Long

    fun rawOf(msgId: Long): String?

-    fun pollUnprocessed(afterId: Long, limit: Int): List<Long>
+    /** 按 ID 区间升序有界读取（`ID > fromExclusive`），不以处理标记为谓词。 */
+    fun readRange(fromExclusive: Long, limit: Int): List<MailboxRow>

-    fun backfillOnSuccess(msgId: Long, sndr: String, type: String, styp: String, seqn: Long)
+    /** 信箱当前最大 ID（空表 null）；仅用于观测水位滞后。 */
+    fun maxId(): Long?
+
+    /** 只把空标写为已处理（§11 单调）；返回 true = 本次实际写入，重复执行无副作用。 */
+    fun markProcessedIfUnmarked(msgId: Long, value: String): Boolean
 }
+
+/** 信箱行读取结果（发现阶段只需要身份与接收时间，原文按需再取）。 */
+data class MailboxRow(val msgId: Long, val receivedAt: Instant?)
@@ -1,14 +1,18 @@
 package com.gzzn.omms.msgexchange.infra.persistence.jdbc

 import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository
+import com.gzzn.omms.msgexchange.infra.persistence.MailboxRow
 import io.micronaut.context.annotation.Requires
 import jakarta.inject.Named
 import jakarta.inject.Singleton
 import javax.sql.DataSource

 /**
- * 共享 MySQL CMINMSGS 信箱适配（ACM2-12：仅 DML，不建表）。
+ * 共享 MySQL CMINMSGS 信箱适配（仅 DML，不建表；message-lifecycle §5.1/§6/§11）。
 * 列名与 legacy `entity/Cminmsg.java` 一致。
+ *
+ * 发现按 ID 区间读（`ID > ?`），**不以 `DATE_PROCESSED` 为扫描谓词**：已入队但尚未
+ * 回填的行否则会永久占据批次，正是 §5.1 与 US-01 条目 3 要求排除的场景。
 */
@Singleton
@Requires(property = "msgx.stubs", notEquals = "true")
@@ -33,31 +37,45 @@ class JdbcCminmsgInboxRepository(
            { ps -> ps.setLong(1, msgId) },
        ) { rs -> rs.getString("CMINMSGS_CLOB_MSG") }

-    override fun pollUnprocessed(afterId: Long, limit: Int): List<Long> =
+    override fun readRange(fromExclusive: Long, limit: Int): List<MailboxRow> =
        ds.query(
                """
-                SELECT CMINMSGS_ID FROM cminmsgs
-                WHERE CMINMSGS_ID > ? AND CMINMSGS_DATE_PROCESSED IS NULL
+                SELECT CMINMSGS_ID, CMINMSGS_DATE_RECEIVED FROM cminmsgs
+                WHERE CMINMSGS_ID > ?
                ORDER BY CMINMSGS_ID ASC
                LIMIT ?
                """.trimIndent(),
                { ps ->
-                    ps.setLong(1, afterId)
+                    ps.setLong(1, fromExclusive)
                    ps.setInt(2, limit)
                },
-            ) { rs -> rs.getLong("CMINMSGS_ID") }
+            ) { rs ->
+                MailboxRow(
+                    msgId = rs.getLong("CMINMSGS_ID"),
+                    receivedAt = rs.getTimestamp("CMINMSGS_DATE_RECEIVED")?.toInstant(),
+                )
+            }

-    override fun backfillOnSuccess(msgId: Long, sndr: String, type: String, styp: String, seqn: Long) {
+    override fun maxId(): Long? =
+        ds.queryOne("SELECT MAX(CMINMSGS_ID) AS max_id FROM cminmsgs", {}) { rs ->
+            rs.getLong("max_id").takeIf { !rs.wasNull() }
+        }
+
+    /**
+     * §11 单调：`DATE_PROCESSED IS NULL` 守卫保证只把空标写为已处理，已有值不回撤、
+     * 不覆盖；影响 0 行 = 已被其他路径标记，调用方按幂等成功处理。
+     * 写入值（DATE_PROCESSED 时间语义与 STATUS 值集）以库方契约为准（Q7）。
+     */
+    override fun markProcessedIfUnmarked(msgId: Long, value: String): Boolean =
        ds.update(
            """
            UPDATE cminmsgs
            SET CMINMSGS_DATE_PROCESSED = CURRENT_TIMESTAMP,
                CMINMSGS_STATUS = ?
-            WHERE CMINMSGS_ID = ?
+            WHERE CMINMSGS_ID = ? AND CMINMSGS_DATE_PROCESSED IS NULL
            """.trimIndent(),
        ) { ps ->
-            ps.setString(1, "PROCESSED")
+            ps.setString(1, value)
            ps.setLong(2, msgId)
-        }
-    }
+        } == 1
 }
@@ -11,9 +11,11 @@ import com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot
 import com.gzzn.omms.msgexchange.domain.flight.FlightState
 import com.gzzn.omms.msgexchange.domain.flight.HistoryCandidate
 import com.gzzn.omms.msgexchange.domain.flight.HistoryRules
-import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository
+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
@@ -68,22 +70,20 @@ class JdbcPipelineLockRepository(
 class JdbcProcStateRepository(
    private val ds: DataSource,
 ) : ProcStateRepository {
-    override fun insert(msgId: Long, state: ProcStatus) {
+    /** MSG_ID 主键幂等入队：重复扫描与 compat 入口并发都不重复建行（§5.1）。 */
+    override fun insertIfAbsent(msgId: Long, receivedAt: Instant?): Boolean =
        ds.update(
-            "INSERT INTO proc_state (msg_id, state, updated_at) VALUES (?, ?, ?)",
-            { ps -> ps.setLong(1, msgId); ps.setString(2, state.name); ps.setTimestamp(3, Instant.now().toSqlTimestamp()) },
-        )
-    }
-
-    override fun exists(msgId: Long): Boolean =
-        ds.queryOne("SELECT 1 FROM proc_state WHERE msg_id = ?", { ps -> ps.setLong(1, msgId) }) { 1 } != null
+            "INSERT INTO proc_state (msg_id, state, received_at, updated_at) VALUES (?, 'PENDING', ?, ?) " +
+                "ON CONFLICT (msg_id) DO NOTHING",
+            { ps ->
+                ps.setLong(1, msgId)
+                ps.setTimestamp(2, receivedAt?.toSqlTimestamp())
+                ps.setTimestamp(3, Instant.now().toSqlTimestamp())
+            },
+        ) == 1

    override fun find(msgId: Long): ProcState? =
-        ds.queryOne(
-            "SELECT msg_id, state, identity_key, attempts, next_attempt_at, error_class, last_error, updated_at FROM proc_state WHERE msg_id = ?",
-            { ps -> ps.setLong(1, msgId) },
-            ::mapProcState,
-        )
+        ds.queryOne("$SELECT_PROC WHERE msg_id = ?", { ps -> ps.setLong(1, msgId) }, ::mapProcState)

    override fun findSuccessTerminal(msgId: Long): Boolean =
        ds.queryOne(
@@ -93,8 +93,7 @@ class JdbcProcStateRepository(

    override fun headUnfinished(): ProcState? =
        ds.queryOne(
-            "SELECT msg_id, state, identity_key, attempts, next_attempt_at, error_class, last_error, updated_at " +
-                "FROM proc_state WHERE state IN ('PENDING', 'FAILED') ORDER BY msg_id ASC LIMIT 1",
+            "$SELECT_PROC WHERE state IN ('PENDING', 'FAILED') ORDER BY msg_id ASC LIMIT 1",
            {},
            ::mapProcState,
        )
@@ -145,6 +144,80 @@ class JdbcProcStateRepository(
        )
    }

+    /** 终态与回填意图同一条 UPDATE：业务事务内调用即原子提交（message-lifecycle §2/§4）。 */
+    override fun markTerminal(
+        msgId: Long,
+        state: ProcStatus,
+        errorClass: ErrorClass?,
+        lastError: String?,
+        attempts: Int?,
+        now: Instant,
+    ) {
+        ds.update(
+            """
+            UPDATE proc_state
+            SET state = ?, error_class = ?, last_error = ?, attempts = COALESCE(?, attempts),
+                next_attempt_at = NULL,
+                backfill_at = NULL, backfill_next_at = ?, backfill_attempts = 0, backfill_error = NULL,
+                updated_at = ?
+            WHERE msg_id = ?
+            """.trimIndent(),
+            { ps ->
+                ps.setString(1, state.name)
+                ps.setString(2, errorClass?.name)
+                ps.setString(3, lastError?.take(1000))
+                attempts?.let { ps.setInt(4, it) } ?: ps.setNull(4, java.sql.Types.INTEGER)
+                ps.setTimestamp(5, now.toSqlTimestamp())
+                ps.setTimestamp(6, now.toSqlTimestamp())
+                ps.setLong(7, msgId)
+            },
+        )
+    }
+
+    override fun markBackfilled(msgId: Long, now: Instant) {
+        ds.update(
+            "UPDATE proc_state SET backfill_at = ?, backfill_next_at = NULL, backfill_error = NULL, updated_at = ? WHERE msg_id = ?",
+            { ps ->
+                ps.setTimestamp(1, now.toSqlTimestamp())
+                ps.setTimestamp(2, now.toSqlTimestamp())
+                ps.setLong(3, msgId)
+            },
+        )
+    }
+
+    override fun recordBackfillFailure(msgId: Long, error: String?, attempts: Int, nextAttemptAt: Instant, now: Instant) {
+        ds.update(
+            "UPDATE proc_state SET backfill_attempts = ?, backfill_next_at = ?, backfill_error = ?, updated_at = ? WHERE msg_id = ?",
+            { ps ->
+                ps.setInt(1, attempts)
+                ps.setTimestamp(2, nextAttemptAt.toSqlTimestamp())
+                ps.setString(3, error?.take(512))
+                ps.setTimestamp(4, now.toSqlTimestamp())
+                ps.setLong(5, msgId)
+            },
+        )
+    }
+
+    /**
+     * 待回填：终态 + 未确认标记，且已到期或已达 §5.2 超期期限（R 覆盖退避，保证
+     * 有限时间内必然补写，否则库方清除的前提"边界内无未标记行"无法成立）。
+     */
+    override fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List<BackfillDue> =
+        ds.query(
+            """
+            SELECT msg_id, backfill_attempts FROM proc_state
+            WHERE backfill_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 < ?))
+            ORDER BY msg_id ASC LIMIT ?
+            """.trimIndent(),
+            { ps ->
+                ps.setTimestamp(1, now.toSqlTimestamp())
+                ps.setTimestamp(2, overdueBefore.toSqlTimestamp())
+                ps.setInt(3, limit)
+            },
+        ) { rs -> BackfillDue(rs.getLong("msg_id"), rs.getInt("backfill_attempts")) }
+
    override fun requeueByErrorClasses(errorClasses: List<ErrorClass>): Int {
        if (errorClasses.isEmpty()) return 0
        val placeholders = errorClasses.joinToString(",") { "?" }
@@ -158,6 +231,25 @@ class JdbcProcStateRepository(
        )
    }

+    /** OPS-2（message-lifecycle §5.3）：积压条数、最老未处理接收时刻、未回填终态条数。 */
+    override fun backlog(): Backlog =
+        ds.queryOne(
+            """
+            SELECT
+                count(*) FILTER (WHERE state IN ('PENDING', 'FAILED')) AS unfinished,
+                min(received_at) FILTER (WHERE state IN ('PENDING', 'FAILED')) AS oldest_received_at,
+                count(*) FILTER (WHERE state IN ('SUCCEEDED', 'SKIPPED', 'DEAD') AND backfill_at IS NULL) AS unmarked_terminal
+            FROM proc_state
+            """.trimIndent(),
+            {},
+        ) { rs ->
+            Backlog(
+                unfinished = rs.getInt("unfinished"),
+                oldestReceivedAt = rs.getInstant("oldest_received_at"),
+                unmarkedTerminal = rs.getInt("unmarked_terminal"),
+            )
+        } ?: Backlog(0, null, 0)
+
    private fun mapProcState(rs: ResultSet) = ProcState(
        msgId = rs.getLong("msg_id"),
        state = ProcStatus.valueOf(rs.getString("state")),
@@ -166,8 +258,45 @@ class JdbcProcStateRepository(
        nextAttemptAt = rs.getInstant("next_attempt_at"),
        errorClass = rs.getString("error_class")?.let(ErrorClass::valueOf),
        lastError = rs.getString("last_error"),
+        receivedAt = rs.getInstant("received_at"),
+        backfillAt = rs.getInstant("backfill_at"),
+        backfillNextAt = rs.getInstant("backfill_next_at"),
+        backfillAttempts = rs.getInt("backfill_attempts"),
+        backfillError = rs.getString("backfill_error"),
        updatedAt = rs.getInstant("updated_at") ?: Instant.now(),
    )
+
+    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, updated_at FROM proc_state"
+    }
+}
+
+/** 消费水位单行游标（message-lifecycle §5.1）；与入队同事务写入。 */
+@Singleton
+@Requires(property = "datasources.default.enabled", value = "true")
+@Requires(missingProperty = "msgx.stubs")
+class JdbcInboxCursorRepository(
+    private val ds: DataSource,
+) : InboxCursorRepository {
+    override fun load(): InboxCursorRepository.Cursor =
+        ds.queryOne(
+            "SELECT committed_up_to, hole_since FROM inbox_cursor WHERE cursor_id = 1",
+            {},
+        ) { rs -> InboxCursorRepository.Cursor(rs.getLong("committed_up_to"), rs.getInstant("hole_since")) }
+            ?: InboxCursorRepository.Cursor()
+
+    override fun save(cursor: InboxCursorRepository.Cursor) {
+        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, Instant.now().toSqlTimestamp())
+            },
+        )
+    }
 }

@Singleton
@@ -673,64 +802,4 @@ class JdbcReqTrackRepository(
    )
 }

-@Singleton
-@Requires(property = "datasources.default.enabled", value = "true")
-@Requires(missingProperty = "msgx.stubs")
-class JdbcBackfillTodoRepository(
-    private val ds: DataSource,
-) : BackfillTodoRepository {
-    override fun record(task: BackfillTodoRepository.BackfillTask, lastError: String?, now: Instant) {
-        ds.update(
-            """
-            INSERT INTO backfill_todo (msg_id, sndr, type, styp, seqn, attempts, next_attempt_at, last_error, created_at, updated_at)
-            VALUES (?, ?, ?, ?, ?, 0, ?, ?, ?, ?)
-            ON CONFLICT (msg_id) DO UPDATE SET last_error = EXCLUDED.last_error, next_attempt_at = EXCLUDED.next_attempt_at, updated_at = EXCLUDED.updated_at
-            """.trimIndent(),
-            { ps ->
-                ps.setLong(1, task.msgId)
-                ps.setString(2, task.sndr)
-                ps.setString(3, task.type)
-                ps.setString(4, task.styp)
-                ps.setLong(5, task.seqn)
-                ps.setTimestamp(6, now.toSqlTimestamp())
-                ps.setString(7, lastError?.take(512))
-                ps.setTimestamp(8, now.toSqlTimestamp())
-                ps.setTimestamp(9, now.toSqlTimestamp())
-            },
-        )
-    }

-    override fun findDue(now: Instant, limit: Int): List<BackfillTodoRepository.BackfillTask> =
-        ds.query(
-            "SELECT msg_id, sndr, type, styp, seqn, attempts FROM backfill_todo WHERE next_attempt_at <= ? ORDER BY next_attempt_at ASC LIMIT ?",
-            { ps -> ps.setTimestamp(1, now.toSqlTimestamp()); ps.setInt(2, limit) },
-        ) { rs ->
-            BackfillTodoRepository.BackfillTask(
-                msgId = rs.getLong("msg_id"),
-                sndr = rs.getString("sndr"),
-                type = rs.getString("type"),
-                styp = rs.getString("styp"),
-                seqn = rs.getLong("seqn"),
-                attempts = rs.getInt("attempts"),
-            )
-        }
-
-    override fun markFailed(msgId: Long, lastError: String?, nextAttemptAt: Instant, now: Instant) {
-        ds.update(
-            "UPDATE backfill_todo SET attempts = attempts + 1, last_error = ?, next_attempt_at = ?, updated_at = ? WHERE msg_id = ?",
-            { ps ->
-                ps.setString(1, lastError?.take(512))
-                ps.setTimestamp(2, nextAttemptAt.toSqlTimestamp())
-                ps.setTimestamp(3, now.toSqlTimestamp())
-                ps.setLong(4, msgId)
-            },
-        )
-    }
-
-    override fun delete(msgId: Long) {
-        ds.update("DELETE FROM backfill_todo WHERE msg_id = ?", { ps -> ps.setLong(1, msgId) })
-    }
-
-    override fun count(): Int =
-        ds.queryOne("SELECT count(*) AS n FROM backfill_todo", {}) { rs -> rs.getInt("n") } ?: 0
-}
@@ -8,7 +8,7 @@ import jakarta.inject.Singleton

 /**
 * ProcState 侧统一失败迁移（U08/U10）：处理/快照路径共用——
- * attempts+1 后若 exhausted → DEAD(EXHAUSTED)（终态，errorClass 规范化，原因保留在 lastError）；
+ * attempts+1 后若 exhausted → DEAD(EXHAUSTED)（终态 + 回填意图，errorClass 规范化，原因保留在 lastError）；
 * 否则 FAILED + attempts + nextAttemptAt(退避)（可重放）。任何“失败”都不得在无退避下直接终态化。
 */
@Singleton
@@ -16,23 +16,25 @@ class ProcFailure(
    private val procState: ProcStateRepository,
    val scheduler: FailureScheduler,
 ) {
-    fun fail(head: ProcState, ec: ErrorClass, reason: String) {
+    /** @return 是否已达终态（DEAD 才是终态；FAILED 仍可重放/重试） */
+    fun fail(head: ProcState, ec: ErrorClass, reason: String): Boolean {
        val attempts = head.attempts + 1
        if (scheduler.exhausted(attempts)) {
-            procState.update(
+            procState.markTerminal(
                head.msgId, ProcStatus.DEAD,
                attempts = attempts,
                errorClass = ErrorClass.EXHAUSTED,
                lastError = "$reason; attempts=$attempts",
            )
-        } else {
-            procState.update(
-                head.msgId, ProcStatus.FAILED,
-                attempts = attempts,
-                nextAttemptAt = scheduler.nextAttemptAt(attempts),
-                errorClass = ec,
-                lastError = reason,
-            )
+            return true
        }
+        procState.update(
+            head.msgId, ProcStatus.FAILED,
+            attempts = attempts,
+            nextAttemptAt = scheduler.nextAttemptAt(attempts),
+            errorClass = ec,
+            lastError = reason,
+        )
+        return false
    }
 }
@@ -11,8 +11,11 @@ import com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot
 import com.gzzn.omms.msgexchange.domain.flight.FlightState
 import com.gzzn.omms.msgexchange.domain.flight.HistoryCandidate
 import com.gzzn.omms.msgexchange.domain.flight.HistoryRules
-import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository
+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.FlightStateRepository
 import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository
 import com.gzzn.omms.msgexchange.infra.persistence.PersistOutcome
@@ -55,12 +58,12 @@ class StubProcState : ProcStateRepository {
        rows.clear(); bound.clear()
    }

-    override fun insert(msgId: Long, state: ProcStatus) {
-        rows.getOrPut(msgId) { ProcState(msgId, state) }
+    override fun insertIfAbsent(msgId: Long, receivedAt: Instant?): Boolean {
+        if (rows.containsKey(msgId)) return false
+        rows[msgId] = ProcState(msgId, ProcStatus.PENDING, receivedAt = receivedAt)
+        return true
    }

-    override fun exists(msgId: Long): Boolean = rows.containsKey(msgId)
-
    override fun find(msgId: Long): ProcState? = rows[msgId]

    override fun findSuccessTerminal(msgId: Long): Boolean = rows[msgId]?.state == ProcStatus.SUCCEEDED
@@ -97,6 +100,67 @@ class StubProcState : ProcStateRepository {
        )
    }

+    /** 终态 + 回填意图同一动作（模拟单条 UPDATE 的原子性）。 */
+    override fun markTerminal(
+        msgId: Long,
+        state: ProcStatus,
+        errorClass: ErrorClass?,
+        lastError: String?,
+        attempts: Int?,
+        now: Instant,
+    ) {
+        val old = rows[msgId] ?: ProcState(msgId, state)
+        rows[msgId] = old.copy(
+            state = state,
+            errorClass = errorClass,
+            lastError = lastError,
+            attempts = attempts ?: old.attempts,
+            nextAttemptAt = null,
+            backfillAt = null,
+            backfillNextAt = now,
+            backfillAttempts = 0,
+            backfillError = null,
+            updatedAt = now,
+        )
+    }
+
+    override fun markBackfilled(msgId: Long, now: Instant) {
+        rows[msgId]?.let {
+            rows[msgId] = it.copy(backfillAt = now, backfillNextAt = null, backfillError = null, updatedAt = now)
+        }
+    }
+
+    override fun recordBackfillFailure(msgId: Long, error: String?, attempts: Int, nextAttemptAt: Instant, now: Instant) {
+        rows[msgId]?.let {
+            rows[msgId] = it.copy(
+                backfillAttempts = attempts,
+                backfillNextAt = nextAttemptAt,
+                backfillError = error,
+                updatedAt = now,
+            )
+        }
+    }
+
+    override fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List<BackfillDue> =
+        rows.values
+            .filter { it.state.isTerminal() && it.backfillAt == null }
+            .filter {
+                it.backfillNextAt == null || it.backfillNextAt <= now ||
+                    (it.receivedAt != null && it.receivedAt < overdueBefore)
+            }
+            .sortedBy { it.msgId }
+            .take(limit)
+            .map { BackfillDue(it.msgId, it.backfillAttempts) }
+
+    override fun backlog(): Backlog {
+        val unfinished = rows.values.filter { it.state == ProcStatus.PENDING || it.state == ProcStatus.FAILED }
+        return Backlog(
+            unfinished = unfinished.size,
+            oldestReceivedAt = unfinished.mapNotNull { it.receivedAt }.minOrNull(),
+            unmarkedTerminal = rows.values.count { it.state.isTerminal() && it.backfillAt == null },
+        )
+    }
+
    override fun requeueByErrorClasses(errorClasses: List<ErrorClass>): Int {
        var n = 0
        rows.forEach { (id, s) ->
@@ -300,60 +364,69 @@ class StubReqTrack : ReqTrackRepository {
    }
 }

-@Singleton
-@Requires(property = "msgx.stubs", value = "true")
-class StubBackfillTodo : BackfillTodoRepository {
-    val tasks = linkedMapOf<Long, BackfillTodoRepository.BackfillTask>()
-    private val errors = linkedMapOf<Long, String?>()
-    private val due = linkedMapOf<Long, Instant>()
-
-    fun clear() { tasks.clear(); errors.clear(); due.clear() }
-
-    fun lastErrorOf(msgId: Long): String? = errors[msgId]
-
-    override fun record(task: BackfillTodoRepository.BackfillTask, lastError: String?, now: Instant) {
-        errors[task.msgId] = lastError
-        due.putIfAbsent(task.msgId, now)
-        tasks[task.msgId] = task.copy(attempts = tasks[task.msgId]?.attempts ?: 0)
-    }
-
-    override fun findDue(now: Instant, limit: Int): List<BackfillTodoRepository.BackfillTask> =
-        tasks.keys.filter { (due[it] ?: Instant.EPOCH) <= now }.take(limit).mapNotNull { tasks[it] }
-
-    override fun markFailed(msgId: Long, lastError: String?, nextAttemptAt: Instant, now: Instant) {
-        errors[msgId] = lastError
-        due[msgId] = nextAttemptAt
-        tasks[msgId]?.let { tasks[msgId] = it.copy(attempts = it.attempts + 1) }
-    }
-
-    override fun delete(msgId: Long) {
-        tasks.remove(msgId); errors.remove(msgId); due.remove(msgId)
-    }
-
-    override fun count(): Int = tasks.size
-}
-
@Singleton
@Requires(property = "msgx.stubs", value = "true")
 class StubInbox : CminmsgInboxRepository {
    val raws = linkedMapOf<Long, String>()
+    private val received = linkedMapOf<Long, Instant>()
+    private val marks = linkedMapOf<Long, String>()
    private val ids = AtomicLong(0)

-    fun clear() = raws.clear()
+    fun clear() {
+        raws.clear(); received.clear(); marks.clear(); ids.set(0)   // ID 自 1 重新分配，用例间不串扰
+    }

    override fun insertRaw(rawXml: String): Long {
        val id = ids.incrementAndGet()
        raws[id] = rawXml
+        received[id] = Instant.now()
        return id
    }

    override fun rawOf(msgId: Long): String? = raws[msgId]

-    override fun pollUnprocessed(afterId: Long, limit: Int): List<Long> =
-        raws.keys.filter { it > afterId }.sorted().take(limit)
+    override fun readRange(fromExclusive: Long, limit: Int): List<MailboxRow> =
+        raws.keys.filter { it > fromExclusive }.sorted().take(limit).map { MailboxRow(it, received[it]) }

-    override fun backfillOnSuccess(msgId: Long, sndr: String, type: String, styp: String, seqn: Long) = Unit
+    override fun maxId(): Long? = raws.keys.maxOrNull()
+
+    /** §11 单调：只把空标写为已处理；已有值不回撤、不覆盖。 */
+    override fun markProcessedIfUnmarked(msgId: Long, value: String): Boolean {
+        if (!raws.containsKey(msgId) || marks.containsKey(msgId)) return false
+        marks[msgId] = value
+        return true
+    }
+
+    fun markOf(msgId: Long): String? = marks[msgId]
+
+    fun isMarked(msgId: Long): Boolean = marks.containsKey(msgId)

    /** 测试辅助：模拟上游外部写入共享信箱（不经本系统）。 */
    fun simulateExternalWrite(rawXml: String): Long = insertRaw(rawXml)
+
+    /** 测试辅助：模拟库方清除（原文不可读），用于 §9 原文缺失与空洞场景。 */
+    fun removeRow(msgId: Long) {
+        raws.remove(msgId); received.remove(msgId); marks.remove(msgId)
+    }
 }
+
+/** 消费水位游标（stub）。 */
+@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) {
+        this.cursor = cursor
+    }
+}
+
+/** 终态判定（§2 状态总纲）。 */
+private fun ProcStatus.isTerminal(): Boolean =
+    this == ProcStatus.SUCCEEDED || this == ProcStatus.SKIPPED || this == ProcStatus.DEAD