fix(processing): 收紧消息生命周期与投递约束

docs/design.md

@@ -17,7 +17,7 @@
 
 | 记录 | 用途 | 关键约束 |
 |---|---|---|
-| `PROC_STATE` | 入站消息的处理状态、身份、重试次数、错误原因与回填事实 | `MSG_ID = CMINMSGS_ID` 主键防止重复入队；`IDENTITY_KEY` 唯一约束防止业务重复；按最小未完成消息 ID 取队头；`RECEIVED_AT` 与 `BACKFILL_*` 承载 [message-lifecycle.md](message-lifecycle.md) §5.2 的超期判据与回填重试。 |
+| `PROC_STATE` | 入站消息的处理状态、身份、重试次数、错误原因与回填事实 | `MSG_ID = CMINMSGS_ID` 主键防止重复入队；`IDENTITY_KEY` 唯一约束防止业务重复；按最小未完成消息 ID 取队头；`PROCESSING_STARTED_AT` 是 HOL deadline 的稳定起点；`RECEIVED_AT` 与 `BACKFILL_*` 承载 [message-lifecycle.md](message-lifecycle.md) §5.2 的超期判据与回填重试。 |
 | `MSG_EVENT` | 等待投递的事件（outbox） | `EVENT_ID` 决定投递顺序；`TARGET` 区分 `KAFKA:msg` / `KAFKA:schd`；`PARTITION_KEY` 恒为 `FLID`；`EVENT_TYPE` 区分 UPSERT 与 TOMBSTONE。 |
 | `REQ_TRACK` | 上游请求及应答关联 | 保存请求类型、覆盖运营日、发送方、出站信箱 ID 与发送/完成时间；同类只允许一个开放请求。登记、超时与应答匹配尚未实现（§10）。 |
 | `REF_MASTER` | 静态参考数据（目标表） | `(RTYPE, RKEY)` 唯一；尚未建表，客户端与刷新流程见 user-stories.md US-13/US-14（US-14 两类映射的存储落点未定）。 |
@@ -83,7 +83,7 @@ SNDR | TYPE | STYP | SEQN
 2. 队头 `FAILED` 且未到 `next_attempt_at`：未超限则等到可重试时刻；已达重试上限或超过队头滞留时限（`head-deadline`，默认 10 分钟）则转 `DEAD(EXHAUSTED)`。
 3. 队头可执行：调用 `MessageProcessor.processOne`，失败迁移在该边界内完成。
 
-维护作业由独立 job 线程调度（§6.1），不占用消息循环；作业有界且不使到期消息无限饥饿。调度取时已经可注入 `Clock`，但滞留起点当前仍使用会随重试刷新的 `updatedAt`，尚未改为稳定起始时刻（§10）。
+维护作业由独立 job 线程调度（§6.1），不占用消息循环；作业有界且不使到期消息无限饥饿。调度取时经可注入 `Clock`，HOL deadline 以首次处理时写入的 `PROCESSING_STARTED_AT` 为稳定起点；人工重放会清空该值，由新一轮首次处理重新记录。
 
 ### 3.3 单条处理
 
@@ -228,7 +228,7 @@ REGISTERED → SENT → WAITING → DONE
 以下缺口直接影响上述设计是否成立，不能以类或接口已存在作为完成依据：
 
 - **事务与外部副作用**：状态、事件、处理终态与回填意图同 PG 事务提交；回填意图落在 `PROC_STATE`（`BACKFILL_AT/NEXT_AT/ATTEMPTS/ERROR`），`BACKFILL_TODO` 已随 V2 迁移下线，[message-lifecycle.md](message-lifecycle.md) §4 的两个崩溃窗口（提交后回填前崩溃、待办二次落账失败）不再存在。回填本身仍是跨库单写，失败按退避重试并由超期期限 R 兜底；R 的取值待 Q6 确认。
-- **收报与调度**：水位 W 落库并与入队同事务推进，遇空洞即停、空洞超过 `max-commit-delay` 判定为永久（Q2 未书面确认前该阈值是假定值）；较小 ID 迟提交与空洞场景的顺序保证仍待与库方联合验证。滞留判断已使用可注入 `Clock` 取时，但起点仍是会随重试刷新的 `updatedAt`，尚未改为稳定起始时刻（Q6）。
+- **收报与调度**：水位 W 落库并与入队同事务推进，遇空洞即停、空洞超过 `max-commit-delay` 判定为永久（Q2 未书面确认前该阈值是假定值）；旧空洞补齐后出现的新空洞会重置老化起点。较小 ID 迟提交与空洞场景的端到端顺序保证仍待与库方联合验证。HOL deadline 已改用 `PROCESSING_STARTED_AT` 和可注入 `Clock`；Q6 仍需确认长期积压与人工重放的期限口径。
 - **快照与业务能力**：DNLD/RESP/ADFT 与 FLOP/FDEL 处理器、整包校验与跨运营日整包拒绝均已接入；但 RESP 应答守卫与出站请求未实现，忽略规则（US-04）未实现，29 类 FLOP 与参考应答的逐类矩阵未补全，ADFT 缺失字段与 `FLID` 重用语义待上游确认。
 - **航班读写**：唯一写入口与权威读已落地；ROUT/ERUT 联合主键、空值/未知属性保真、事件在事务内只算一次、逐航班多次查询仍待修正（见 flight-state.md §6）。`/all/flights` 尚未实现。
 - **请求、参考数据与归档**：`REQ_TRACK` 表与仓储已建，但无运行时协调与 `COUTMSGS` 出站适配；`REF_MASTER` 未建表；`PROC_STATE_HST` 未建表。航班历史清理脚手架已实现，历史存储未接通时删 0 条。

docs/user-stories.md

@@ -85,7 +85,7 @@
 7. 领域决策逻辑只读取当前完整态与已解码报文，返回下一完整态和待发事件，不执行 I/O；Processor 作为事务协调器，将状态写入、事件、处理终态与回填意图收敛在同一事务边界内，不直接触碰 Kafka。失败只在持有消息上下文的边界落账，中断向上传递，不作为普通失败吞掉。
 8. 权威存储不可用或未完成恢复时停止业务处理；不能把“整个状态丢失”误判为“单航班不存在”而批量成功结束增量报文。
 
-**当前基础与落点**：`processing/Pump.kt`（含 `MessageProcessor`）、`DynamicProcessors.kt`、`Identity.kt`、`codec/JacksonXmlCodec.kt`、`infra/retry/`。严格 FIFO 主泵、SCHD（DNLD/RESP/ADFT）/FLOP/FDEL 处理器、PG 单事务（含回填待办预登记）、身份绑定与重试已实现，调度取时已经由可注入 `Clock` 提供。尚未完成：忽略规则分支（US-04）、基于稳定起始时刻的滞留判据，以及逐类矩阵与 golden 样例（US-05）。
+**当前基础与落点**：`processing/Pump.kt`（含 `MessageProcessor`）、`DynamicProcessors.kt`、`Identity.kt`、`codec/JacksonXmlCodec.kt`、`infra/retry/`。严格 FIFO 主泵、SCHD（DNLD/RESP/ADFT）/FLOP/FDEL 处理器、PG 单事务（含回填待办预登记）、身份绑定与重试已实现；调度取时由可注入 `Clock` 提供，HOL deadline 以 `PROCESSING_STARTED_AT` 为稳定起点。尚未完成：忽略规则分支（US-04），以及逐类矩阵与 golden 样例（US-05）。长期积压和人工重放如何计算 deadline 仍由 Q6 定案。
 
 **前置**：US-01；Q1 已定单库方向，Q6 决定 deadline 边界。数据库迁移只落自有库。
 

src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt

@@ -79,7 +79,7 @@ class Dispatcher(
     }
 
     private fun flushDue(): Boolean =
-        lastFlush?.let { Duration.between(it, scheduler.now()) >= props.schd.flushPeriod } ?: false
+        lastFlush?.let { Duration.between(it, scheduler.now()) >= props.schd.flushPeriod } ?: true
 
     private fun deliver(target: String, e: MsgEvent) {
         try {
@@ -95,7 +95,7 @@ class Dispatcher(
     /** 批量发 KAFKA_SCHD：每个 FLID 只发版本号最新的那条未发事件，删除通知发空 value。 */
     internal fun flushSchd() {
         val batch = try {
-            msgEvents.mergePendingSchd(props.schd.flushLimit)
+            msgEvents.mergePendingSchd(scheduler.now(), props.schd.flushLimit)
         } catch (e: Exception) {
             log.warn("mergePendingSchd failed: {}", e.message)
             return
@@ -121,7 +121,7 @@ class Dispatcher(
         val sentVersions = batch.associate { it.partitionKey to it.stateVersion }
         runCatching {
             while (true) {
-                val superseded = msgEvents.mergePendingSchd(props.schd.flushLimit)
+                val superseded = msgEvents.mergePendingSchd(scheduler.now(), props.schd.flushLimit)
                     .filter { sentVersions[it.partitionKey]?.let { v -> it.stateVersion < v } == true }
                 if (superseded.isEmpty()) break
                 msgEvents.markAllSent(superseded.mapNotNull { it.eventId })

src/main/kotlin/com/gzzn/omms/msgexchange/domain/ProcState.kt

@@ -72,5 +72,7 @@ data class ProcState(
     val backfillNextAt: Instant? = null,
     val backfillAttempts: Int = 0,
     val backfillError: String? = null,
+    /** 首次被主泵取得的时刻；重试不刷新，用作 HOL deadline 的稳定起点。 */
+    val processingStartedAt: Instant? = null,
     val updatedAt: Instant = Instant.now(),
 )

src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt

@@ -58,6 +58,9 @@ interface ProcStateRepository {
     /** 当前队头：还没处理完的消息里 ID 最小的那条。 */
     fun headUnfinished(): ProcState?
 
+    /** 首次开始处理时记下稳定起点；重试不覆盖。 */
+    fun markProcessingStartedIfAbsent(msgId: Long, now: Instant)
+
     /** 给消息绑定业务身份；返回 false 表示这个身份已经被另一条消息占了（业务重复）。 */
     fun tryBindIdentity(msgId: Long, identityKey: String): Boolean
 
@@ -135,7 +138,7 @@ interface MsgEventRepository {
     fun claimBatch(target: String, limit: Int): List<MsgEvent>
 
     /** 挑出待发的整态事件，同一条航班只取版本最高的那一条（中间的版本不用发）。 */
-    fun mergePendingSchd(limit: Int): List<MsgEvent>
+    fun mergePendingSchd(now: Instant, limit: Int): List<MsgEvent>
 
     fun markSent(eventId: Long)
 
@@ -287,6 +290,9 @@ interface CminmsgInboxRepository {
     /** 读某条消息的原文；返回 null 表示读不到（行已被清除，或原文本身为空）。 */
     fun rawOf(msgId: Long): String?
 
+    /** 读取信箱库记录的实际接收时间。 */
+    fun receivedAtOf(msgId: Long): Instant?
+
     /** 按 ID 升序读一批 `ID > fromExclusive` 的行，不带别的过滤条件。 */
     fun readRange(fromExclusive: Long, limit: Int): List<MailboxRow>
 
@@ -295,10 +301,13 @@ interface CminmsgInboxRepository {
 
     /**
      * 把处理标记写回信箱，并且**只写还是空标记的行**：库里已有值时不覆盖、不回退，
-     * 重复调用没有副作用。返回 true 表示这次真的写进去了。
+     * 重复调用没有副作用。结果明确区分本次写入、已有标记和信箱行缺失。
      */
-    fun markProcessedIfUnmarked(msgId: Long, value: String): Boolean
+    fun markProcessedIfUnmarked(msgId: Long, value: String): MailboxMarkResult
 }
 
+/** 信箱回填的可区分结果；行缺失不能冒充“已有标记”。 */
+enum class MailboxMarkResult { MARKED, ALREADY_MARKED, MISSING }
+
 /** 从信箱读到的一行：ID 加接收时间。原文按需再取，扫描时不读大字段。 */
 data class MailboxRow(val msgId: Long, val receivedAt: Instant?)

src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcCminmsgInboxRepository.kt

@@ -2,10 +2,12 @@ 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 com.gzzn.omms.msgexchange.infra.persistence.MailboxMarkResult
 import io.micronaut.context.annotation.Requires
 import jakarta.inject.Named
 import jakarta.inject.Singleton
 import javax.sql.DataSource
+import java.time.Instant
 
 /**
  * 共享 MySQL CMINMSGS 信箱的 JDBC 实现。列名沿用 legacy 的 `CMINMSGS_*`。
@@ -39,6 +41,12 @@ class JdbcCminmsgInboxRepository(
             { ps -> ps.setLong(1, msgId) },
         ) { rs -> rs.getString("CMINMSGS_CLOB_MSG") }
 
+    override fun receivedAtOf(msgId: Long): Instant? =
+        ds.queryOne(
+            "SELECT CMINMSGS_DATE_RECEIVED FROM cminmsgs WHERE CMINMSGS_ID = ?",
+            { ps -> ps.setLong(1, msgId) },
+        ) { rs -> rs.getTimestamp("CMINMSGS_DATE_RECEIVED")?.toInstant() }
+
     override fun readRange(fromExclusive: Long, limit: Int): List<MailboxRow> =
         ds.query(
                 """
@@ -65,11 +73,11 @@ class JdbcCminmsgInboxRepository(
 
     /**
      * 只更新还是空标记的行，所以重复调用不会覆盖库里已有的值；
-     * 影响 0 行说明已经被标记过了，调用方按"成功"处理即可。
+     * 影响 0 行时再查一次主键，区分“已有标记”和“信箱行缺失”；后者不能记为回填成功。
      * 具体写什么值、时间怎么解释，以与库方约定为准。
      */
-    override fun markProcessedIfUnmarked(msgId: Long, value: String): Boolean =
-        ds.update(
+    override fun markProcessedIfUnmarked(msgId: Long, value: String): MailboxMarkResult {
+        val updated = ds.update(
             """
             UPDATE cminmsgs
             SET CMINMSGS_DATE_PROCESSED = CURRENT_TIMESTAMP,
@@ -79,5 +87,12 @@ class JdbcCminmsgInboxRepository(
         ) { ps ->
             ps.setString(1, value)
             ps.setLong(2, msgId)
-        } == 1
+        }
+        if (updated == 1) return MailboxMarkResult.MARKED
+        val exists = ds.queryOne(
+            "SELECT 1 FROM cminmsgs WHERE CMINMSGS_ID = ?",
+            { ps -> ps.setLong(1, msgId) },
+        ) { 1 } != null
+        return if (exists) MailboxMarkResult.ALREADY_MARKED else MailboxMarkResult.MISSING
+    }
 }

src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcOps.kt

@@ -10,17 +10,20 @@ internal fun Instant.toSqlTimestamp(): Timestamp = Timestamp.from(this)
 internal fun ResultSet.getInstant(column: String): Instant? =
     getTimestamp(column)?.toInstant()
 
-private val transactionConnection = ThreadLocal<java.sql.Connection?>()
+private val transactionConnections = ThreadLocal.withInitial {
+    java.util.IdentityHashMap<DataSource, java.sql.Connection>()
+}
 
 internal fun <T> DataSource.withTransaction(block: () -> T): T {
-    val existing = transactionConnection.get()
+    val connections = transactionConnections.get()
+    val existing = connections[this]
     if (existing != null) {
         return block()
     }
     val conn = this.connection
     val oldAutoCommit = conn.autoCommit
     conn.autoCommit = false
-    transactionConnection.set(conn)
+    connections[this] = conn
     try {
         val result = block()
         conn.commit()
@@ -33,7 +36,8 @@ internal fun <T> DataSource.withTransaction(block: () -> T): T {
         }
         throw t
     } finally {
-        transactionConnection.remove()
+        connections.remove(this)
+        if (connections.isEmpty()) transactionConnections.remove()
         try {
             conn.autoCommit = oldAutoCommit
         } catch (_: Throwable) {
@@ -43,11 +47,13 @@ internal fun <T> DataSource.withTransaction(block: () -> T): T {
 }
 
 internal fun DataSource.obtainConnection(): java.sql.Connection =
-    transactionConnection.get() ?: this.connection
+    transactionConnections.get()[this] ?: this.connection
 
-internal fun java.sql.Connection.releaseIfNotInTransaction() {
-    if (transactionConnection.get() !== this) {
+internal fun java.sql.Connection.releaseIfNotInTransaction(dataSource: DataSource) {
+    val connections = transactionConnections.get()
+    if (connections[dataSource] !== this) {
         this.close()
+        if (connections.isEmpty()) transactionConnections.remove()
     }
 }
 
@@ -63,7 +69,7 @@ internal fun <T> DataSource.query(sql: String, bind: (java.sql.PreparedStatement
             }
         }
     } finally {
-        conn.releaseIfNotInTransaction()
+        conn.releaseIfNotInTransaction(this)
     }
 }
 
@@ -78,7 +84,7 @@ internal fun DataSource.update(sql: String, bind: (java.sql.PreparedStatement) -
             ps.executeUpdate()
         }
     } finally {
-        conn.releaseIfNotInTransaction()
+        conn.releaseIfNotInTransaction(this)
     }
 }
 
@@ -94,6 +100,6 @@ internal fun DataSource.updateReturningLong(sql: String, bind: (java.sql.Prepare
             }
         }
     } finally {
-        conn.releaseIfNotInTransaction()
+        conn.releaseIfNotInTransaction(this)
     }
 }

src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt

@@ -99,6 +99,16 @@ class JdbcProcStateRepository(
             ::mapProcState,
         )
 
+    override fun markProcessingStartedIfAbsent(msgId: Long, now: Instant) {
+        ds.update(
+            "UPDATE proc_state SET processing_started_at = ? WHERE msg_id = ? AND processing_started_at IS NULL",
+            { ps ->
+                ps.setTimestamp(1, now.toSqlTimestamp())
+                ps.setLong(2, msgId)
+            },
+        )
+    }
+
     override fun tryBindIdentity(msgId: Long, identityKey: String): Boolean {
         ownerOfIdentity(identityKey)?.let { owner ->
             return owner == msgId
@@ -223,7 +233,7 @@ class JdbcProcStateRepository(
         if (errorClasses.isEmpty()) return 0
         val placeholders = errorClasses.joinToString(",") { "?" }
         return ds.update(
-            "UPDATE proc_state SET state = 'PENDING', attempts = 0, next_attempt_at = NULL, updated_at = ? " +
+            "UPDATE proc_state SET state = 'PENDING', attempts = 0, next_attempt_at = NULL, processing_started_at = NULL, updated_at = ? " +
                 "WHERE state IN ('FAILED', 'DEAD') AND error_class IN ($placeholders)",
             { ps ->
                 ps.setTimestamp(1, Instant.now().toSqlTimestamp())
@@ -264,13 +274,14 @@ class JdbcProcStateRepository(
         backfillNextAt = rs.getInstant("backfill_next_at"),
         backfillAttempts = rs.getInt("backfill_attempts"),
         backfillError = rs.getString("backfill_error"),
+        processingStartedAt = rs.getInstant("processing_started_at"),
         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"
+                "received_at, backfill_at, backfill_next_at, backfill_attempts, backfill_error, processing_started_at, updated_at FROM proc_state"
     }
 }
 
@@ -342,16 +353,21 @@ class JdbcMsgEventRepository(
         )
 
     /** 同一条航班只取版本最高的待发整态事件；用了 PostgreSQL 的 DISTINCT ON 语法。 */
-    override fun mergePendingSchd(limit: Int): List<MsgEvent> =
+    override fun mergePendingSchd(now: Instant, limit: Int): List<MsgEvent> =
         ds.query(
             """
             SELECT * FROM (
                 SELECT DISTINCT ON (partition_key) *
                 FROM msg_event WHERE target = 'KAFKA:schd' AND state = 'PENDING'
                 ORDER BY partition_key, state_version DESC, event_id DESC
-            ) latest ORDER BY event_id ASC LIMIT ?
+            ) latest
+            WHERE next_attempt_at IS NULL OR next_attempt_at <= ?
+            ORDER BY event_id ASC LIMIT ?
             """.trimIndent(),
-            { ps -> ps.setInt(1, limit) },
+            { ps ->
+                ps.setTimestamp(1, now.toSqlTimestamp())
+                ps.setInt(2, limit)
+            },
             ::mapEvent,
         )
 
@@ -805,5 +821,3 @@ class JdbcReqTrackRepository(
         sentAt = rs.getInstant("sent_at"),
     )
 }
-
-

src/main/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayService.kt

@@ -2,6 +2,7 @@ package com.gzzn.omms.msgexchange.infra.retry
 
 import com.gzzn.omms.msgexchange.domain.ErrorClass
 import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository
+import com.gzzn.omms.msgexchange.processing.MessageLifecycleGate
 import jakarta.inject.Singleton
 
 /**
@@ -13,6 +14,7 @@ import jakarta.inject.Singleton
 @Singleton
 class ReplayService(
     private val procState: ProcStateRepository,
+    private val lifecycleGate: MessageLifecycleGate = MessageLifecycleGate(),
 ) {
     private val log = org.slf4j.LoggerFactory.getLogger(ReplayService::class.java)
     /** 可以重放的错误类：解码逻辑修好后能过、处理器补齐后能过、基础设施抖动已恢复、以及重试耗尽但人工复核认为还能再试的。 */
@@ -26,7 +28,7 @@ class ReplayService(
             log.warn("replay requested only non-replayable classes: {}", requested)
             return 0
         }
-        val n = procState.requeueByErrorClasses(allowed)
+        val n = lifecycleGate.exclusive { procState.requeueByErrorClasses(allowed) }
         log.info("replayed rows={} classes={}", n, allowed)
         return n
     }

src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt

@@ -16,6 +16,7 @@ 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
 import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository
 import com.gzzn.omms.msgexchange.infra.persistence.PersistOutcome
@@ -74,6 +75,12 @@ class StubProcState : ProcStateRepository {
     override fun headUnfinished(): ProcState? =
         rows.values.filter { it.state == ProcStatus.PENDING || it.state == ProcStatus.FAILED }.minByOrNull { it.msgId }
 
+    override fun markProcessingStartedIfAbsent(msgId: Long, now: Instant) {
+        rows[msgId]?.let { row ->
+            if (row.processingStartedAt == null) rows[msgId] = row.copy(processingStartedAt = now)
+        }
+    }
+
     override fun tryBindIdentity(msgId: Long, identityKey: String): Boolean {
         val owner = bound[identityKey]
         if (owner != null && owner != msgId) return false
@@ -168,7 +175,7 @@ class StubProcState : ProcStateRepository {
         var n = 0
         rows.forEach { (id, s) ->
             if (s.errorClass in errorClasses && (s.state == ProcStatus.FAILED || s.state == ProcStatus.DEAD)) {
-                rows[id] = s.copy(state = ProcStatus.PENDING, attempts = 0, nextAttemptAt = null)
+                rows[id] = s.copy(state = ProcStatus.PENDING, attempts = 0, nextAttemptAt = null, processingStartedAt = null)
                 n++
             }
         }
@@ -200,11 +207,12 @@ class StubMsgEvents : MsgEventRepository {
         rows.values.filter { it.target == target && it.state == EventStatus.PENDING }.sortedBy { it.eventId!! }.take(limit)
 
     /** 同一条航班只取版本最高的待发整态事件，按事件先后输出。 */
-    override fun mergePendingSchd(limit: Int): List<MsgEvent> =
+    override fun mergePendingSchd(now: Instant, limit: Int): List<MsgEvent> =
         rows.values
             .filter { it.target == "KAFKA:schd" && it.state == EventStatus.PENDING }
             .groupBy { it.partitionKey }
             .map { (_, group) -> group.maxBy { it.stateVersion } }
+            .filter { it.nextAttemptAt == null || it.nextAttemptAt <= now }
             .sortedBy { it.eventId!! }
             .take(limit)
 
@@ -389,16 +397,19 @@ class StubInbox : CminmsgInboxRepository {
 
     override fun rawOf(msgId: Long): String? = raws[msgId]
 
+    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()
 
-    /** 只写还没有标记的行；已经标记过就返回 false，不覆盖已有值。 */
-    override fun markProcessedIfUnmarked(msgId: Long, value: String): Boolean {
-        if (!raws.containsKey(msgId) || marks.containsKey(msgId)) return false
+    /** 只写还没有标记的行，并区分已有标记与行缺失。 */
+    override fun markProcessedIfUnmarked(msgId: Long, value: String): MailboxMarkResult {
+        if (!raws.containsKey(msgId)) return MailboxMarkResult.MISSING
+        if (marks.containsKey(msgId)) return MailboxMarkResult.ALREADY_MARKED
         marks[msgId] = value
-        return true
+        return MailboxMarkResult.MARKED
     }
 
     fun markOf(msgId: Long): String? = marks[msgId]
@@ -408,6 +419,12 @@ class StubInbox : CminmsgInboxRepository {
     /** 测试辅助：模拟上游直接往信箱写报文（不经过本系统）。 */
     fun simulateExternalWrite(rawXml: String): Long = insertRaw(rawXml)
 
+    /** 测试辅助：模拟先分配 ID、稍后才可见的迟提交行。 */
+    fun restoreRow(msgId: Long, rawXml: String, receivedAt: Instant = Instant.now()) {
+        raws[msgId] = rawXml
+        received[msgId] = receivedAt
+    }
+
     /** 测试辅助：模拟库方清除这条行，用来构造"原文读不到"和 ID 缺口。 */
     fun removeRow(msgId: Long) {
         raws.remove(msgId); received.remove(msgId); marks.remove(msgId)

src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt

@@ -54,7 +54,8 @@ class InboxPoller(
         var committedTo = contiguous
         var holeSince: Instant? = null
         if (rows.last().msgId > contiguous) {
-            val since = watermark.holeSince ?: now
+            // 连续上界已前移说明旧空洞已补齐；后面是新空洞，不得继承旧等待时间。
+            val since = if (contiguous == watermark.committedUpTo) watermark.holeSince ?: now else now
             if (Duration.between(since, now) < props.pipeline.maxCommitDelay) {
                 holeSince = since
             } else {

src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxService.kt

@@ -21,10 +21,11 @@ class InboxService(
 
     data class Receipt(val msgId: Long, val receivedAt: Instant)
 
-    fun accept(rawXml: String, now: Instant = Instant.now()): Receipt {
+    fun accept(rawXml: String): Receipt {
         val id = inbox.insertRaw(rawXml)
-        procState.insertIfAbsent(id, now)
+        val receivedAt = checkNotNull(inbox.receivedAtOf(id)) { "mailbox receive time missing for msgId=$id" }
+        procState.insertIfAbsent(id, receivedAt)
         log.info("compat-accepted msgId={}", id)
-        return Receipt(id, now)
+        return Receipt(id, receivedAt)
     }
 }

src/main/kotlin/com/gzzn/omms/msgexchange/processing/BackfillService.kt

@@ -2,7 +2,9 @@ package com.gzzn.omms.msgexchange.processing
 
 import com.gzzn.omms.msgexchange.config.MailboxProps
 import com.gzzn.omms.msgexchange.config.PipelineProps
+import com.gzzn.omms.msgexchange.domain.ProcStatus
 import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository
+import com.gzzn.omms.msgexchange.infra.persistence.MailboxMarkResult
 import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository
 import jakarta.inject.Singleton
 import java.time.Clock
@@ -32,12 +34,14 @@ class BackfillService(
     private val mailboxProps: MailboxProps,
     private val props: PipelineProps,
     private val clock: Clock,
+    private val lifecycleGate: MessageLifecycleGate = MessageLifecycleGate(),
 ) {
     private val log = org.slf4j.LoggerFactory.getLogger(BackfillService::class.java)
 
     companion object {
         private val INITIAL_BACKOFF: Duration = Duration.ofSeconds(30)
         private val MAX_BACKOFF: Duration = Duration.ofMinutes(15)
+        private val TERMINAL_STATES = setOf(ProcStatus.SUCCEEDED, ProcStatus.SKIPPED, ProcStatus.DEAD)
 
         fun backoffDelayFor(attempts: Int): Duration {
             val shift = (attempts - 1).coerceIn(0, 20)
@@ -50,8 +54,13 @@ class BackfillService(
      * 调用方是主泵的处理路径，不能被回填问题拖住。
      */
     fun attempt(msgId: Long, now: Instant = clock.instant()) {
-        record(msgId, attempts = 0, now = now)?.let {
-            log.warn("backfill failed msgId={} error={} (sweep will retry)", msgId, it)
+        lifecycleGate.exclusive {
+            val row = procState.find(msgId) ?: return@exclusive
+            if (row.state !in TERMINAL_STATES) return@exclusive
+            if (row.backfillAt != null) return@exclusive
+            record(msgId, attempts = row.backfillAttempts, now = now)?.let {
+                log.warn("backfill failed msgId={} error={} (sweep will retry)", msgId, it)
+            }
         }
     }
 
@@ -62,7 +71,7 @@ class BackfillService(
      */
     fun sweep(now: Instant = clock.instant()): Int {
         val due = procState.findBackfillDue(now, now.minus(props.pipeline.overdueBackfill), props.pipeline.backfillBatch)
-        due.forEach { record(it.msgId, it.attempts, now) }
+        due.forEach { attempt(it.msgId, now) }
         return due.size
     }
 
@@ -70,7 +79,8 @@ class BackfillService(
     private fun record(msgId: Long, attempts: Int, now: Instant): String? =
         try {
             // 没有真正写进去说明库里已经有标记了，同样算成功（不覆盖已有值）
-            mailbox.markProcessedIfUnmarked(msgId, mailboxProps.processedValue)
+            val result = mailbox.markProcessedIfUnmarked(msgId, mailboxProps.processedValue)
+            check(result != MailboxMarkResult.MISSING) { "mailbox-row-missing" }
             procState.markBackfilled(msgId, now)
             null
         } catch (e: Exception) {
@@ -80,4 +90,5 @@ class BackfillService(
             }.onFailure { log.error("record backfill failure failed msgId={}", msgId, it) }
             reason
         }
+
 }

src/main/kotlin/com/gzzn/omms/msgexchange/processing/MessageLifecycleGate.kt

@@ -0,0 +1,17 @@
+package com.gzzn.omms.msgexchange.processing
+
+import jakarta.inject.Singleton
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.withLock
+
+/**
+ * 串行化“终态回填”与“人工重放”，避免旧回填任务在消息重新入队后写入信箱标记。
+ * 生产只允许单活动实例；该锁处理同一实例内的 job 线程和管理请求并发。
+ */
+@Singleton
+class MessageLifecycleGate {
+    private val lock = ReentrantLock()
+
+    fun <T> exclusive(block: () -> T): T = lock.withLock(block)
+}
+

src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt

@@ -66,33 +66,39 @@ class Pump(
         val head = procState.headUnfinished()
         when {
             head == null -> sleepQuietly(props.pipeline.pollInterval)
+            head.state == ProcStatus.FAILED && poisoned(head) -> {
+                log.error("poison -> DEAD msgId={} attempts={} lastError={}", head.msgId, head.attempts, head.lastError)
+                procState.markTerminal(
+                    head.msgId, ProcStatus.DEAD,
+                    errorClass = ErrorClass.EXHAUSTED,
+                    lastError = head.lastError ?: "head-deadline-exceeded",
+                    attempts = head.attempts,
+                    now = clock.instant(),
+                )
+                backfill.attempt(head.msgId)
+            }
             head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > clock.instant() ->
-                if (poisoned(head)) {
-                    log.error("poison -> DEAD msgId={} attempts={} lastError={}", head.msgId, head.attempts, head.lastError)
-                    procState.markTerminal(
-                        head.msgId, ProcStatus.DEAD,
-                        errorClass = ErrorClass.EXHAUSTED,
-                        lastError = head.lastError ?: "head-deadline-exceeded",
-                        attempts = head.attempts,
-                    )
-                    backfill.attempt(head.msgId)
-                } else {
-                    sleepQuietly(Duration.between(clock.instant(), head.nextAttemptAt))
-                }
+                sleepQuietly(Duration.between(clock.instant(), head.nextAttemptAt))
             // 其余情况（新消息，或退避到期的重试）交给处理入口
-            else -> processor.processOne(head)
+            else -> {
+                procState.markProcessingStartedIfAbsent(head.msgId, clock.instant())
+                processor.processOne(head)
+            }
         }
     }
 
     private fun poisoned(head: ProcState): Boolean =
-        head.attempts >= props.pipeline.maxAttempts ||
-            Duration.between(head.updatedAt, clock.instant()) > props.pipeline.headDeadline
+        isHeadPoisoned(head, clock.instant(), props)
 
     private fun sleepQuietly(d: Duration) {
         if (!d.isNegative && !d.isZero) Thread.sleep(d.toMillis().coerceAtLeast(1))
     }
 }
 
+internal fun isHeadPoisoned(head: ProcState, now: Instant, props: PipelineProps): Boolean =
+    head.attempts >= props.pipeline.maxAttempts ||
+        Duration.between(head.processingStartedAt ?: head.updatedAt, now) >= props.pipeline.headDeadline
+
 /**
  * 处理一条消息：读原文 → 解码 → 绑定业务身份 → 分派给对应处理器 → 提交后回填标记。
  *

src/main/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessor.kt

@@ -95,7 +95,11 @@ class ScheduleProcessor(
         val ok = validation as SnapshotValidation.Ok
 
         if (ok.perRecordDay.isEmpty()) {
-            // 报文合法但没有记录：不需要写数据，照样算处理成功
+            // 报文合法但没有记录：不写航班，但终态与回填意图仍在锁事务内一起提交
+            txManager.inTransaction {
+                lock.lock()
+                procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED)
+            }
             logSnapshot(head, body, SnapshotResult.COMMITTED, upserted = 0, setOf(SnapshotFlag.EMPTY), started)
             return ApplyResult.Succeeded
         }

src/main/resources/db/migration/V3__stable_processing_start.sql

@@ -0,0 +1,3 @@
+-- HOL deadline 使用首次开始处理的稳定时刻，不再被每次重试更新的 UPDATED_AT 推后。
+ALTER TABLE PROC_STATE ADD COLUMN PROCESSING_STARTED_AT TIMESTAMP(6) WITH TIME ZONE;
+

src/test/kotlin/com/gzzn/omms/msgexchange/delivery/DispatcherTickTest.kt

@@ -33,7 +33,7 @@ class DispatcherTickTest {
         MsgEvent(eventId = id, target = target, partitionKey = key, stateVersion = version, payloadJson = payload)
 
     @Test
-    fun `per-target tick delivers KAFKA_MSG only and never consumes schd`() {
+    fun `first tick delivers msg and starts the independent schd flush`() {
         val repo = StubMsgEvents()
         val port = StubDeliveryPort()
         val props = PipelineProps().apply { schd.flushPeriod = java.time.Duration.ofHours(1) }
@@ -47,7 +47,19 @@ class DispatcherTickTest {
         d.tick()
 
         assertEquals(1, port.sent.count { it.topic == "msg" })
-        assertEquals(0, port.sent.count { it.topic == "schd" })   // KAFKA_SCHD 只能由 flushSchd 发，tick 不许碰
+        assertEquals(1, port.sent.count { it.topic == "schd" })
+    }
+
+    @Test
+    fun `first tick starts schd delivery without a prior manual flush`() {
+        val repo = StubMsgEvents()
+        val port = StubDeliveryPort()
+        val props = PipelineProps().apply { pipeline.pollInterval = java.time.Duration.ofMillis(1) }
+        repo.insertAll(listOf(ev(1, Targets.KAFKA_SCHD, "F1", """{"v":1}""", 1)))
+
+        dispatcher(repo, port, props).tick()
+
+        assertEquals(1, port.sent.count { it.topic == "schd" })
     }
 
     @Test
@@ -70,6 +82,20 @@ class DispatcherTickTest {
         assertEquals(0, repo.rows.values.count { it.state.name == "PENDING" })
     }
 
+    @Test
+    fun `schd retry is not selected before its backoff expires`() {
+        val repo = StubMsgEvents()
+        val port = FailingSchdPort()
+        val d = dispatcher(repo, port)
+        repo.insertAll(listOf(ev(1, Targets.KAFKA_SCHD, "F1", """{"v":1}""", 1)))
+
+        d.flushSchd()
+        d.flushSchd()
+
+        assertEquals(1, port.calls)
+        assertEquals(1, repo.rows.values.single().attempts)
+    }
+
     @Test
     fun `tombstone is delivered as null value message`() {
         val repo = StubMsgEvents()

src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt

@@ -10,7 +10,7 @@ import java.sql.DriverManager
 
 /**
  * 在真实 PostgreSQL 上跑一遍迁移，确认结果符合预期：
- * V1 基线加 V2 信箱生命周期都能成功执行、迁移记录显示成功、该建的表和单行种子
+ * V1 基线、V2 信箱生命周期和 V3 稳定处理起点都能成功执行，迁移记录显示成功，该建的表和单行种子
  * （PIPELINE_LOCK、INBOX_CURSOR）都在，回填相关字段进了 PROC_STATE、BACKFILL_TODO 已下线。
  *
  * 没有可用的 PostgreSQL 时跳过（不假装通过）。
@@ -40,7 +40,7 @@ class FlywayMigrationTest {
 
         DriverManager.getConnection(url, user, pass).use { conn ->
             conn.createStatement().use { stmt ->
-                // 迁移记录：V1 基线 + V2 信箱生命周期
+                // 迁移记录：V1 基线 + V2 信箱生命周期 + V3 稳定处理起点
                 stmt.executeQuery(
                     "SELECT version, script, success FROM flyway_schema_history ORDER BY installed_rank ASC",
                 ).use { rs ->
@@ -48,11 +48,13 @@ class FlywayMigrationTest {
                     while (rs.next()) {
                         records.add(Triple(rs.getString("version"), rs.getString("script"), rs.getBoolean("success")))
                     }
-                    assertTrue(records.size >= 2, "flyway_schema_history must record both migrations")
+                    assertTrue(records.size >= 3, "flyway_schema_history must record all migrations")
                     assertEquals("1", records[0].first)
                     assertEquals("V1__flight_state_baseline.sql", records[0].second)
                     assertEquals("2", records[1].first)
                     assertEquals("V2__inbox_lifecycle.sql", records[1].second)
+                    assertEquals("3", records[2].first)
+                    assertEquals("V3__stable_processing_start.sql", records[2].second)
                     assertTrue(records.all { it.third })
                 }
 
@@ -81,12 +83,12 @@ class FlywayMigrationTest {
                 stmt.executeQuery(
                     "SELECT column_name FROM information_schema.columns WHERE table_name = 'proc_state' " +
                         "AND column_name IN ('received_at', 'backfill_at', 'backfill_next_at', " +
-                        "'backfill_attempts', 'backfill_error')",
+                        "'backfill_attempts', 'backfill_error', 'processing_started_at')",
                 ).use { rs ->
                     val cols = mutableSetOf<String>()
                     while (rs.next()) cols.add(rs.getString("column_name"))
                     assertEquals(
-                        setOf("received_at", "backfill_at", "backfill_next_at", "backfill_attempts", "backfill_error"),
+                        setOf("received_at", "backfill_at", "backfill_next_at", "backfill_attempts", "backfill_error", "processing_started_at"),
                         cols,
                     )
                 }

src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/InboxLifecycleJdbcSqlTest.kt

@@ -3,6 +3,7 @@ package com.gzzn.omms.msgexchange.infra.persistence.jdbc
 import com.gzzn.omms.msgexchange.domain.ErrorClass
 import com.gzzn.omms.msgexchange.domain.ProcStatus
 import com.gzzn.omms.msgexchange.infra.persistence.InboxCursorRepository
+import com.gzzn.omms.msgexchange.infra.persistence.MailboxMarkResult
 import org.h2.jdbcx.JdbcDataSource
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertFalse
@@ -50,6 +51,7 @@ class InboxLifecycleJdbcSqlTest {
                 st.execute(PROC_STATE_DDL)
                 st.execute(CURSOR_DDL)
                 st.execute(MAILBOX_DDL)
+                st.execute(EVENT_DDL)
                 st.execute(
                     "INSERT INTO inbox_cursor (cursor_id, committed_up_to, hole_since, updated_at) " +
                         "VALUES (1, 0, NULL, CURRENT_TIMESTAMP)",
@@ -137,6 +139,33 @@ class InboxLifecycleJdbcSqlTest {
         assertEquals(InboxCursorRepository.Cursor(42L, t0), cursor.load())
     }
 
+    @Test
+    fun `processing start is written once and is not refreshed`() {
+        seed(11L, t0)
+        proc.markProcessingStartedIfAbsent(11L, t0.plusSeconds(10))
+        proc.markProcessingStartedIfAbsent(11L, t0.plusSeconds(20))
+
+        assertEquals(t0.plusSeconds(10), proc.find(11L)!!.processingStartedAt)
+    }
+
+    @Test
+    fun `pipeline transaction rolls back terminal and outbox writes after a late failure`() {
+        seed(11L, t0)
+        val tx = JdbcPipelineTransactionManager(ds)
+
+        org.junit.jupiter.api.Assertions.assertThrows(IllegalStateException::class.java) {
+            tx.inTransaction {
+                ds.update("INSERT INTO msg_event_test (event_id) VALUES (1)", {})
+                proc.markTerminal(11L, ProcStatus.SUCCEEDED, now = t0)
+                error("late-failure")
+            }
+        }
+
+        assertEquals(ProcStatus.PENDING, proc.find(11L)!!.state)
+        assertNull(proc.find(11L)!!.backfillNextAt)
+        assertEquals(0, ds.queryOne("SELECT count(*) AS n FROM msg_event_test", {}) { it.getInt("n") })
+    }
+
     @Test
     fun `mailbox range read ignores processing marks and marking never overwrites`() {
         val first = mailbox.insertRaw("<A/>")
@@ -149,12 +178,12 @@ class InboxLifecycleJdbcSqlTest {
         assertEquals(listOf(second, third), mailbox.readRange(first, 50).map { it.msgId })
         assertEquals(third, mailbox.maxId())
 
-        assertTrue(mailbox.markProcessedIfUnmarked(second, "PROCESSED"))
-        assertFalse(mailbox.markProcessedIfUnmarked(second, "OTHER"))   // 已有标记，不覆盖
+        assertEquals(MailboxMarkResult.MARKED, mailbox.markProcessedIfUnmarked(second, "PROCESSED"))
+        assertEquals(MailboxMarkResult.ALREADY_MARKED, mailbox.markProcessedIfUnmarked(second, "OTHER"))
         assertEquals("PROCESSED", statusOf(second))
         // 已标记行仍出现在区间读结果中（发现与标记彻底解耦）
         assertEquals(listOf(first, second, third), mailbox.readRange(0L, 50).map { it.msgId })
-        assertFalse(mailbox.markProcessedIfUnmarked(999L, "PROCESSED"))
+        assertEquals(MailboxMarkResult.MISSING, mailbox.markProcessedIfUnmarked(999L, "PROCESSED"))
     }
 
     private fun seed(msgId: Long, receivedAt: Instant?, state: ProcStatus = ProcStatus.PENDING) {
@@ -194,6 +223,7 @@ class InboxLifecycleJdbcSqlTest {
               backfill_next_at  TIMESTAMP WITH TIME ZONE,
               backfill_attempts INT NOT NULL DEFAULT 0,
               backfill_error    VARCHAR(512),
+              processing_started_at TIMESTAMP WITH TIME ZONE,
               updated_at        TIMESTAMP WITH TIME ZONE NOT NULL,
               CONSTRAINT uk_proc_identity UNIQUE (identity_key)
             )
@@ -217,5 +247,11 @@ class InboxLifecycleJdbcSqlTest {
               CMINMSGS_STATUS         VARCHAR(32)
             )
         """
+
+        const val EVENT_DDL = """
+            CREATE TABLE msg_event_test (
+              event_id BIGINT PRIMARY KEY
+            )
+        """
     }
 }

src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcOpsTest.kt

@@ -0,0 +1,31 @@
+package com.gzzn.omms.msgexchange.infra.persistence.jdbc
+
+import org.h2.jdbcx.JdbcDataSource
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import java.util.UUID
+
+class JdbcOpsTest {
+    @Test
+    fun `transaction connection is scoped to its data source`() {
+        val first = dataSource("first")
+        val second = dataSource("second")
+        first.update("CREATE TABLE marker (marker_value INT)", {})
+        second.update("CREATE TABLE marker (marker_value INT)", {})
+        second.update("INSERT INTO marker (marker_value) VALUES (2)", {})
+
+        first.withTransaction {
+            first.update("INSERT INTO marker (marker_value) VALUES (1)", {})
+            val fromSecond = second.queryOne("SELECT marker_value FROM marker", {}) { it.getInt(1) }
+            assertEquals(2, fromSecond)
+        }
+
+        assertEquals(1, first.queryOne("SELECT marker_value FROM marker", {}) { it.getInt(1) })
+    }
+
+    private fun dataSource(label: String) = JdbcDataSource().apply {
+        setURL("jdbc:h2:mem:$label-${UUID.randomUUID()};DB_CLOSE_DELAY=-1;MODE=PostgreSQL")
+        user = "sa"
+        password = ""
+    }
+}

src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt

@@ -20,13 +20,17 @@ class ReplayServiceTest {
         val requeueCalls = mutableListOf<List<ErrorClass>>()
 
         fun seed(id: Long, status: ProcStatus, ec: ErrorClass?) {
-            rows[id] = ProcState(id, status, identityKey = "k$id", attempts = 3, errorClass = ec, lastError = "x")
+            rows[id] = ProcState(
+                id, status, identityKey = "k$id", attempts = 3, errorClass = ec, lastError = "x",
+                processingStartedAt = Instant.parse("2026-09-08T03:00:00Z"),
+            )
         }
 
         override fun insertIfAbsent(msgId: Long, receivedAt: Instant?): Boolean = rows.putIfAbsent(msgId, ProcState(msgId, ProcStatus.PENDING, receivedAt = receivedAt)) == null
         override fun find(msgId: Long): ProcState? = rows[msgId]
         override fun findSuccessTerminal(msgId: Long): Boolean = rows[msgId]?.state == ProcStatus.SUCCEEDED
         override fun headUnfinished(): ProcState? = null
+        override fun markProcessingStartedIfAbsent(msgId: Long, now: Instant) = Unit
         override fun tryBindIdentity(msgId: Long, identityKey: String): Boolean = true
         override fun ownerOfIdentity(identityKey: String): Long? = null
         override fun update(
@@ -55,7 +59,7 @@ class ReplayServiceTest {
                 val s = rows[id]!!
                 if (s.errorClass != null && s.errorClass in errorClasses &&
                     (s.state == ProcStatus.FAILED || s.state == ProcStatus.DEAD)) {
-                    rows[id] = s.copy(state = ProcStatus.PENDING, attempts = 0, nextAttemptAt = null)
+                    rows[id] = s.copy(state = ProcStatus.PENDING, attempts = 0, nextAttemptAt = null, processingStartedAt = null)
                     n++
                 }
             }
@@ -77,6 +81,7 @@ class ReplayServiceTest {
         assertEquals(ProcStatus.PENDING, repo.rows[1]!!.state)
         assertEquals(0, repo.rows[1]!!.attempts)
         assertNull(repo.rows[1]!!.nextAttemptAt)
+        assertNull(repo.rows[1]!!.processingStartedAt)
         assertEquals(ProcStatus.DEAD, repo.rows[2]!!.state)   // MALFORMED 永不被重放
         assertEquals(ProcStatus.PENDING, repo.rows[3]!!.state)
     }

src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPollerTest.kt

@@ -112,10 +112,36 @@ class InboxPollerTest {
 
     @Test
     fun `compat http path and poller do not double enqueue the same message`() {
-        val receipt = InboxService(inbox, proc).accept("<MSG/>", t0)
+        val receipt = InboxService(inbox, proc).accept("<MSG/>")
 
         assertEquals(0, poller.pollOnce(t0))
         assertEquals(receipt.msgId, cursor.cursor.committedUpTo)   // 已在 PG：读取进度照常推进
         assertNotNull(proc.find(receipt.msgId))
     }
+
+    @Test
+    fun `a newly exposed hole does not inherit the age of the previous hole`() {
+        val first = inbox.simulateExternalWrite("<MSG/>")
+        val oldHole = inbox.simulateExternalWrite("<MSG/>")
+        val third = inbox.simulateExternalWrite("<MSG/>")
+        val newHole = inbox.simulateExternalWrite("<MSG/>")
+        val fifth = inbox.simulateExternalWrite("<MSG/>")
+        inbox.removeRow(oldHole)
+        inbox.removeRow(newHole)
+
+        assertEquals(1, poller.pollOnce(t0))
+        val almostAged = t0.plus(props.pipeline.maxCommitDelay).minusSeconds(1)
+        inbox.restoreRow(oldHole, "<MSG/>", almostAged)
+
+        assertEquals(2, poller.pollOnce(almostAged))
+        assertEquals(third, cursor.cursor.committedUpTo)
+        assertEquals(almostAged, cursor.cursor.holeSince)
+        assertNull(proc.find(fifth))
+
+        // 旧空洞的期限已到，但新空洞必须获得完整等待窗口。
+        assertEquals(0, poller.pollOnce(t0.plus(props.pipeline.maxCommitDelay)))
+        assertEquals(third, cursor.cursor.committedUpTo)
+        assertNull(proc.find(fifth))
+        assertEquals(first + 2, third)
+    }
 }

src/test/kotlin/com/gzzn/omms/msgexchange/processing/BackfillServiceTest.kt

@@ -6,18 +6,26 @@ import com.gzzn.omms.msgexchange.domain.ErrorClass
 import com.gzzn.omms.msgexchange.domain.ProcStatus
 import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository
 import com.gzzn.omms.msgexchange.infra.persistence.MailboxRow
+import com.gzzn.omms.msgexchange.infra.persistence.MailboxMarkResult
 import com.gzzn.omms.msgexchange.infra.stub.StubInbox
 import com.gzzn.omms.msgexchange.infra.stub.StubProcState
+import com.gzzn.omms.msgexchange.infra.retry.ReplayService
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertFalse
 import org.junit.jupiter.api.Assertions.assertNotNull
 import org.junit.jupiter.api.Assertions.assertNull
 import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Assertions.assertThrows
 import org.junit.jupiter.api.Test
 import java.time.Clock
 import java.time.Duration
 import java.time.Instant
 import java.time.ZoneOffset
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Callable
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.TimeoutException
 
 /**
  * 回填环节的规矩：
@@ -37,11 +45,12 @@ class BackfillServiceTest {
         val marked = linkedSetOf<Long>()
         override fun insertRaw(rawXml: String): Long = 1L
         override fun rawOf(msgId: Long): String? = null
+        override fun receivedAtOf(msgId: Long): Instant? = null
         override fun readRange(fromExclusive: Long, limit: Int): List<MailboxRow> = emptyList()
         override fun maxId(): Long? = null
-        override fun markProcessedIfUnmarked(msgId: Long, value: String): Boolean {
+        override fun markProcessedIfUnmarked(msgId: Long, value: String): MailboxMarkResult {
             if (fail) throw IllegalStateException("mysql-down")
-            return marked.add(msgId)
+            return if (marked.add(msgId)) MailboxMarkResult.MARKED else MailboxMarkResult.ALREADY_MARKED
         }
     }
 
@@ -75,8 +84,8 @@ class BackfillServiceTest {
         val inbox = StubInbox().apply { clear() }
         val id = inbox.insertRaw("<MSG/>")
 
-        assertTrue(inbox.markProcessedIfUnmarked(id, "PROCESSED"))
-        assertFalse(inbox.markProcessedIfUnmarked(id, "OTHER"))   // 已经有标记了，不再写第二次
+        assertEquals(MailboxMarkResult.MARKED, inbox.markProcessedIfUnmarked(id, "PROCESSED"))
+        assertEquals(MailboxMarkResult.ALREADY_MARKED, inbox.markProcessedIfUnmarked(id, "OTHER"))
         assertEquals("PROCESSED", inbox.markOf(id))
     }
 
@@ -89,10 +98,12 @@ class BackfillServiceTest {
 
         val backfill = service(proc, inbox)
         backfill.attempt(id)
+        val firstCompletion = proc.find(id)!!.backfillAt
         backfill.attempt(id)   // 重复补写没有副作用
 
         assertNull(proc.find(id)!!.backfillError)
         assertEquals(0, proc.find(id)!!.backfillAttempts)
+        assertEquals(firstCompletion, proc.find(id)!!.backfillAt)
     }
 
     @Test
@@ -111,6 +122,23 @@ class BackfillServiceTest {
         assertNull(row.backfillAt)
     }
 
+    @Test
+    fun `missing mailbox row remains an unconfirmed backfill failure`() {
+        val proc = StubProcState()
+        val inbox = StubInbox().apply { clear() }
+        val id = inbox.insertRaw("<MSG/>")
+        succeeded(proc, id)
+        inbox.removeRow(id)
+
+        service(proc, inbox).attempt(id)
+
+        val row = proc.find(id)!!
+        assertNull(row.backfillAt)
+        assertEquals(1, row.backfillAttempts)
+        assertEquals("mailbox-row-missing", row.backfillError)
+        assertNotNull(row.backfillNextAt)
+    }
+
     @Test
     fun `sweep retries due rows and completes once the mailbox recovers`() {
         val proc = StubProcState()
@@ -163,6 +191,25 @@ class BackfillServiceTest {
         assertFalse(inbox.isMarked(failed))
     }
 
+    @Test
+    fun `immediate attempt also refuses pending and failed messages`() {
+        val proc = StubProcState()
+        val inbox = StubInbox().apply { clear() }
+        val pending = inbox.insertRaw("<MSG/>")
+        val failed = inbox.insertRaw("<MSG/>")
+        proc.insertIfAbsent(pending, t0)
+        proc.insertIfAbsent(failed, t0)
+        proc.update(failed, ProcStatus.FAILED, errorClass = ErrorClass.INFRA)
+
+        service(proc, inbox).attempt(pending)
+        service(proc, inbox).attempt(failed)
+
+        assertFalse(inbox.isMarked(pending))
+        assertFalse(inbox.isMarked(failed))
+        assertNull(proc.find(pending)!!.backfillAt)
+        assertNull(proc.find(failed)!!.backfillAt)
+    }
+
     @Test
     fun `backoff delay doubles per attempt and caps at fifteen minutes`() {
         assertEquals(Duration.ofSeconds(30), BackfillService.backoffDelayFor(1))
@@ -170,4 +217,49 @@ class BackfillServiceTest {
         assertEquals(Duration.ofMinutes(15), BackfillService.backoffDelayFor(10))
         assertEquals(Duration.ofMinutes(15), BackfillService.backoffDelayFor(50))
     }
+
+    @Test
+    fun `replay waits for an in-flight backfill before reopening the message`() {
+        val entered = CountDownLatch(1)
+        val release = CountDownLatch(1)
+        val replayStarted = CountDownLatch(1)
+        val mailbox = object : CminmsgInboxRepository {
+            override fun insertRaw(rawXml: String) = 1L
+            override fun rawOf(msgId: Long): String? = "<MSG/>"
+            override fun receivedAtOf(msgId: Long) = t0
+            override fun readRange(fromExclusive: Long, limit: Int) = emptyList<MailboxRow>()
+            override fun maxId(): Long? = 1L
+            override fun markProcessedIfUnmarked(msgId: Long, value: String): MailboxMarkResult {
+                entered.countDown()
+                release.await()
+                return MailboxMarkResult.MARKED
+            }
+        }
+        val proc = StubProcState().apply {
+            insertIfAbsent(1L, t0)
+            markTerminal(1L, ProcStatus.DEAD, errorClass = ErrorClass.EXHAUSTED, now = t0)
+        }
+        val gate = MessageLifecycleGate()
+        val backfill = BackfillService(proc, mailbox, MailboxProps(), props, Clock.fixed(t0, ZoneOffset.UTC), gate)
+        val replay = ReplayService(proc, gate)
+        val pool = Executors.newFixedThreadPool(2)
+        try {
+            val backfillTask = pool.submit { backfill.attempt(1L) }
+            assertTrue(entered.await(1, TimeUnit.SECONDS))
+            val replayTask = pool.submit(Callable {
+                replayStarted.countDown()
+                replay.replay(listOf(ErrorClass.EXHAUSTED))
+            })
+            assertTrue(replayStarted.await(1, TimeUnit.SECONDS))
+            assertThrows(TimeoutException::class.java) { replayTask.get(100, TimeUnit.MILLISECONDS) }
+
+            release.countDown()
+            backfillTask.get(1, TimeUnit.SECONDS)
+            assertEquals(1, replayTask.get(1, TimeUnit.SECONDS))
+            assertEquals(ProcStatus.PENDING, proc.find(1L)!!.state)
+        } finally {
+            release.countDown()
+            pool.shutdownNow()
+        }
+    }
 }

src/test/kotlin/com/gzzn/omms/msgexchange/processing/MessageLifecycleGateTest.kt

@@ -0,0 +1,44 @@
+package com.gzzn.omms.msgexchange.processing
+
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+
+class MessageLifecycleGateTest {
+    @Test
+    fun `replay cannot enter while a backfill operation owns the lifecycle gate`() {
+        val gate = MessageLifecycleGate()
+        val entered = CountDownLatch(1)
+        val release = CountDownLatch(1)
+        val replayStarted = CountDownLatch(1)
+        val replayEntered = CountDownLatch(1)
+        val pool = Executors.newFixedThreadPool(2)
+        try {
+            val backfill = pool.submit {
+                gate.exclusive {
+                    entered.countDown()
+                    release.await()
+                }
+            }
+            assertTrue(entered.await(1, TimeUnit.SECONDS))
+
+            val replay = pool.submit {
+                replayStarted.countDown()
+                gate.exclusive { replayEntered.countDown() }
+            }
+            assertTrue(replayStarted.await(1, TimeUnit.SECONDS))
+            assertFalse(replayEntered.await(100, TimeUnit.MILLISECONDS))
+
+            release.countDown()
+            backfill.get(1, TimeUnit.SECONDS)
+            replay.get(1, TimeUnit.SECONDS)
+            assertTrue(replayEntered.await(1, TimeUnit.SECONDS))
+        } finally {
+            release.countDown()
+            pool.shutdownNow()
+        }
+    }
+}

src/test/kotlin/com/gzzn/omms/msgexchange/processing/PumpDeadlineTest.kt

@@ -0,0 +1,40 @@
+package com.gzzn.omms.msgexchange.processing
+
+import com.gzzn.omms.msgexchange.config.PipelineProps
+import com.gzzn.omms.msgexchange.domain.ProcState
+import com.gzzn.omms.msgexchange.domain.ProcStatus
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+import java.time.Instant
+
+class PumpDeadlineTest {
+    private val props = PipelineProps()
+    private val started = Instant.parse("2026-09-08T03:00:00Z")
+
+    @Test
+    fun `deadline uses stable processing start rather than refreshed update time`() {
+        val row = ProcState(
+            msgId = 1,
+            state = ProcStatus.FAILED,
+            attempts = 1,
+            processingStartedAt = started,
+            updatedAt = started.plusSeconds(590),
+        )
+
+        assertTrue(isHeadPoisoned(row, started.plusSeconds(600), props))
+    }
+
+    @Test
+    fun `head below attempt and time limits remains retryable`() {
+        val row = ProcState(
+            msgId = 1,
+            state = ProcStatus.FAILED,
+            attempts = 1,
+            processingStartedAt = started,
+            updatedAt = started.plusSeconds(590),
+        )
+
+        assertFalse(isHeadPoisoned(row, started.plusSeconds(599), props))
+    }
+}

src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt

@@ -112,6 +112,19 @@ class ScheduleProcessorTest {
         assertNotNull(row.backfillNextAt)
     }
 
+    @Test
+    fun `valid empty schedule commits terminal state and backfill intent`() {
+        val proc = StubProcState()
+        proc.insertIfAbsent(msgId, null)
+        val body = ScheduleBody(recsDeclared = 0, records = emptyList())
+
+        val result = processor(proc = proc).applyScheduleRecords(head(), message(body))
+
+        assertEquals(ApplyResult.Succeeded, result)
+        assertEquals(ProcStatus.SUCCEEDED, proc.find(msgId)!!.state)
+        assertNotNull(proc.find(msgId)!!.backfillNextAt)
+    }
+
     @Test
     fun `replay of succeeded message records idempotent success without writes`() {
         val proc = StubProcState()