test(migration): 带数据 V1.1→最新升级链验证 + DELY Apply 拒绝 (ACM2-29 P2-4)

- FlywayMigrationTest 新增升级链用例：隔离 schema 迁至 V1.1.0、写入含 *_TXT 的存量行、一路升级到最新——断言标量数据存活、v2 追踪列默认就位、 display 视图/backfill_todo/明细表可用、*_TXT 列按已知损失边界消失（评审 F6：升级前必须导出或重放，测试固化该边界） - L2 收尾：DELY 无协议序号属性（DLNO 非法）——SEQ_ATTR 移除 DELY，Apply 对 DELY fail fast 拒绝（语义矩阵 §2 同步口径） - 级联回归已在 P2-2 落地（deleteByFlids/deleteDiffByDay 明细级联断言）验证：MSGX_PG_PORT=5433 真实 PG 全量 107 用例 0 失败 0 跳过
2026-09-08 15:54:50 +08:00
parent 20aafb9ab0
commit 06a1081bdd
3 changed files with 116 additions and 2 deletions
@@ -31,12 +31,12 @@ object FlightStateEngine {
        "CLDT" to "CLNO",
        "PSDT" to "PSNO",
        "CHDT" to "CHNO",
-        "DELY" to "DLNO",
        "ABTM" to "ASNO",
        "CHOT" to "CSNO",
        "ROUT" to "RTNO",
        "ERUT" to "RTNO",
    )
+    // 注意：DELY 无协议序号属性（DLNO 非法）→ 不支持 Apply；清除走空数组 Replace(空集)

    /** DNLD/FLOP 字段集 → 命令（出现即 Set/Replace；未出现即 Unchanged；序号 0 条目 = 显式清除标记）。 */
    fun commandsFromFields(flid: String, fields: FlightFields, snapshotReplace: Boolean): FlightFieldCommands {
@@ -96,8 +96,10 @@ object FlightStateEngine {
                is CollectionCommand.Replace -> baseCollections[key] = cmd.items.toMutableList()
                CollectionCommand.Clear -> baseCollections.remove(key)
                is CollectionCommand.Apply -> {
+                    val seqAttr = SEQ_ATTR[key]
+                        ?: throw IllegalArgumentException("$key does not support Apply: protocol defines no source sequence attribute")
                    val list = baseCollections.getOrPut(key) { mutableListOf() }.toMutableList()
-                    val idx = list.indexOfFirst { it[SEQ_ATTR[key] ?: ""] == cmd.sourceSeq }
+                    val idx = list.indexOfFirst { it[seqAttr] == cmd.sourceSeq }
                    if (idx >= 0) {
                        list[idx] = cmd.item
                    } else {
@@ -143,6 +143,29 @@ class FlightStateEngineTest {
        assertEquals(emptyList<Map<String, String>>(), next.collections["GTDT"])
    }

+    @Test
+    fun `apply on DELY fails fast because protocol defines no sequence attribute`() {
+        val current = FlightNextState(
+            "F1",
+            emptyMap(),
+            mapOf("DELY" to listOf(mapOf("CODE" to "YY"))),
+            1L,
+            "m1",
+        )
+        val error = runCatching {
+            FlightStateEngine.apply(
+                current,
+                FlightFieldCommands(
+                    "F1",
+                    collections = mapOf("DELY" to CollectionCommand.Apply(mapOf("CODE" to "ZZ"), "1")),
+                ),
+                "m2",
+                bumpVersion = true,
+            )
+        }.exceptionOrNull()
+        assertTrue(error is IllegalArgumentException)
+    }
+
    @Test
    fun `exception null payload translates to explicit clear and apply records cleared key`() {
        val current = FlightNextState(
@@ -3,6 +3,8 @@ package com.gzzn.omms.msgexchange.infra.persistence.jdbc
 import com.gzzn.omms.msgexchange.support.PgTestSupport
 import org.flywaydb.core.Flyway
 import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertNull
 import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.Assumptions.assumeTrue
 import org.junit.jupiter.api.Test
@@ -66,4 +68,91 @@ class FlywayMigrationTest {
            }
        }
    }
+
+    /**
+     * P2-4：带数据的 V1.1.0 → 最新版本升级链验证（隔离 schema，不污染 public）。
+     * V1.2.0 删除 *_TXT 集合列是「已知损失边界」（评审 F6）：标量列与结构完整存活，
+     * 集合文本不可恢复——升级前必须完整导出或重放（docs/flight-state-design-v2.md §7）。
+     */
+    @Test
+    fun `upgrade chain from seeded V1_1_0 data keeps scalars and adds v2 tracking`() {
+        assumeTrue(PgTestSupport.canConnect(), PgTestSupport.skipMessage())
+        val schema = "upgrade_chain_test"
+        // 清理历史遗留的隔离 schema
+        DriverManager.getConnection(PgTestSupport.jdbcUrl, PgTestSupport.user, PgTestSupport.password).use { conn ->
+            conn.createStatement().use { it.execute("DROP SCHEMA IF EXISTS $schema CASCADE") }
+        }
+
+        // 1. 迁移到 V1.1.0（含 *_TXT 集合文本列形态）并写入带数据行
+        Flyway.configure()
+            .dataSource(PgTestSupport.jdbcUrl, PgTestSupport.user, PgTestSupport.password)
+            .locations("classpath:db/migration")
+            .schemas(schema)
+            .defaultSchema(schema)
+            .target("1.1.0")
+            .load()
+            .migrate()
+        DriverManager.getConnection(PgTestSupport.jdbcUrl, PgTestSupport.user, PgTestSupport.password).use { conn ->
+            conn.createStatement().use {
+                it.execute(
+                    """
+                    INSERT INTO $schema.flight_schd (flid, fday, flno, remc, gtdt_txt, created_at, updated_at)
+                    VALUES ('UPG_FL_1', '2026-09-07', 'CA_UPG', 'legacy-note', '[{"GTNO":"1","GATE":"G1"}]', now(), now())
+                    """.trimIndent(),
+                )
+            }
+        }
+
+        // 2. 一路升级到最新版本
+        val result = Flyway.configure()
+            .dataSource(PgTestSupport.jdbcUrl, PgTestSupport.user, PgTestSupport.password)
+            .locations("classpath:db/migration")
+            .schemas(schema)
+            .defaultSchema(schema)
+            .load()
+            .migrate()
+        assertTrue(result.success)
+
+        DriverManager.getConnection(PgTestSupport.jdbcUrl, PgTestSupport.user, PgTestSupport.password).use { conn ->
+            conn.createStatement().use { stmt ->
+                // 3. 标量数据存活；v2 追踪列就位（存量行 state_version 默认 0）
+                stmt.executeQuery(
+                    "SELECT flno, remc, state_version, last_message_id FROM $schema.flight_schd WHERE flid = 'UPG_FL_1'",
+                ).use { rs ->
+                    assertTrue(rs.next())
+                    assertEquals("CA_UPG", rs.getString("flno"))
+                    assertEquals("legacy-note", rs.getString("remc"))
+                    assertEquals(0L, rs.getLong("state_version"))
+                    assertNull(rs.getString("last_message_id"))
+                }
+
+                // 4. 已知损失边界：*_TXT 集合列已不存在（升级前须导出/重放，评审 F6）
+                stmt.executeQuery(
+                    "SELECT column_name FROM information_schema.columns WHERE table_schema = '$schema' AND table_name = 'flight_schd' AND column_name LIKE '%\\_TXT'",
+                ).use { rs ->
+                    assertFalse(rs.next(), "legacy *_TXT columns must be gone after V1.2.0 (documented lossy boundary)")
+                }
+
+                // 5. v2 结构就位：display 视图可查询（存量行 gate_total=0）、backfill_todo 与明细表存在
+                stmt.executeQuery(
+                    "SELECT flno, gate_total FROM $schema.flight_schd_display WHERE flid = 'UPG_FL_1'",
+                ).use { rs ->
+                    assertTrue(rs.next())
+                    assertEquals("CA_UPG", rs.getString("flno"))
+                    assertEquals(0, rs.getInt("gate_total"))
+                }
+                stmt.executeQuery("SELECT COUNT(*) FROM $schema.backfill_todo").use { rs ->
+                    assertTrue(rs.next())
+                }
+                stmt.executeQuery("SELECT COUNT(*) FROM $schema.flight_gate WHERE flid = 'UPG_FL_1'").use { rs ->
+                    assertTrue(rs.next())
+                    assertEquals(0, rs.getInt(1))
+                }
+            }
+        }
+
+        DriverManager.getConnection(PgTestSupport.jdbcUrl, PgTestSupport.user, PgTestSupport.password).use { conn ->
+            conn.createStatement().use { it.execute("DROP SCHEMA IF EXISTS $schema CASCADE") }
+        }
+    }
 }