fix(jobs): recheck archived flight version before purge

HistorySweepJob 归档成功后进 PIPELINE_LOCK 事务，按 (FLID, STATE_VERSION) 用同一判据复查，仍合格才登记 tombstone 并删除；purgeArchived 改收候选而非裸 FLID 列表，并带 (flid, state_version) 删除守卫。事件与删除同事务提交或回滚（INV-17、INV-18）。

验证：./gradlew test 131 tests / 0 fail（新增「归档后主泵更新同一 FLID → 不删、不发 tombstone」用例）。tombstone 与删除的回滚原子性依赖真实 PG 事务，本机无 Docker 时无法执行。

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<HistoryCandidate>
 
     /**
-     * 物理删除主行与明细。**只能在历史存储确认归档成功之后调用**；
+     * 物理删除主行与明细，按 `(FLID, STATE_VERSION)` 守卫：只删归档确认过的那个版本，
-     * 历史存储没接通时，调用方必须传空集合，也就是一条都不删。
+     * 期间被主泵写过的行影响 0 行。
+     *
+     * **只能在历史存储确认归档成功、且在 `PIPELINE_LOCK` 事务内按同一判据复查之后调用**（`INV-18`）；
+     * 历史存储没接通时，调用方必须传空列表，也就是一条都不删。
      */
-    fun purgeArchived(flids: Collection<String>): Int
+    fun purgeArchived(candidates: List<HistoryCandidate>): Int
 }
 
 /** 日计划处理留痕：只追加、不参与业务判断，一次处理（含重放）记一行，写失败不影响业务。 */

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

@@ -706,21 +706,23 @@ class JdbcFlightStateRepository(
         }
     }
 
-    override fun purgeArchived(flids: Collection<String>): Int {
+    override fun purgeArchived(candidates: List<HistoryCandidate>): Int {
-        if (flids.isEmpty()) return 0
+        if (candidates.isEmpty()) return 0
         var purged = 0
-        flids.chunked(500).forEach { chunk ->
+        candidates.forEach { candidate ->
-            val placeholders = chunk.joinToString(",") { "?" }
+            // 按 (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)",
+                    "DELETE FROM $table WHERE flid = ?",
-                    { ps -> chunk.forEachIndexed { i, id -> ps.setString(i + 1, id) } },
+                    { 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
     }

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

@@ -366,11 +366,15 @@ class StubFlightState : FlightStateRepository {
             }
         }
 
-    override fun purgeArchived(flids: Collection<String>): Int {
+    override fun purgeArchived(candidates: List<HistoryCandidate>): Int {
         var n = 0
-        flids.forEach { flid ->
+        candidates.forEach { candidate ->
-            if (mains.remove(flid) != null) n++
+            val main = mains[candidate.flid] ?: return@forEach
-            snapshots.remove(flid)
+            // 只删归档确认过的那个版本：期间被主泵写过的行留着。
+            if (main.stateVersion != candidate.stateVersion) return@forEach
+            mains.remove(candidate.flid)
+            snapshots.remove(candidate.flid)
+            n++
         }
         return n
     }

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、被这里直接清掉的航班，删除前补发一次通知，否则下游一直以为它还在
+        // 归档与删除之间主泵可能已写入同一 FLID：进锁事务后按 (FLID, STATE_VERSION) 复查，
-        val preDelete = toPurge.filter { it.wasNeverFdel }
+        // 仍合格才登记 tombstone 并删除；事件与删除同事务提交或回滚（INV-17、INV-18）。
-        if (preDelete.isNotEmpty()) {
+        var purged = 0
-            msgEvents.insertAll(preDelete.map { tombstone(it) })
+        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)
     }
 

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<HistoryCandidate>): Set<String> {
+                // 模拟归档期间主泵对同一 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)