test(integration): JDBC 集成测试 Testcontainers 兜底 + 全结构回环矩阵 (ACM2-29 P3-A)

- PgTestSupport 三级解析：MSGX_PG_* 显式环境变量 → Testcontainers 自动起
  postgres:17-alpine 隔离容器（JVM 单例复用，docker 不可用才跳过）→ assumeTrue
  绝不误报通过；解决了「本机端口被占/无库则集成测试静默跳过」的验收假绿问题
- loadCollections 只输出非空属性（缺失键语义），write→read 线格式一致
- 新增 readAfterWrite 矩阵用例：16 类结构（多行/重复资源号/非连续源序号/异常
  前缀列/文本旁列）persist → findNextStateByFlid 深度等值 + wire(nextState) ==
  wire(readAfterWrite) 双断言（评审 F5 持久化侧+线格式侧闭环）

验证：MSGX_PG_PORT=5433 外接库 108 用例全绿；无环境变量 Testcontainers
自动容器路径 18/18 全绿

build.gradle.kts

@@ -58,6 +58,8 @@ dependencies {
     testRuntimeOnly(libs.h2)        // U03：配置绑定回归测试的测试库（内存 datasource，冒烟用）
     testImplementation("io.micronaut.test:micronaut-test-junit5")
     testImplementation("org.jetbrains.kotlin:kotlin-test")
+    // P3-A（ACM2-29）：无外部 PG 时 JDBC 集成测试自动起隔离容器；有 MSGX_PG_* 环境变量时不启用
+    testImplementation("org.testcontainers:postgresql:1.20.4")
 }
 
 application {

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

@@ -136,9 +136,11 @@ internal object FlightDetailTables {
                 "SELECT * FROM ${spec.table} WHERE flid = ? ORDER BY ordinal ASC",
                 { ps -> ps.setString(1, flid) },
             ) { rs ->
-                spec.columns.associate { (attr, col) -> attr to rs.getString(col) }.toMutableMap().apply {
+                // 只输出非空属性（缺失键语义），保证 write→read 线格式一致（评审 F5）
-                    put(spec.seqAttr, rs.getString("source_seq"))
+                spec.columns.associate { (attr, col) -> attr to rs.getString(col) }
-                }
+                    .filterValues { it != null }
+                    .toMutableMap()
+                    .apply { rs.getString("source_seq")?.let { put(spec.seqAttr, it) } }
             }
             if (rows.isNotEmpty()) {
                 out[spec.collectionKey] = mapper.writeValueAsString(rows)
@@ -149,12 +151,12 @@ internal object FlightDetailTables {
                 "SELECT source_seq, apcd, scat, scdt FROM flight_route_point WHERE flid = ? AND route_kind = ? ORDER BY ordinal ASC",
                 { ps -> ps.setString(1, flid); ps.setString(2, kind) },
             ) { rs ->
-                mapOf(
+                buildMap {
-                    "RTNO" to rs.getString("source_seq"),
+                    rs.getString("source_seq")?.let { put("RTNO", it) }
-                    "APCD" to rs.getString("apcd"),
+                    rs.getString("apcd")?.let { put("APCD", it) }
-                    "SCAT" to rs.getString("scat"),
+                    rs.getString("scat")?.let { put("SCAT", it) }
-                    "SCDT" to rs.getString("scdt"),
+                    rs.getString("scdt")?.let { put("SCDT", it) }
-                )
+                }
             }
             if (rows.isNotEmpty()) {
                 out[key] = mapper.writeValueAsString(rows)

src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlightSchdJdbcPgTest.kt

@@ -738,6 +738,62 @@ class FlightSchdJdbcPgTest {
         assertNull(repo.findNextStateByFlid(flid)!!.scalars["FRET"])
     }
 
+    @Test
+    fun `PG dialect - readAfterWrite equals nextState across all structures and wire stays consistent`() {
+        val day = "2026-09-07"
+        val flid = "TEST_MATRIX"
+        val engine = com.gzzn.omms.msgexchange.domain.flight.FlightStateEngine
+        val fields = mapOf(
+            "FLNO" to "CA_MATRIX",
+            "SODT" to "07SEP260800",
+            "GTDT" to """[{"GTNO":"1","GATE":"G28","GTYP":"D"},{"GTNO":"3","GATE":"G33"},{"GTNO":"3","GATE":"G23"}]""",
+            "CKDT" to """[{"CKNO":"1","CHKC":"01","CCLS":"Y"},{"CKNO":"2","CHKC":"01","CCLS":"F"}]""",
+            "CLDT" to """[{"CLNO":"1","BELT":"B01"},{"CLNO":"2","BELT":"B02"}]""",
+            "PSDT" to """[{"PSNO":"1","PSST":"S209"}]""",
+            "CHDT" to """[{"CHNO":"1","CHUT":"C1"}]""",
+            "DELY" to """[{"CODE":"YY","STRT":"07SEP261605","DURA":"0200","REMC":"delayed"},{"CODE":"WX","STRT":"07SEP261800","DURA":"0030"}]""",
+            "ABTM" to """[{"ASNO":"1","ABDG":"B01","ABOP":"A","AOTM":"07SEP261700"},{"ASNO":"2","ABDG":"B01","ABOP":"D","AOTM":"07SEP261900"}]""",
+            "CHOT" to """[{"CSNO":"1","CHID":"ON","CHTM":"07SEP261650"},{"CSNO":"2","CHID":"OFF","CHTM":"07SEP261920"}]""",
+            "ROUT" to """[{"RTNO":"1","APCD":"CTU"},{"RTNO":"2","APCD":"PEK","SCDT":"07SEP261100"}]""",
+            "FDIV" to """{"DDES":"PEK","DDIR":"TO"}""",
+            "SRVT" to """[{"SRTC":"WHEEL"}]""",
+        )
+        val state = engine.apply(null, engine.commandsFromFields(flid, fields, snapshotReplace = true), "msg-matrix-1", bumpVersion = true)
+
+        repo.persistNextStates(day, listOf(state), snapshotReplace = true)
+
+        // 断言一：readAfterWrite == nextState（评审 F5 的持久化侧闭环）
+        val readBack = repo.findNextStateByFlid(flid)!!
+        assertEquals(state.stateVersion, readBack.stateVersion)
+        assertEquals(state.lastMessageId, readBack.lastMessageId)
+        assertEquals(state.scalars.keys, readBack.scalars.keys)
+        state.scalars.forEach { (k, v) -> assertJsonEquivalent(v, readBack.scalars[k], "scalar[$k]") }
+        assertEquals(state.collections.keys, readBack.collections.keys)
+        state.collections.forEach { (key, items) ->
+            val actual = readBack.collections[key]!!
+            assertEquals(items.size, actual.size, "collection[$key] row count")
+            items.forEachIndexed { i, item ->
+                item.forEach { (attr, v) -> assertEquals(v, actual[i][attr], "collection[$key][$i].$attr") }
+            }
+        }
+
+        // 断言二：wire(nextState) == wire(readAfterWrite)（出站载荷与库态同源，F5 的线格式侧闭环）
+        val wireBefore = mapper.readTree(com.gzzn.omms.msgexchange.infra.persistence.FlightFieldsJson.toJson(state.toFlightFields()))
+        val wireAfter = mapper.readTree(com.gzzn.omms.msgexchange.infra.persistence.FlightFieldsJson.toJson(repo.findByFlid(flid)!!))
+        assertEquals(wireBefore, wireAfter)
+    }
+
+    /** JSON 文本按解析树等价比较（键序无关）；非 JSON 退化为字符串比较。 */
+    private fun assertJsonEquivalent(expected: String?, actual: String?, path: String) {
+        if (expected == null || actual == null) {
+            assertEquals(expected, actual, path)
+            return
+        }
+        val e = runCatching { mapper.readTree(expected) }.getOrNull()
+        val a = runCatching { mapper.readTree(actual) }.getOrNull()
+        if (e != null && a != null) assertEquals(e, a, path) else assertEquals(expected, actual, path)
+    }
+
     private fun assertLegacyCollectionColumnsAllNull(flid: String) {
         val columnList = com.gzzn.omms.msgexchange.support.LEGACY_COLLECTION_STORAGE_COLUMNS.joinToString(", ")
         ds.queryOne(

src/test/kotlin/com/gzzn/omms/msgexchange/support/PgTestSupport.kt

@@ -1,7 +1,15 @@
 package com.gzzn.omms.msgexchange.support
 
+import org.testcontainers.containers.PostgreSQLContainer
+
 /**
- * JDBC 集成测试 PostgreSQL 连接参数（环境变量化，解决本机 5432 端口冲突）。
+ * JDBC 集成测试 PostgreSQL 连接解析（ACM2-29 P3-A）。
+ *
+ * 优先级：
+ * 1. 显式环境变量 `MSGX_PG_URL` / `MSGX_PG_HOST` / `MSGX_PG_PORT` / `MSGX_PG_NAME`（外接库/CI 固定库）；
+ * 2. Testcontainers 自动起 `postgres:17-alpine` 隔离容器（未配置环境变量且 docker 可用时；
+ *    JVM 单例，首个用例启动后全程复用，容器随 JVM 退出由 Ryuk 清理）；
+ * 3. 都不可用 → canConnect()=false，用例 assumeTrue 跳过（绝不误报通过）。
  *
  * | 变量 | 默认 |
  * |---|---|
@@ -10,14 +18,41 @@ package com.gzzn.omms.msgexchange.support
  * | MSGX_PG_PASSWORD | msgx_dev_pass |
  */
 object PgTestSupport {
-    private val host = System.getenv("MSGX_PG_HOST") ?: "127.0.0.1"
+    private val envUrl = System.getenv("MSGX_PG_URL")
-    private val port = System.getenv("MSGX_PG_PORT") ?: "5432"
+    private val envHost = System.getenv("MSGX_PG_HOST")
-    private val name = System.getenv("MSGX_PG_NAME") ?: "msgx"
+    private val envPort = System.getenv("MSGX_PG_PORT")
+    private val envName = System.getenv("MSGX_PG_NAME")
+    private val envConfigured = envUrl != null || envHost != null || envPort != null || envName != null
 
-    val jdbcUrl: String = System.getenv("MSGX_PG_URL")
+    private val container: PostgreSQLContainer<*>? by lazy {
-        ?: "jdbc:postgresql://$host:$port/$name"
+        if (envConfigured) {
-    val user: String = System.getenv("MSGX_PG_USER") ?: "msgx_dev"
+            null // 显式指定外接库时不拉容器
-    val password: String = System.getenv("MSGX_PG_PASSWORD") ?: "msgx_dev_pass"
+        } else {
+            try {
+                PostgreSQLContainer("postgres:17-alpine")
+                    .withDatabaseName("msgx")
+                    .withUsername("msgx_dev")
+                    .withPassword("msgx_dev_pass")
+                    .also { it.start() }
+            } catch (_: Exception) {
+                null // docker 不可用 → 回退本地默认地址，由 assumeTrue 决定跳过
+            }
+        }
+    }
+
+    val jdbcUrl: String
+        get() = envUrl
+            ?: if (envConfigured) {
+                "jdbc:postgresql://${envHost ?: "127.0.0.1"}:${envPort ?: "5432"}/${envName ?: "msgx"}"
+            } else {
+                container?.jdbcUrl ?: "jdbc:postgresql://127.0.0.1:5432/msgx"
+            }
+
+    val user: String
+        get() = System.getenv("MSGX_PG_USER") ?: container?.username ?: "msgx_dev"
+
+    val password: String
+        get() = System.getenv("MSGX_PG_PASSWORD") ?: container?.password ?: "msgx_dev_pass"
 
     fun canConnect(): Boolean = try {
         java.sql.DriverManager.getConnection(jdbcUrl, user, password).use { true }
@@ -26,5 +61,5 @@ object PgTestSupport {
     }
 
     fun skipMessage(): String =
-        "PostgreSQL not accessible at $jdbcUrl (set MSGX_PG_URL / MSGX_PG_PORT to avoid port conflicts)"
+        "PostgreSQL not accessible at $jdbcUrl (set MSGX_PG_URL / MSGX_PG_PORT, or enable docker for Testcontainers)"
 }