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 777436f..2da364c 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 @@ -234,10 +234,13 @@ interface FlightStateRepository { fun findHistoryCandidates(rules: HistoryRules, zone: ZoneId, now: Instant): List /** - * 物理删除主行与明细。**只能在历史存储确认归档成功之后调用**; - * 历史存储没接通时,调用方必须传空集合,也就是一条都不删。 + * 物理删除主行与明细,按 `(FLID, STATE_VERSION)` 守卫:只删归档确认过的那个版本, + * 期间被主泵写过的行影响 0 行。 + * + * **只能在历史存储确认归档成功、且在 `PIPELINE_LOCK` 事务内按同一判据复查之后调用**(`INV-18`); + * 历史存储没接通时,调用方必须传空列表,也就是一条都不删。 */ - fun purgeArchived(flids: Collection): Int + fun purgeArchived(candidates: List): 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 d69b3fc..aa92773 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 @@ -706,21 +706,23 @@ class JdbcFlightStateRepository( } } - override fun purgeArchived(flids: Collection): Int { - if (flids.isEmpty()) return 0 + override fun purgeArchived(candidates: List): Int { + if (candidates.isEmpty()) return 0 var purged = 0 - flids.chunked(500).forEach { chunk -> - val placeholders = chunk.joinToString(",") { "?" } + candidates.forEach { candidate -> + // 按 (FLID, STATE_VERSION) 守卫:归档确认之后被主泵写过的行不删。 + val deleted = ds.update( + "DELETE FROM flight_schd WHERE flid = ? AND state_version = ?", + { ps -> ps.setString(1, candidate.flid); ps.setLong(2, candidate.stateVersion) }, + ) + if (deleted == 0) return@forEach + purged += deleted DETAIL_TABLES.forEach { table -> purged += ds.update( - "DELETE FROM $table WHERE flid IN ($placeholders)", - { ps -> chunk.forEachIndexed { i, id -> ps.setString(i + 1, id) } }, + "DELETE FROM $table WHERE flid = ?", + { ps -> ps.setString(1, candidate.flid) }, ) } - purged += ds.update( - "DELETE FROM flight_schd WHERE flid IN ($placeholders)", - { ps -> chunk.forEachIndexed { i, id -> ps.setString(i + 1, id) } }, - ) } return purged } 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 276dfa7..c0e713c 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 @@ -366,11 +366,15 @@ class StubFlightState : FlightStateRepository { } } - override fun purgeArchived(flids: Collection): Int { + override fun purgeArchived(candidates: List): Int { var n = 0 - flids.forEach { flid -> - if (mains.remove(flid) != null) n++ - snapshots.remove(flid) + candidates.forEach { candidate -> + val main = mains[candidate.flid] ?: return@forEach + // 只删归档确认过的那个版本:期间被主泵写过的行留着。 + if (main.stateVersion != candidate.stateVersion) return@forEach + mains.remove(candidate.flid) + snapshots.remove(candidate.flid) + n++ } return n } 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 9e19310..4e5dc3c 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJob.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJob.kt @@ -9,6 +9,8 @@ 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.PipelineLockRepository +import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager import com.gzzn.omms.msgexchange.infra.persistence.SnapshotLogPurge import jakarta.inject.Singleton import java.time.Duration @@ -32,6 +34,8 @@ class HistorySweepJob( private val msgEvents: MsgEventRepository, private val props: HistoryProps, private val operationDayProps: OperationDayProps, + private val txManager: PipelineTransactionManager, + private val lock: PipelineLockRepository, /** 历史存储端口:返回确认归档成功的 FLID 集合;部署侧没接通时是 null。 */ private val historyStore: HistoryStore? = null, /** 留痕清理端口:删掉超过保留期(默认 90 天)的日计划留痕行;不依赖历史存储开关,没接通时为 null。 */ @@ -63,12 +67,22 @@ class HistorySweepJob( if (archivedFlids.isEmpty()) return SweepOutcome(candidates.size, archived = 0, purged = 0, snapLogPurged = snapLogPurged) val toPurge = candidates.filter { it.flid in archivedFlids } - // 从没收到过 FDEL、被这里直接清掉的航班,删除前补发一次通知,否则下游一直以为它还在 - val preDelete = toPurge.filter { it.wasNeverFdel } - if (preDelete.isNotEmpty()) { - msgEvents.insertAll(preDelete.map { tombstone(it) }) + // 归档与删除之间主泵可能已写入同一 FLID:进锁事务后按 (FLID, STATE_VERSION) 复查, + // 仍合格才登记 tombstone 并删除;事件与删除同事务提交或回滚(INV-17、INV-18)。 + var purged = 0 + txManager.inTransaction { + lock.lock() + val current = flightState.findHistoryCandidates(rules, zone, now).associateBy { it.flid } + val rechecked = toPurge.filter { candidate -> current[candidate.flid]?.stateVersion == candidate.stateVersion } + if (rechecked.isNotEmpty()) { + // 从没收到过 FDEL、被这里直接清掉的航班,删除前补发一次通知,否则下游一直以为它还在 + val preDelete = rechecked.filter { it.wasNeverFdel } + if (preDelete.isNotEmpty()) { + msgEvents.insertAll(preDelete.map { tombstone(it) }) + } + purged = flightState.purgeArchived(rechecked) + } } - val purged = flightState.purgeArchived(archivedFlids) return SweepOutcome(candidates.size, archivedFlids.size, purged, snapLogPurged) } 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 0bcaabd..53dd4ea 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJobTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJobTest.kt @@ -7,6 +7,8 @@ import com.gzzn.omms.msgexchange.domain.flight.FlightState import com.gzzn.omms.msgexchange.domain.flight.HistoryCandidate import com.gzzn.omms.msgexchange.infra.stub.StubFlightState import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents +import com.gzzn.omms.msgexchange.infra.stub.StubPipelineLock +import com.gzzn.omms.msgexchange.infra.stub.StubPipelineTx import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.Test @@ -51,7 +53,7 @@ class HistorySweepJobTest { fun `history store not connected must delete zero rows`() { val f = seededFlight("F1", deleted = true, idleDays = 30) val events = StubMsgEvents() - val job = HistorySweepJob(f, events, HistoryProps().apply { historyStoreEnabled = false }, OperationDayProps()) + val job = HistorySweepJob(f, events, HistoryProps().apply { historyStoreEnabled = false }, OperationDayProps(), StubPipelineTx(), StubPipelineLock()) val outcome = job.run(now) @@ -72,7 +74,7 @@ class HistorySweepJobTest { val events = StubMsgEvents() val store = RecordingHistoryStore().apply { confirmed.add("F1") } // 只有 F1 归档确认成功 val job = HistorySweepJob( - f, events, HistoryProps().apply { historyStoreEnabled = true }, OperationDayProps(), historyStore = store, + f, events, HistoryProps().apply { historyStoreEnabled = true }, OperationDayProps(), StubPipelineTx(), StubPipelineLock(), historyStore = store, ) val outcome = job.run(now) @@ -91,7 +93,7 @@ class HistorySweepJobTest { val events = StubMsgEvents() val store = RecordingHistoryStore() val job = HistorySweepJob( - f, events, HistoryProps().apply { historyStoreEnabled = true }, OperationDayProps(), historyStore = store, + f, events, HistoryProps().apply { historyStoreEnabled = true }, OperationDayProps(), StubPipelineTx(), StubPipelineLock(), historyStore = store, ) job.run(now) @@ -103,13 +105,37 @@ class HistorySweepJobTest { assertEquals(null, f.findMainRow("F3")) } + @Test + fun `a flight updated after archival is not purged and emits no tombstone`() { + val f = seededFlight("F4", deleted = false, idleDays = 30) + val events = StubMsgEvents() + val store = object : HistorySweepJob.HistoryStore { + override fun archive(candidates: List): Set { + // 模拟归档期间主泵对同一 FLID 提交了新版本:清理判据已不再成立/版本已变 + f.markDeleted("F4", msgId = 9, now = now) + return setOf("F4") + } + } + val job = HistorySweepJob( + f, events, HistoryProps().apply { historyStoreEnabled = true }, OperationDayProps(), + StubPipelineTx(), StubPipelineLock(), historyStore = store, + ) + + val outcome = job.run(now) + + assertEquals(1, outcome.archived) + assertEquals(0, outcome.purged) // 归档后版本变了:本次不删 + assertTrue(f.findMainRow("F4") != null) // 在用航班不许被清掉 + assertEquals(0, events.rows.size) // 也不登记 tombstone + } + @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 }, OperationDayProps(), snapLogPurge = purge, + f, StubMsgEvents(), HistoryProps().apply { historyStoreEnabled = false }, OperationDayProps(), StubPipelineTx(), StubPipelineLock(), snapLogPurge = purge, ) val outcome = job.run(now)