test(jobs): verify purge rollback on real postgres

新增 HistorySweepPurgePgTest（Testcontainers 本地 PG）：归档确认后删除阶段注入失败，验证同事务登记的 tombstone 随事务回滚、当前态保留（INV-17/INV-18）。
2026-09-12 20:45:41 +08:00
parent 15c035182f
commit ab18e4ec3f
1 changed files with 85 additions and 0 deletions
@@ -0,0 +1,85 @@
+package com.gzzn.omms.msgexchange.jobs
+
+import com.gzzn.omms.msgexchange.config.HistoryProps
+import com.gzzn.omms.msgexchange.config.OperationDayProps
+import com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot
+import com.gzzn.omms.msgexchange.domain.flight.FlightState
+import com.gzzn.omms.msgexchange.domain.flight.HistoryCandidate
+import com.gzzn.omms.msgexchange.infra.persistence.FlightStateRepository
+import com.gzzn.omms.msgexchange.infra.persistence.jdbc.JdbcFlightStateRepository
+import com.gzzn.omms.msgexchange.infra.persistence.jdbc.JdbcMsgEventRepository
+import com.gzzn.omms.msgexchange.infra.persistence.jdbc.JdbcPipelineLockRepository
+import com.gzzn.omms.msgexchange.infra.persistence.jdbc.JdbcPipelineTransactionManager
+import com.gzzn.omms.msgexchange.support.PgTestSupport
+import com.zaxxer.hikari.HikariConfig
+import com.zaxxer.hikari.HikariDataSource
+import org.flywaydb.core.Flyway
+import org.junit.jupiter.api.Assertions.assertNotNull
+import org.junit.jupiter.api.Assertions.assertThrows
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Assumptions.assumeTrue
+import org.junit.jupiter.api.Test
+import java.time.Clock
+import java.time.Instant
+import java.time.LocalDate
+import java.time.ZoneOffset
+import java.util.UUID
+
+/**
+ * 在真实 PostgreSQL 上验证历史清退的原子性：归档确认后若在删当前态之前失败，
+ * 同事务登记的 tombstone 必须一起回滚（`INV-17`、`INV-18`）。
+ */
+class HistorySweepPurgePgTest {
+
+    private val t0: Instant = Instant.parse("2026-09-12T00:00:00Z")
+
+    @Test
+    fun `a failure after the tombstone insert rolls back the whole purge transaction`() {
+        val ds = dataSource()
+        val clock = Clock.fixed(t0, ZoneOffset.UTC)
+        val flights = JdbcFlightStateRepository(ds, clock)
+        val events = JdbcMsgEventRepository(ds, clock)
+        val flid = "PG-" + UUID.randomUUID().toString().take(8)
+        flights.persistFullState(
+            FlightSnapshot(flid, LocalDate.of(2026, 8, 1), FlightState.ACTIVE, 1, emptyMap(), emptyMap()),
+            msgId = 1,
+            now = t0.minusSeconds(30L * 86400),   // 静默超过 idle-hours，命中生命周期清退
+        )
+
+        // 删除阶段注入失败：tombstone 已登记，删除抛错 → 整个事务必须回滚。
+        val failing = object : FlightStateRepository by flights {
+            override fun purgeArchived(candidates: List<HistoryCandidate>): Int =
+                throw IllegalStateException("purge-failed")
+        }
+        val job = HistorySweepJob(
+            failing, events, HistoryProps().apply { historyStoreEnabled = true },
+            OperationDayProps(), JdbcPipelineTransactionManager(ds), JdbcPipelineLockRepository(ds),
+            historyStore = { setOf(flid) },
+        )
+
+        assertThrows(IllegalStateException::class.java) { job.run(t0) }
+
+        assertNotNull(flights.findMainRow(flid), "当前态不得被删")
+        assertTrue(
+            events.mergePendingSchd(t0, 1000).none { it.partitionKey == flid },
+            "tombstone 必须随事务回滚，不留半状态",
+        )
+    }
+
+    private fun dataSource(): javax.sql.DataSource {
+        assumeTrue(PgTestSupport.canConnect(), PgTestSupport.skipMessage())
+        Flyway.configure()
+            .dataSource(PgTestSupport.jdbcUrl, PgTestSupport.user, PgTestSupport.password)
+            .locations("classpath:db/migration")
+            .load()
+            .migrate()
+        return HikariDataSource(
+            HikariConfig().apply {
+                jdbcUrl = PgTestSupport.jdbcUrl
+                username = PgTestSupport.user
+                password = PgTestSupport.password
+                maximumPoolSize = 2
+            },
+        )
+    }
+}