From 1d95926e064d2038f42471219eb435250192751c Mon Sep 17 00:00:00 2001 From: windyboy Date: Mon, 21 Sep 2026 12:52:26 +0800 Subject: [PATCH] =?UTF-8?q?feat(jobs):=20PROC=5FSTATE=20=E7=BB=88=E6=80=81?= =?UTF-8?q?=E8=A1=8C=E5=88=B0=E6=9C=9F=E6=B8=85=E7=90=86=EF=BC=88ACM2-82?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 deleteExpiredTerminal：仅删终态 + BACKFILL_AT 非空 + UPDATED_AT 超期的行（US-11） - ProcStateCleanupJob 每轮 tick 执行，与人工重放共用 MessageLifecycleGate，不持 PIPELINE_LOCK - 登记参数 msgx.pipeline.terminal-retention（默认 90d，暂定）进 reference.md - Stub 与 JDBC 实现同口径；补四类候选条件回归 --- docs/reference.md | 1 + .../omms/msgexchange/config/PipelineProps.kt | 7 ++ .../infra/persistence/Repositories.kt | 7 ++ .../persistence/jdbc/JdbcPgRepositories.kt | 18 ++++ .../infra/stub/StubRepositories.kt | 17 ++++ .../gzzn/omms/msgexchange/jobs/JobRunner.kt | 5 ++ .../msgexchange/jobs/ProcStateCleanupJob.kt | 35 ++++++++ .../infra/retry/ReplayServiceTest.kt | 2 + .../omms/msgexchange/jobs/JobRunnerTest.kt | 4 +- .../jobs/ProcStateCleanupJobTest.kt | 86 +++++++++++++++++++ 10 files changed, 181 insertions(+), 1 deletion(-) create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/jobs/ProcStateCleanupJob.kt create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/jobs/ProcStateCleanupJobTest.kt diff --git a/docs/reference.md b/docs/reference.md index 9d19ea5..a7b901f 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -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`） | 暂定 | ### 写回信箱与编号扫描 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt index 4c522ac..4656e4e 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt @@ -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) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt index d504fd3..e5d3990 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt @@ -126,6 +126,13 @@ interface ProcStateRepository { /** 人工重放：把指定错误类别的 FAILED / DEAD 记录改回 PENDING，重试次数清零。 */ fun requeueByErrorClasses(errorClasses: List): Int + /** + * 到期清理（US-11）：删除终态（SUCCEEDED/SKIPPED/DEAD）、已回填（BACKFILL_AT 非空） + * 且 `UPDATED_AT < cutoff` 的 PROC_STATE 行；未完成或未回填的行一律不删。 + * @return 实际删除的行数 + */ + fun deleteExpiredTerminal(cutoff: Instant, limit: Int): Int + /** 积压观测：还没处理完的条数、最老一条的接收时间、处理完但还没回填的条数。 */ fun backlog(): Backlog diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt index 8359b61..02262f7 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt @@ -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 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt index 8c52265..6d0d8a1 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt @@ -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 = + setOf(ProcStatus.SUCCEEDED, ProcStatus.SKIPPED, ProcStatus.DEAD) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt index 3135be9..181f4ca 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt @@ -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 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/ProcStateCleanupJob.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/ProcStateCleanupJob.kt new file mode 100644 index 0000000..8e1351e --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/ProcStateCleanupJob.kt @@ -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 + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt index 3f576fa..d5a3e08 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt @@ -77,6 +77,8 @@ class ReplayServiceTest { } return n } + + override fun deleteExpiredTerminal(cutoff: java.time.Instant, limit: Int): Int = 0 } @Test diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt index 658ebd2..57dabca 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt @@ -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 diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/ProcStateCleanupJobTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/ProcStateCleanupJobTest.kt new file mode 100644 index 0000000..d980506 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/ProcStateCleanupJobTest.kt @@ -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 + } +}