feat(processing): 快照流程锁内复核收敛 + 异常显式清除语义 + 语义矩阵定稿 (ACM2-29 P2-5/M3)

- SnapshotFlow 收敛到 v2 §5 事务顺序：锁外仅解析校验；取得 PIPELINE_LOCK 后锁内复核快照身份（last_message_id 重放短路）→ 锁内读代与当前航班态 → 计算 nextState → 写入 → CAS → 事件 → SUCCEEDED。锁内 CAS 失败属数据异常，一律回滚 FAILED(INFRA)，删除「版本号相同即视为重放」的推断路径（评审 F7/F8 收口） - M3/F4：异常对象（FDIV/FRET/FLAB）显式清除语义——null 字面量/空对象/空串翻译为 Clear 命令；FlightNextState.clearedKeys 携带本次清除键；增量路径按映射把前缀标量列置 NULL；库内清除后读视图无该键（线格式保持键缺失，显式清除表达留待阶段 2 按下游契约定） - StubFlightSchd 增量分支改为 nextState 权威整体替换（引擎已合并当前态），天然承载 Clear；flattenScalarsOnly 去除冗余循环（L2） - docs/flight-state-semantics.md：16 类结构语义矩阵定稿（P0-A 交付物落库） - 新测试：引擎 0 标记/异常清除/混排 fail-fast 单测；JDBC 异常清除回环用例验证：MSGX_PG_PORT=5433 真实 PG ./gradlew test --rerun-tasks 105 用例 0 失败 0 跳过
2026-09-08 15:44:48 +08:00
parent 48a27b26dd
commit 20aafb9ab0
8 changed files with 223 additions and 46 deletions
@@ -9,6 +9,8 @@ data class FlightNextState(
    val collections: Map<String, List<Map<String, String>>>,
    val stateVersion: Long,
    val lastMessageId: String,
+    /** 本次迁移中被显式清除的标量/异常/文本键（仓储据此写 NULL 列；快照全量替换下无意义）。 */
+    val clearedKeys: Set<String> = emptySet(),
 ) {
    fun toFlightFields(mapper: ObjectMapper = ObjectMapper()): FlightFields {
        val out = linkedMapOf<String, String>()
@@ -14,6 +14,17 @@ object FlightStateEngine {
        "GTDT", "CKDT", "CLDT", "PSDT", "CHDT", "DELY", "ABTM", "CHOT", "ROUT", "ERUT",
    )

+    /** 1:0..1 单值异常结构：主表前缀标量列承载；显式 null 载荷 = 清除（v2 §3.2/§9.1）。 */
+    val EXCEPTION_KEYS = setOf("FDIV", "FRET", "FLAB")
+
+    /** 异常/单值载荷的显式清除形态：空串、null 字面量、空对象（Oracle 空串即 NULL 语义的显式来源）。 */
+    private fun isClearPayload(value: String): Boolean {
+        val trimmed = value.trim()
+        if (trimmed.isEmpty() || trimmed == "null") return true
+        val node = runCatching { mapper.readTree(trimmed) }.getOrNull() ?: return false
+        return node.isNull || (node.isObject && node.size() == 0)
+    }
+
    private val SEQ_ATTR = mapOf(
        "GTDT" to "GTNO",
        "CKDT" to "CKNO",
@@ -46,6 +57,7 @@ object FlightStateEngine {
                        )
                    }
                }
+                key in EXCEPTION_KEYS && isClearPayload(value) -> scalars[key] = ScalarCommand.Clear
                else -> scalars[key] = ScalarCommand.Set(value)
            }
        }
@@ -62,12 +74,19 @@ object FlightStateEngine {
        val baseCollections = current?.collections?.mapValues { it.value.toMutableList() }
            ?.toMutableMap() ?: mutableMapOf()
        val baseVersion = current?.stateVersion ?: 0L
+        val cleared = mutableSetOf<String>()

        commands.scalars.forEach { (key, cmd) ->
            when (cmd) {
                ScalarCommand.Unchanged -> Unit
-                is ScalarCommand.Set -> baseScalars[key] = cmd.value
-                ScalarCommand.Clear -> baseScalars.remove(key)
+                is ScalarCommand.Set -> {
+                    baseScalars[key] = cmd.value
+                    cleared.remove(key)
+                }
+                ScalarCommand.Clear -> {
+                    baseScalars.remove(key)
+                    cleared += key
+                }
            }
        }

@@ -96,6 +115,7 @@ object FlightStateEngine {
            collections = baseCollections.mapValues { it.value.toList() },
            stateVersion = nextVersion,
            lastMessageId = messageId,
+            clearedKeys = cleared.toSet(),
        )
    }

@@ -518,8 +518,6 @@ class JdbcFlightSchdRepository(

    /** v2：仅标量/异常/文本列 flatten；集合键由明细表读写，不经过槽位列。 */
    private fun flattenScalarsOnly(scalars: Map<String, String>): Map<String, String?> {
-        val fields = linkedMapOf<String, String>()
-        scalars.forEach { (key, value) -> fields[key] = value }
        val out = mutableMapOf<String, String?>()
        scalars.forEach { (key, value) ->
            when {
@@ -527,7 +525,7 @@ class JdbcFlightSchdRepository(
                key in setOf("SRVT", "VIPF", "MAFL") -> out["${key}_TEXT"] = value
            }
        }
-        flattenExceptions(fields, out)
+        flattenExceptions(scalars, out)
        return out
    }

@@ -936,15 +934,31 @@ class JdbcFlightSchdRepository(
        }
    }

-    /** v2 增量：仅更新出现的标量列 + 追踪字段。 */
+    /** 显式清除键 → 需要置 NULL 的物理列（异常前缀列 / 文本列 / 普通标量列）。 */
+    private fun clearColumnsFor(keys: Set<String>): List<String> = keys.flatMap { key ->
+        when {
+            key == "FDIV" -> listOf("FDIV_DDES", "FDIV_DDIR", "FDIV_REMC")
+            key == "FRET" -> listOf("FRET_REID", "FRET_RSN")
+            key == "FLAB" -> listOf("FLAB_ARES", "FLAB_RSN")
+            key == "SRVT" -> listOf("SRVT_TEXT")
+            key == "VIPF" -> listOf("VIPF_TEXT")
+            key == "MAFL" -> listOf("MAFL_TEXT")
+            key in SCALAR_KEY_SET -> listOf(key)
+            else -> emptyList()
+        }
+    }
+
+    /** v2 增量：仅更新出现的标量列 + 显式清除列置 NULL + 追踪字段。 */
    private fun upsertIncrementalScalars(state: com.gzzn.omms.msgexchange.domain.flight.FlightNextState, now: Instant) {
        val stored = flattenScalarsOnly(state.scalars)
-        if (stored.isEmpty()) {
+        val clearedColumns = clearColumnsFor(state.clearedKeys)
+        if (stored.isEmpty() && clearedColumns.isEmpty()) {
            updateTrackingFields(state, now)
            return
        }
-        val assignments = stored.keys.map { "$it = ?" }.toMutableList()
-        assignments += listOf("last_message_id = ?", "state_version = ?", "updated_at = ?")
+        val assignments = stored.keys.map { "$it = ?" } +
+            clearedColumns.map { "$it = NULL" } +
+            listOf("last_message_id = ?", "state_version = ?", "updated_at = ?")
        val sql = "UPDATE flight_schd SET ${assignments.joinToString(", ")} WHERE flid = ?"
        val sqlTimestamp = now.toSqlTimestamp()
        ds.update(sql) { ps ->
@@ -335,21 +335,17 @@ class StubFlightSchd : FlightSchdRepository {
                )
                detailCollections[state.flid] = state.collections
            } else {
-                val mergedScalars = (existing?.fields?.filterKeys {
-                    it == "FLID" || it !in com.gzzn.omms.msgexchange.infra.persistence.jdbc.FlightDetailTables.DETAIL_COLLECTION_KEYS
-                } ?: emptyMap()) + scalarFields
+                // nextState 由引擎在当前态上合并而来，是完整权威态：直接整体替换（含清除语义）
                records[state.flid] = Record(
                    flid = state.flid,
                    fday = existing?.fday,
-                    fields = mergedScalars,
+                    fields = scalarFields,
                    stateVersion = state.stateVersion,
                    lastMessageId = state.lastMessageId,
                    createdAt = existing?.createdAt ?: now,
                    updatedAt = now,
                )
-                val mergedCollections = (detailCollections[state.flid] ?: emptyMap()).toMutableMap()
-                state.collections.forEach { (key, items) -> mergedCollections[key] = items }
-                detailCollections[state.flid] = mergedCollections
+                detailCollections[state.flid] = state.collections
            }
        }
    }
@@ -57,35 +57,39 @@ class SnapshotFlow(
            return
        }

-        val gen = flightSchd.getGen(day)
-        if (gen?.lastMessageId == messageId) {
-            txManager.inTransaction {
-                procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED)
-            }
-            safeBackfill(head, msg)
-            log.info("snapshot replay no-op id={} day={}", head.cminmsgsId, day)
-            return
-        }
-
-        val expected = gen?.version ?: 0L
-        val newFlids = normalized.map { it.first }.toSet()
-        val delFields = gen?.flids?.minus(newFlids) ?: emptySet()
-        val newVersion = expected + 1L
-
-        val nextStates = normalized.map { (flid, fields) ->
-            val current = flightSchd.findNextStateByFlid(flid)
-            val commands = FlightStateEngine.commandsFromFields(flid, fields, snapshotReplace = true)
-            FlightStateEngine.apply(current, commands, messageId, bumpVersion = true)
-        }
-
+        // v2 §5 事务流程：锁外只做解析与整包校验；取得 PIPELINE_LOCK 后在锁内
+        // 复核快照身份 → 读取当前代与当前航班态 → 计算 nextState → 写入 → 提交。
+        var replayNoOp = false
        try {
            txManager.inTransaction {
+                // 锁内复核快照身份：同一消息已成功提交 → 重放短路（不加版本、不重复发事件）
+                val gen = flightSchd.getGen(day)
+                if (gen?.lastMessageId == messageId) {
+                    procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED)
+                    replayNoOp = true
+                    return@inTransaction
+                }
+
+                val expected = gen?.version ?: 0L
+                val newFlids = normalized.map { it.first }.toSet()
+                val delFields = gen?.flids?.minus(newFlids) ?: emptySet()
+                val newVersion = expected + 1L
+
+                // 锁内读取当前航班态，计算 nextState（单写者互斥下读到的一定是已提交最新态）
+                val nextStates = normalized.map { (flid, fields) ->
+                    val current = flightSchd.findNextStateByFlid(flid)
+                    val commands = FlightStateEngine.commandsFromFields(flid, fields, snapshotReplace = true)
+                    FlightStateEngine.apply(current, commands, messageId, bumpVersion = true)
+                }
+
                flightSchd.persistNextStates(day, nextStates, snapshotReplace = true)

                if (delFields.isNotEmpty()) {
                    flightSchd.deleteDiffByDay(day, delFields)
                }

+                // 锁内 CAS：expected 即锁内刚读到的最新版本，仍失败属数据异常——
+                // 一律回滚进入可重试 FAILED(INFRA)，绝不凭版本号推断“已经是我的提交”（v2 §5）
                val casSuccess = flightSchd.putGenIfVersion(
                    day = day,
                    expected = expected,
@@ -97,14 +101,9 @@ class SnapshotFlow(
                    ),
                )
                if (!casSuccess) {
-                    val again = flightSchd.getGen(day)
-                    val replayIdentity = again?.lastMessageId == messageId && again.version == newVersion
-                    if (!replayIdentity) {
-                        throw CasConflictException(
-                            "gen-cas-conflict: expected=$expected current=${again?.version} msg=${again?.lastMessageId}",
-                        )
-                    }
-                    log.info("gen idempotent replay within tx id={} day={}", head.cminmsgsId, day)
+                    throw CasConflictException(
+                        "gen-cas-conflict: expected=$expected msg=$messageId (lock-held CAS must not fail)",
+                    )
                }

                val schdKind = msg.kind as? MsgKind.Schd
@@ -128,8 +127,13 @@ class SnapshotFlow(
                procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED)
            }

+            // 提交后补偿：重放短路同样补做回填（其待办可能仍在重试中）
            safeBackfill(head, msg)
-            log.info("snapshot SUCCEEDED id={} day={} flights={}", head.cminmsgsId, day, normalized.size)
+            if (replayNoOp) {
+                log.info("snapshot replay no-op id={} day={}", head.cminmsgsId, day)
+            } else {
+                log.info("snapshot SUCCEEDED id={} day={} flights={}", head.cminmsgsId, day, normalized.size)
+            }
        } catch (e: CasConflictException) {
            log.warn("gen CAS conflict -> FAILED(INFRA) id={} msg={}", head.cminmsgsId, e.message)
            procFailure.fail(head, ErrorClass.INFRA, e.message ?: "gen-cas-conflict")