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 b44aa44..d97d167 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt +++ b/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() { 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 0501937..2398e34 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 @@ -160,8 +160,6 @@ data class Backlog( interface MsgEventRepository { fun insertAll(events: List): List - fun headUnsent(target: String): MsgEvent? - fun claimBatch(target: String, limit: Int): List /** 挑出待发的整态事件,同一条航班只取版本最高的那一条(中间的版本不用发)。 */ @@ -227,9 +225,6 @@ interface FlightStateRepository { * 历史存储没接通时,调用方必须传空集合,也就是一条都不删。 */ fun purgeArchived(flids: Collection): Int - - /** 运营日还没定下来的航班有多少条(观测用;这些航班暂时不参与清理)。 */ - fun countOperationDayNull(): Int } /** 日计划处理留痕:只追加、不参与业务判断,一次处理(含重放)记一行,写失败不影响业务。 */ diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/SnapshotLogPurge.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/SnapshotLogPurge.kt new file mode 100644 index 0000000..1f432bb --- /dev/null +++ b/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 +} 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 3a060f2..090c85e 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 @@ -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 = 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 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/retry/ProcFailure.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/retry/ProcFailure.kt index e8f6915..749ac86 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/retry/ProcFailure.kt +++ b/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 } } 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 c59f9ce..7f4e9c5 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 @@ -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 = 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() 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 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxService.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxService.kt index 3035157..1ec9457 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxService.kt +++ b/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( diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJob.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJob.kt index f07e040..6465237 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJob.kt +++ b/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): Set } - 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) } 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 599e1c5..5f57ec2 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt +++ b/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) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/BackfillService.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/BackfillService.kt index 5c07bb5..2da61a0 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/BackfillService.kt +++ b/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 分钟的 * 退避往后推;如果一条消息从收到现在已经超过超期期限,则无视退避强制补写—— * 否则退避可能一直失败下去,这些行永远打不上标记,库方就没法清理信箱。 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt index 154800e..f7ca06f 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt +++ b/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 } } diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 9577d05..41f142e 100644 --- a/src/main/resources/application.yml +++ b/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=跳过可见存量 | diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt index 347fd84..0ae5231 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt @@ -219,7 +219,7 @@ class PipelineSmokeTest { } /** - * FIFO 修复(G2)端到端验收:兼容入口写入的高 ID **不会被提前领取**, + * FIFO 不变量"只领取已发现的行"端到端验收:兼容入口写入的高 ID **不会被提前领取**, * 必须等水位追平(较小 ID 补齐并入队)之后才按顺序处理。 */ @Test diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJobTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJobTest.kt index fbb18b7..229dde7 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJobTest.kt +++ b/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) + } }