feat(jobs): PROC_STATE 终态行到期清理（ACM2-82）

- 新增 deleteExpiredTerminal：仅删终态 + BACKFILL_AT 非空 + UPDATED_AT 超期的行（US-11） - ProcStateCleanupJob 每轮 tick 执行，与人工重放共用 MessageLifecycleGate，不持 PIPELINE_LOCK - 登记参数 msgx.pipeline.terminal-retention（默认 90d，暂定）进 reference.md - Stub 与 JDBC 实现同口径；补四类候选条件回归
2026-09-21 12:52:26 +08:00
parent 477f038cc1
commit 1d95926e06
10 changed files with 181 additions and 1 deletions
@@ -20,6 +20,7 @@
 | `msgx.schd.flush-period` | `3s` | 每隔多久尝试发送一批 `schd` 消息 | 沿用旧系统 |
 | `msgx.schd.flush-limit` | `500` 条 | 一批 `schd` 消息最多包含多少个航班 | 暂定 |
 | `msgx.pipeline.event-retention` | `7d` | 发送成功的消息在数据库保留多久，从 `SENT_AT` 起算 | 暂定 |
+| `msgx.pipeline.terminal-retention` | `90d` | 处理记录（PROC_STATE 终态行）保留多久，从 `UPDATED_AT` 起算；到期且已回填才删除（`US-11`） | 暂定 |

 ### 写回信箱与编号扫描

@@ -94,6 +94,13 @@ class PipelineProps {
         */
        var eventRetention: Duration = Duration.ofDays(7)

+        /**
+         * PROC_STATE 终态行（SUCCEEDED/SKIPPED/DEAD）的保留期（`US-11`）。
+         * 清理作业删除「终态且已回填、`UPDATED_AT < now - terminalRetention`」的行；
+         * 未完成或未回填的行不删。保留期内身份键仍可去重，到期删除后同身份报文不再判重复。
+         */
+        var terminalRetention: Duration = Duration.ofDays(90)
+
        /** N28：attempt ≤ 0（如 FAILED 未递增 attempts 的行）不得抛异常，取下界=首档退避。 */
        fun backoffFor(attempt: Int): Long {
            val index = (attempt - 1).coerceAtLeast(0)
@@ -126,6 +126,13 @@ interface ProcStateRepository {
    /** 人工重放：把指定错误类别的 FAILED / DEAD 记录改回 PENDING，重试次数清零。 */
    fun requeueByErrorClasses(errorClasses: List<ErrorClass>): Int

+    /**
+     * 到期清理（US-11）：删除终态（SUCCEEDED/SKIPPED/DEAD）、已回填（BACKFILL_AT 非空）
+     * 且 `UPDATED_AT < cutoff` 的 PROC_STATE 行；未完成或未回填的行一律不删。
+     * @return 实际删除的行数
+     */
+    fun deleteExpiredTerminal(cutoff: Instant, limit: Int): Int
+
    /** 积压观测：还没处理完的条数、最老一条的接收时间、处理完但还没回填的条数。 */
    fun backlog(): Backlog

@@ -288,6 +288,24 @@ class JdbcProcStateRepository(
        )
    }

+    /** 到期清理（US-11）：只删终态且已回填、UPDATED_AT 超期的行；条件带不进子查询的行不受影响。 */
+    override fun deleteExpiredTerminal(cutoff: Instant, limit: Int): Int =
+        ds.update(
+            """
+            DELETE FROM proc_state WHERE msg_id IN (
+                SELECT msg_id FROM proc_state
+                WHERE state IN ('SUCCEEDED', 'SKIPPED', 'DEAD')
+                  AND backfill_at IS NOT NULL
+                  AND updated_at < ?
+                ORDER BY msg_id LIMIT ?
+            )
+            """.trimIndent(),
+            { ps ->
+                ps.setTimestamp(1, cutoff.toSqlTimestamp())
+                ps.setInt(2, limit)
+            },
+        )
+
    /** 积压观测：未处理条数、最老一条的接收时间、处理完但未回填的条数。 */
    override fun hasAny(): Boolean =
        ds.queryOne("SELECT 1 FROM proc_state LIMIT 1", {}) { 1 } != null
@@ -225,6 +225,19 @@ class StubProcState(private val clock: Clock = Clock.systemUTC()) : ProcStateRep
        return n
    }

+    /** 到期清理（US-11）：仅删终态且已回填、UPDATED_AT 超期的行。 */
+    override fun deleteExpiredTerminal(cutoff: Instant, limit: Int): Int {
+        val due = rows.values
+            .filter {
+                it.state in TERMINAL_STATES && it.backfillAt != null && it.updatedAt.isBefore(cutoff)
+            }
+            .sortedBy { it.msgId }
+            .take(limit)
+            .map { it.msgId }
+        due.forEach { rows.remove(it) }
+        return due.size
+    }
+
    fun snapshotOf(msgId: Long): ProcState? = rows[msgId]
 }

@@ -549,3 +562,7 @@ class StubInboxCursor : InboxCursorRepository {
 /** 判断是不是终态：处理已经结束、不会再重试的状态。 */
 private fun ProcStatus.isTerminal(): Boolean =
    this == ProcStatus.SUCCEEDED || this == ProcStatus.SKIPPED || this == ProcStatus.DEAD
+
+/** 终态集合：US-11 到期清理与回填扫描共用的口径。 */
+private val TERMINAL_STATES: Set<ProcStatus> =
+    setOf(ProcStatus.SUCCEEDED, ProcStatus.SKIPPED, ProcStatus.DEAD)
@@ -25,6 +25,7 @@ class JobRunner(
    private val backfill: BackfillService,
    private val historySweep: HistorySweepJob,
    private val eventCleanup: EventCleanupJob,
+    private val procStateCleanup: ProcStateCleanupJob,
    private val clock: Clock,
    private val activity: JobActivity,
    private val props: PipelineProps,
@@ -82,6 +83,10 @@ class JobRunner(
            if (cleanupOutcome.deleted > 0) {
                log.info("event cleanup: deleted {} expired SENT rows", cleanupOutcome.deleted)
            }
+            val procCleanup = procStateCleanup.run(startedAt)
+            if (procCleanup.deleted > 0) {
+                log.info("proc_state cleanup: deleted {} expired terminal rows", procCleanup.deleted)
+            }
            activity.tickFinished(startedAt, (System.nanoTime() - startedNanos) / 1_000_000, selected)
        } catch (e: InterruptedException) {
            throw e
@@ -0,0 +1,35 @@
+package com.gzzn.omms.msgexchange.jobs
+
+import com.gzzn.omms.msgexchange.config.PipelineProps
+import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository
+import com.gzzn.omms.msgexchange.processing.MessageLifecycleGate
+import jakarta.inject.Singleton
+import java.time.Instant
+
+/**
+ * PROC_STATE 终态行到期清理（US-11）：删除「终态（SUCCEEDED/SKIPPED/DEAD）、
+ * 已回填（BACKFILL_AT 非空）且 `UPDATED_AT` 超过保留期」的处理记录。
+ *
+ * - 与人工重放共用 [MessageLifecycleGate]：重放把 DEAD/FAILED 翻回 PENDING 的瞬间，
+ *   本作业不得并发删除同一批行；反之亦然。
+ * - **不持 `PIPELINE_LOCK`**：清理不是 FIFO 关键路径，锁住会阻塞主泵。
+ * - 每轮 JobRunner tick 跑一次，单轮删除量有上限，避免长事务。
+ */
+@Singleton
+class ProcStateCleanupJob(
+    private val procState: ProcStateRepository,
+    private val lifecycleGate: MessageLifecycleGate,
+    private val props: PipelineProps,
+) {
+    data class CleanupOutcome(val deleted: Int)
+
+    fun run(now: Instant): CleanupOutcome {
+        val cutoff = now.minus(props.pipeline.terminalRetention)
+        val deleted = lifecycleGate.exclusive { procState.deleteExpiredTerminal(cutoff, DELETE_BATCH) }
+        return CleanupOutcome(deleted)
+    }
+
+    private companion object {
+        const val DELETE_BATCH = 500
+    }
+}
@@ -77,6 +77,8 @@ class ReplayServiceTest {
            }
            return n
        }
+
+        override fun deleteExpiredTerminal(cutoff: java.time.Instant, limit: Int): Int = 0
    }

    @Test
@@ -38,7 +38,9 @@ class JobRunnerTest {

    private fun runner(proc: ProcStateRepository, activity: JobActivity) = JobRunner(
        BackfillService(proc, StubInbox(), MailboxProps(), props, clock, MessageLifecycleGate()),
-        historySweep(), EventCleanupJob(StubMsgEvents(), props), clock, activity, props, OperationDayProps(),
+        historySweep(), EventCleanupJob(StubMsgEvents(), props),
+        ProcStateCleanupJob(proc, MessageLifecycleGate(), props),
+        clock, activity, props, OperationDayProps(),
    )

    @Test
@@ -0,0 +1,86 @@
+package com.gzzn.omms.msgexchange.jobs
+
+import com.gzzn.omms.msgexchange.config.PipelineProps
+import com.gzzn.omms.msgexchange.domain.ProcStatus
+import com.gzzn.omms.msgexchange.infra.stub.StubProcState
+import com.gzzn.omms.msgexchange.processing.MessageLifecycleGate
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertNotNull
+import org.junit.jupiter.api.Assertions.assertNull
+import org.junit.jupiter.api.Test
+import java.time.Duration
+import java.time.Instant
+
+/**
+ * US-11 处理记录清理的几条规矩：
+ * - 只删「终态 + 已回填 + 超过保留期」三关都过的行；
+ * - 未完成（PENDING/FAILED）或没回填的行一条不删；
+ * - 清理与人工重放共用同一把生命周期门，不持 PIPELINE_LOCK。
+ */
+class ProcStateCleanupJobTest {
+
+    private val now: Instant = Instant.parse("2026-09-21T04:00:00Z")
+    private val retention: Duration = Duration.ofDays(90)
+
+    @Test
+    fun `expired terminal and backfilled rows are deleted`() {
+        val proc = StubProcState()
+        val id = seed(proc, 1L, ProcStatus.SUCCEEDED, backfilled = true, updatedAt = now.minus(retention).minusSeconds(1))
+        val job = ProcStateCleanupJob(proc, MessageLifecycleGate(), PipelineProps().apply { pipeline.terminalRetention = retention })
+
+        val outcome = job.run(now)
+
+        assertEquals(1, outcome.deleted)
+        assertNull(proc.find(id))
+    }
+
+    @Test
+    fun `terminal rows without backfill are never deleted`() {
+        val proc = StubProcState()
+        val id = seed(proc, 2L, ProcStatus.DEAD, backfilled = false, updatedAt = now.minus(retention).minusSeconds(1))
+        val job = ProcStateCleanupJob(proc, MessageLifecycleGate(), PipelineProps().apply { pipeline.terminalRetention = retention })
+
+        val outcome = job.run(now)
+
+        assertEquals(0, outcome.deleted)
+        assertNotNull(proc.find(id))
+    }
+
+    @Test
+    fun `unfinished rows are never deleted even when expired`() {
+        val proc = StubProcState()
+        val failed = seed(proc, 3L, ProcStatus.FAILED, backfilled = true, updatedAt = now.minus(retention).minusSeconds(1))
+        val pending = seed(proc, 4L, ProcStatus.PENDING, backfilled = true, updatedAt = now.minus(retention).minusSeconds(1))
+        val job = ProcStateCleanupJob(proc, MessageLifecycleGate(), PipelineProps().apply { pipeline.terminalRetention = retention })
+
+        val outcome = job.run(now)
+
+        assertEquals(0, outcome.deleted)
+        assertNotNull(proc.find(failed))
+        assertNotNull(proc.find(pending))
+    }
+
+    @Test
+    fun `rows within retention stay`() {
+        val proc = StubProcState()
+        val id = seed(proc, 5L, ProcStatus.SKIPPED, backfilled = true, updatedAt = now.minus(retention).plusSeconds(1))
+        val job = ProcStateCleanupJob(proc, MessageLifecycleGate(), PipelineProps().apply { pipeline.terminalRetention = retention })
+
+        val outcome = job.run(now)
+
+        assertEquals(0, outcome.deleted)
+        assertNotNull(proc.find(id))
+    }
+
+    /** 手工插一行并按指定 UPDATED_AT 写终态，模拟「很久以前就了结并回填」的记录。 */
+    private fun seed(proc: StubProcState, msgId: Long, state: ProcStatus, backfilled: Boolean, updatedAt: Instant): Long {
+        proc.insertIfAbsent(msgId, Instant.EPOCH, Instant.EPOCH)
+        if (backfilled) {
+            proc.markTerminal(msgId, state, errorClass = null, lastError = null, attempts = 1, now = Instant.EPOCH)
+            proc.markBackfilled(msgId, now = updatedAt)
+        }
+        // 直接改 UPDATED_AT 到目标时刻（stub 只在写路径刷新它，这里用回填时刻代表终态时刻）
+        proc.rows[msgId] = proc.rows.getValue(msgId).copy(updatedAt = updatedAt)
+        return msgId
+    }
+}