fix(jobs): 留痕清理独立接通历史存储开关；精简死逻辑并对齐陈旧注释

src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt

@@ -38,11 +38,11 @@ class PipelineProps {
         var maxCommitDelay: Duration = Duration.ofMinutes(5)
 
         /**
-         * 超期补写期限：一条消息处理完之后，如果过了这么久还是没能把处理标记写回信箱
-         * （比如回填一直失败），就直接强制补写一次，不再等退避。
+         * 超期补写期限 R：一条消息到达终态后，过了这么久处理标记仍未写回信箱
+         * （比如回填一直失败），扫描谓词的超期分支成立，无视退避强制补写（只会提前、从不推迟打标）。
          *
-         * 这是保证库方能清理信箱的兜底期限，必须覆盖人工重放所需的保留期，
-         * 确认之前不要调小，否则还在重放窗口内的消息会先被库方清掉。
+         * 唯一约束是 R ≤ R_keep，**不保护重放窗口**（打标时刻与 R 解耦，完整论证见
+         * message-lifecycle.md §5.2）。取值待 Q6 定案，确认之前不要为提速下调。
          */
         var overdueBackfill: Duration = Duration.ofDays(30)
 
@@ -104,7 +104,7 @@ class PipelineProps {
          * 配置自检：退避表档位数必须等于 `max-attempts − 1`。
          *
          * 因为 attempts 一达到 `max-attempts` 就转 `DEAD`、不再计算下次重试，能真正用到的档位
-         * 只有 `max-attempts − 1` 个。表更长会有一段**永远走不到**（默认配置里的 16 秒档就是这样），
+         * 只有 `max-attempts − 1` 个。表更长会有一段**永远走不到**，
          * 表更短则会提前封顶。两种错位都应该在启动时暴露，而不是静默生效。
          */
         fun validate() {

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

@@ -160,8 +160,6 @@ data class Backlog(
 interface MsgEventRepository {
     fun insertAll(events: List<MsgEvent>): List<Long>
 
-    fun headUnsent(target: String): MsgEvent?
-
     fun claimBatch(target: String, limit: Int): List<MsgEvent>
 
     /** 挑出待发的整态事件，同一条航班只取版本最高的那一条（中间的版本不用发）。 */
@@ -227,9 +225,6 @@ interface FlightStateRepository {
      * 历史存储没接通时，调用方必须传空集合，也就是一条都不删。
      */
     fun purgeArchived(flids: Collection<String>): Int
-
-    /** 运营日还没定下来的航班有多少条（观测用；这些航班暂时不参与清理）。 */
-    fun countOperationDayNull(): Int
 }
 
 /** 日计划处理留痕：只追加、不参与业务判断，一次处理（含重放）记一行，写失败不影响业务。 */

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

@@ -0,0 +1,12 @@
+package com.gzzn.omms.msgexchange.infra.persistence
+
+import java.time.Instant
+
+/**
+ * 留痕清理端口：物理删除指定时刻之前的 `SCHD_SNAP_LOG` 行。
+ *
+ * 只依赖自有 PG，不依赖历史存储开关——留痕是本地可重建记录，清理不需要外部归档确认。
+ */
+fun interface SnapshotLogPurge {
+    fun purgeBefore(instant: Instant): Int
+}

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

@@ -23,6 +23,7 @@ import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager
 import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository
 import com.gzzn.omms.msgexchange.infra.persistence.ReqTrackRepository
 import com.gzzn.omms.msgexchange.domain.SnapshotLogEntry
+import com.gzzn.omms.msgexchange.infra.persistence.SnapshotLogPurge
 import com.gzzn.omms.msgexchange.infra.persistence.SnapshotLogRepository
 import io.micronaut.context.annotation.Requires
 import jakarta.inject.Singleton
@@ -450,13 +451,6 @@ class JdbcMsgEventRepository(
             )
         }
 
-    override fun headUnsent(target: String): MsgEvent? =
-        ds.queryOne(
-            "SELECT * FROM msg_event WHERE target = ? AND state = 'PENDING' ORDER BY event_id ASC LIMIT 1",
-            { ps -> ps.setString(1, target) },
-            ::mapEvent,
-        )
-
     override fun claimBatch(target: String, limit: Int): List<MsgEvent> =
         ds.query(
             "SELECT * FROM msg_event WHERE target = ? AND state = 'PENDING' ORDER BY event_id ASC LIMIT ?",
@@ -682,9 +676,6 @@ class JdbcFlightStateRepository(
         return purged
     }
 
-    override fun countOperationDayNull(): Int =
-        ds.queryOne("SELECT count(*) AS n FROM flight_schd WHERE operation_day IS NULL", {}) { rs -> rs.getInt("n") } ?: 0
-
     // ---- 内部实现 ----
 
     private fun bindMain(
@@ -825,7 +816,7 @@ class JdbcFlightStateRepository(
 class JdbcSnapshotLogRepository(
     private val ds: DataSource,
     private val clock: Clock,
-) : SnapshotLogRepository {
+) : SnapshotLogRepository, SnapshotLogPurge {
     /** 只追加写；这里不抛异常，写失败由调用方捕获并记为指标。 */
     override fun append(entry: SnapshotLogEntry) {
         ds.update(
@@ -848,6 +839,17 @@ class JdbcSnapshotLogRepository(
             },
         )
     }
+
+    /** 留痕清理：收信时间与覆盖窗尾都早于分界点的行物理删除（走 idx_snaplog_cleanup）。 */
+    override fun purgeBefore(instant: Instant): Int =
+        ds.update(
+            "DELETE FROM schd_snap_log WHERE scope_end < ? AND recv_at < ?",
+            { ps ->
+                val cutoffDay = java.time.LocalDate.ofInstant(instant, java.time.ZoneId.of("Asia/Shanghai"))
+                ps.setDate(1, java.sql.Date.valueOf(cutoffDay))
+                ps.setTimestamp(2, instant.toSqlTimestamp())
+            },
+        )
 }
 
 @Singleton

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

@@ -20,8 +20,7 @@ class ProcFailure(
     private val procState: ProcStateRepository,
     val scheduler: FailureScheduler,
 ) {
-    /** @return 是否已经落到终态：DEAD 是终态，FAILED 还会再试 */
-    fun fail(head: ProcState, ec: ErrorClass, reason: String): Boolean {
+    fun fail(head: ProcState, ec: ErrorClass, reason: String) {
         val attempts = head.attempts + 1
         if (scheduler.exhausted(attempts)) {
             procState.markTerminal(
@@ -31,7 +30,6 @@ class ProcFailure(
                 lastError = "$reason; attempts=$attempts",
                 now = scheduler.now(),
             )
-            return true
         }
         procState.update(
             head.msgId, ProcStatus.FAILED,
@@ -40,6 +38,5 @@ class ProcFailure(
             errorClass = ec,
             lastError = reason,
         )
-        return false
     }
 }

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

@@ -19,6 +19,7 @@ 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.SnapshotLogPurge
 import com.gzzn.omms.msgexchange.infra.persistence.PersistOutcome
 import com.gzzn.omms.msgexchange.infra.persistence.PipelineLockRepository
 import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager
@@ -233,9 +234,6 @@ class StubMsgEvents : MsgEventRepository {
             id
         }
 
-    override fun headUnsent(target: String): MsgEvent? =
-        rows.values.filter { it.target == target && it.state == EventStatus.PENDING }.minByOrNull { it.eventId!! }
-
     override fun claimBatch(target: String, limit: Int): List<MsgEvent> =
         rows.values.filter { it.target == target && it.state == EventStatus.PENDING }.sortedBy { it.eventId!! }.take(limit)
 
@@ -344,12 +342,11 @@ class StubFlightState : FlightStateRepository {
         return n
     }
 
-    override fun countOperationDayNull(): Int = mains.values.count { it.operationDay == null }
 }
 
 @Singleton
 @Requires(property = "msgx.stubs", value = "true")
-class StubSnapshotLog : SnapshotLogRepository {
+class StubSnapshotLog : SnapshotLogRepository, SnapshotLogPurge {
     val entries = mutableListOf<SnapshotLogEntry>()
 
     fun clear() = entries.clear()
@@ -357,6 +354,12 @@ class StubSnapshotLog : SnapshotLogRepository {
     override fun append(entry: SnapshotLogEntry) {
         entries.add(entry)
     }
+
+    override fun purgeBefore(instant: Instant): Int {
+        val before = entries.size
+        entries.removeAll { it.recvAt < instant }
+        return before - entries.size
+    }
 }
 
 @Singleton

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

@@ -13,8 +13,8 @@ import java.time.Instant
  * 两步不在同一个事务里（跨库没有事务）。如果信箱写成功、PG 入队失败，原文仍然在信箱里，
  * 收报轮询会按 ID 把它补进来，所以不会丢消息。
  *
- * 【缺口】这里**不参与水位**：它直接写 PROC_STATE，因此当水位卡在低位空洞时，
- * 这条高 ID 会被主泵提前领取，破坏 FIFO。修复方向见 ACM2-37 / message-lifecycle.md §5.1。
+ * 该入口**不参与水位**：它直接写 PROC_STATE，登记的行可能超出水位；主泵只领 `msgId ≤ W`，
+ * 因此不破坏 FIFO——这类行等水位追平后按序自然领取（message-lifecycle.md §5.1）。
  */
 @Singleton
 class InboxService(

src/main/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJob.kt

@@ -8,6 +8,7 @@ import com.gzzn.omms.msgexchange.domain.flight.HistoryCandidate
 import com.gzzn.omms.msgexchange.domain.flight.HistoryRules
 import com.gzzn.omms.msgexchange.infra.persistence.FlightStateRepository
 import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository
+import com.gzzn.omms.msgexchange.infra.persistence.SnapshotLogPurge
 import jakarta.inject.Singleton
 import java.time.Duration
 import java.time.Instant
@@ -32,7 +33,7 @@ class HistorySweepJob(
     private val props: HistoryProps,
     /** 历史存储端口：返回确认归档成功的 FLID 集合；部署侧没接通时是 null。 */
     private val historyStore: HistoryStore? = null,
-    /** 留痕清理端口：删掉超过保留期（默认 90 天）的日计划留痕行；没接通时是 null。 */
+    /** 留痕清理端口：删掉超过保留期（默认 90 天）的日计划留痕行；不依赖历史存储开关，没接通时为 null。 */
     private val snapLogPurge: SnapshotLogPurge? = null,
 ) {
     /** 归档到历史存储的接口，由部署侧适配具体存储；默认不接通。 */
@@ -41,25 +42,24 @@ class HistorySweepJob(
         fun archive(candidates: List<HistoryCandidate>): Set<String>
     }
 
-    fun interface SnapshotLogPurge {
-        fun purgeBefore(instant: Instant): Int
-    }
-
     data class SweepOutcome(val selected: Int, val archived: Int, val purged: Int, val snapLogPurged: Int = 0)
 
     fun run(now: Instant = Instant.now()): SweepOutcome {
+        // 留痕是本地可重建记录，清理不需要外部归档确认，不受历史存储开关牵制。
+        val snapLogPurged = snapLogPurge?.purgeBefore(now.minus(Duration.ofDays(props.snapLogRetentionDays))) ?: 0
+
         if (!props.historyStoreEnabled || historyStore == null) {
-            // 红线：历史存储没接通就一条都不删。先删当前态、事后再补历史是不允许的。
-            return SweepOutcome(selected = 0, archived = 0, purged = 0)
+            // 红线：历史存储没接通就一条航班都不删。先删当前态、事后再补历史是不允许的。
+            return SweepOutcome(selected = 0, archived = 0, purged = 0, snapLogPurged = snapLogPurged)
         }
 
         val rules = HistoryRules(props.cancelledHours, props.terminalHours, props.deletedHours, props.idleHours)
         val zone = ZoneId.of("Asia/Shanghai")   // 保留期窗口按机场时区算，不用 UTC
         val candidates = flightState.findHistoryCandidates(rules, zone, now)
-        if (candidates.isEmpty()) return SweepOutcome(0, 0, 0)
+        if (candidates.isEmpty()) return SweepOutcome(0, 0, 0, snapLogPurged = snapLogPurged)
 
         val archivedFlids = historyStore.archive(candidates)
-        if (archivedFlids.isEmpty()) return SweepOutcome(candidates.size, archived = 0, purged = 0)
+        if (archivedFlids.isEmpty()) return SweepOutcome(candidates.size, archived = 0, purged = 0, snapLogPurged = snapLogPurged)
 
         val toPurge = candidates.filter { it.flid in archivedFlids }
         // 从没收到过 FDEL、被这里直接清掉的航班，删除前补发一次通知，否则下游一直以为它还在
@@ -68,8 +68,6 @@ class HistorySweepJob(
             msgEvents.insertAll(preDelete.map { tombstone(it) })
         }
         val purged = flightState.purgeArchived(archivedFlids)
-
-        val snapLogPurged = snapLogPurge?.purgeBefore(now.minus(Duration.ofDays(props.snapLogRetentionDays))) ?: 0
         return SweepOutcome(candidates.size, archivedFlids.size, purged, snapLogPurged)
     }
 

src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt

@@ -1,7 +1,5 @@
 package com.gzzn.omms.msgexchange.jobs
 
-import com.gzzn.omms.msgexchange.config.HistoryProps
-import com.gzzn.omms.msgexchange.config.PipelineProps
 import com.gzzn.omms.msgexchange.processing.BackfillService
 import jakarta.inject.Singleton
 import java.time.Clock
@@ -23,8 +21,6 @@ import java.time.ZoneId
 class JobRunner(
     private val backfill: BackfillService,
     private val historySweep: HistorySweepJob,
-    @Suppress("unused") private val pipelineProps: PipelineProps,
-    @Suppress("unused") private val historyProps: HistoryProps,
     private val clock: Clock,
 ) {
     private val log = org.slf4j.LoggerFactory.getLogger(JobRunner::class.java)

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

@@ -18,8 +18,7 @@ import java.time.Instant
  * 同时留下"还欠一次回填"的意图（PROC_STATE 上的 BACKFILL_NEXT_AT）；
  * 这个类负责第二件：
  *
- * - [attempt]：处理刚结束时马上试一次，让标记尽快落到信箱。失败也不影响处理结果，
- *   留给扫描重试即可。
+ * - [attempt]：单行写标记，只由 [sweep] 逐行调用——主泵与处理器不直接调它（回填一律扫描驱动）。
  * - [sweep]：定时把还欠回填的记录挑出来重试。失败就按 30 秒起步、最长 15 分钟的
  *   退避往后推；如果一条消息从收到现在已经超过超期期限，则无视退避强制补写——
  *   否则退避可能一直失败下去，这些行永远打不上标记，库方就没法清理信箱。

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

@@ -81,7 +81,7 @@ class Pump(
         //
         // 水位以内的行都是收报按 ID 顺序发现并登记的；水位之外的行只可能来自兼容入口
         // 直接写 PROC_STATE（它不参与水位）。若允许领取，它就会越过那些尚未入队的较小 ID，
-        // 破坏 FIFO（缺口 G2）。这种行在空洞补齐、`W` 追平之后自然可领取。
+         // 破坏 FIFO（不变量"只领取已发现的行"，message-lifecycle.md §11）。这种行在空洞补齐、`W` 追平之后自然可领取。
         val watermark = cursor.load().committedUpTo
         if (head.msgId > watermark) {
             warnBeyondWatermark(head.msgId, watermark)
@@ -175,21 +175,7 @@ class MessageProcessor(
         }
     }
 
-    /** @return 这条消息是否落到终态；回填由扫描驱动，调用方不再据此立即回填。 */
-    private fun processInternal(head: ProcState): Boolean {
-        // 守卫：手工/遗留 FAILED 行若 attempts 已达上限，直接终态（防止退避到期后无限重试）
-        if (head.state == ProcStatus.FAILED && procFailure.scheduler.exhausted(head.attempts)) {
-            log.error("head exhausted at entry -> DEAD msgId={} attempts={}", head.msgId, head.attempts)
-            procState.markTerminal(
-                head.msgId, ProcStatus.DEAD,
-                errorClass = ErrorClass.EXHAUSTED,
-                lastError = head.lastError ?: "max-attempts",
-                attempts = head.attempts,
-                now = clock.instant(),
-            )
-            return true
-        }
-
+    private fun processInternal(head: ProcState) {
         val raw = inbox.rawOf(head.msgId)
         if (raw == null) {
             log.error("raw missing -> DEAD(MALFORMED) msgId={}", head.msgId)
@@ -215,7 +201,7 @@ class MessageProcessor(
                 val owner = procState.ownerOfIdentity(identity) ?: -1L
                 log.info("duplicate-of:{} -> SKIPPED msgId={}", owner, head.msgId)
                 procState.markTerminal(head.msgId, ProcStatus.SKIPPED, lastError = "duplicate-of:$owner", now = clock.instant())
-                return true
+                return
             }
         }
 
@@ -260,16 +246,14 @@ class MessageProcessor(
                 lastError = result.reason.take(1000),
                 now = clock.instant(),
             )
-            return true
+            return
         }
 
         // Succeeded / ReplaySkipped：SUCCEEDED 终态与回填意图已由处理器在自己的事务内落库
         log.info("SUCCEEDED msgId={} kind={}", head.msgId, decoded.typeTag)
-        return true
     }
 
-    private fun deadMalformed(head: ProcState, detail: String): Boolean {
+    private fun deadMalformed(head: ProcState, detail: String) {
         procState.markTerminal(head.msgId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = detail, now = clock.instant())
-        return true
     }
 }

src/main/resources/application.yml

@@ -20,7 +20,7 @@ msgx:
     backoff-cap-ms: 60000
     head-deadline: 10m              # 队头滞留上界 = 最坏 HOL 时长（毒丸升级）
     max-commit-delay: 5m            # §5.1 空洞老化：W+1 空洞超过该时延判定为永久（Q2 最大提交时延）
-    overdue-backfill: 30d           # §5.2 超期补写期限 R：≥ 人工重放期限 + 人工处置期限（Q6）
+    overdue-backfill: 30d           # §5.2 超期补写期限 R（Q6）：仅须 R ≤ R_keep，不保护重放窗口（message-lifecycle §5.2）
     backfill-batch: 100             # 回填扫描单批条数
     backfill-max-attempts: 100      # 回填自动重试上限；达上限停止自动重试（可人工恢复），避免占满批次
     # 一次性切流播种：默认（注释掉）不播种。min=读现存全部 | zero=从 0 按空洞规则 | max=跳过可见存量 | <id>

src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt

@@ -219,7 +219,7 @@ class PipelineSmokeTest {
     }
 
     /**
-     * FIFO 修复（G2）端到端验收：兼容入口写入的高 ID **不会被提前领取**，
+     * FIFO 不变量"只领取已发现的行"端到端验收：兼容入口写入的高 ID **不会被提前领取**，
      * 必须等水位追平（较小 ID 补齐并入队）之后才按顺序处理。
      */
     @Test

src/test/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJobTest.kt

@@ -101,4 +101,21 @@ class HistorySweepJobTest {
         assertTrue(tombstone.payloadJson.contains("\"deleted\":true"))
         assertEquals(null, f.findMainRow("F3"))
     }
+
+    @Test
+    fun `snap log purge runs independently of the history store switch`() {
+        val f = seededFlight("F1", deleted = true, idleDays = 30)
+        var cutoff: Instant? = null
+        val purge = com.gzzn.omms.msgexchange.infra.persistence.SnapshotLogPurge { instant -> cutoff = instant; 7 }
+        val job = HistorySweepJob(
+            f, StubMsgEvents(), HistoryProps().apply { historyStoreEnabled = false }, snapLogPurge = purge,
+        )
+
+        val outcome = job.run(now)
+
+        assertEquals(7, outcome.snapLogPurged)      // 留痕清理不依赖历史存储开关
+        assertEquals(0, outcome.purged)             // 航班当前态仍一条不许删
+        assertTrue(f.findMainRow("F1") != null)
+        assertTrue(cutoff != null)
+    }
 }