feat(processing): 实现自有 PostgreSQL 运营航班权威存储与单事务闭环 (ACM2-28)

- FS1: 增加 Flyway 迁移 V1.1.0__flight_schd.sql，创建 FLIGHT_SCHD 与 SCHD_GEN
- FS2: 实现 FlightSchdRepository 接口及 JdbcFlightSchdRepository 与 StubFlightSchd，增强 JdbcOps 事务管理
- FS3: 扩展 MessageProcessor 事务 2 与按 FLID 点查视图，合并变更、事件与终态入单事务提交
- FS4: SnapshotFlow SQL 化（批处理 upsert、域内差删、SQL CAS 推进与熔断保护），JobExecutor 接入 PG 清场删除
- FS5: 彻底退役 Redis 权威与写路径，移除 FlightRedisClient、Lua 脚本、健康指示器与配置残留
- FS6: 补齐 U09/U29 不变量门禁（崩溃幂等、CAS 防并发、ADFT 存活保障、非 UTC JVM/会话时区无漂移）与 FlywayMigrationTest
- FS7: 交付影子对拍比较内核 FlightStoreDiffTool 与单元测试
- FS8: 全面回改 decision-flight-state、architecture、design、user-stories 权威文档与规范

src/main/kotlin/com/gzzn/omms/msgexchange/processing/Handler.kt

@@ -11,7 +11,7 @@ import com.gzzn.omms.msgexchange.domain.MsgKind
 interface Handler {
     val kind: MsgKind
 
-    /** 在线状态以只读快照传入（阶段 A 读 Redis 权威、阶段 B 读 FLIGHT_STATE，由调用方装配）。 */
+    /** 在线状态以只读视图传入（阶段 A 点查自有 PG FLIGHT_SCHD，由调用方装配，ACM2-28 定案）。 */
     fun decide(flightView: Map<String, String>, msg: DecodedMessage): Decision
 }
 

src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt

@@ -11,7 +11,8 @@ import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository
 import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository
 import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository
 import com.gzzn.omms.msgexchange.infra.persistence.PumpJobRepository
-import com.gzzn.omms.msgexchange.infra.redis.FlightRedisClient
+import com.gzzn.omms.msgexchange.infra.persistence.FlightSchdRepository
+import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager
 import com.gzzn.omms.msgexchange.infra.retry.ProcFailure
 import com.gzzn.omms.msgexchange.jobs.JobExecutor
 import jakarta.inject.Singleton
@@ -20,7 +21,7 @@ import java.time.Instant
 
 /**
  * ACMA-8 流程 2：处理主泵——严格 FIFO + HOL + 毒丸 DEAD 升级（I1）；
- * 阶段 A 全部 Redis 写在本线程（I5），且先于事件创建（I2 happens-before）。
+ * 阶段 A 运营航班 FLIGHT_SCHD 与待发事件、终态同事务原子提交（I2/I5，ACM2-28 定案）。
  */
 @Singleton
 class Pump(
@@ -114,12 +115,25 @@ class MessageProcessor(
     private val msgEvents: MsgEventRepository,
     private val codecHolder: CodecHolder,
     private val handlers: HandlerHolder,
-    private val redis: FlightRedisClient,
+    private val flightSchd: FlightSchdRepository,
     private val snapshotFlow: SnapshotFlow,
     private val procFailure: ProcFailure,
     private val props: PipelineProps,
+    private val txManager: PipelineTransactionManager,
 ) {
     private val log = org.slf4j.LoggerFactory.getLogger(MessageProcessor::class.java)
+    private val flidRegex = Regex("<FLID>(.*?)</FLID>", RegexOption.IGNORE_CASE)
+
+    private fun extractCandidateFlids(decoded: DecodedMessage): Set<String> {
+        val fromXml = flidRegex.findAll(decoded.rawXml).map { it.groupValues[1].trim() }.filter { it.isNotEmpty() }.toSet()
+        if (fromXml.isNotEmpty()) return fromXml
+        val b = decoded.body
+        if (b is Map<*, *>) {
+            val flid = b["FLID"] ?: b["flid"]
+            if (flid != null) return setOf(flid.toString())
+        }
+        return emptySet()
+    }
 
     fun processOne(head: ProcState) {
         com.gzzn.omms.msgexchange.infra.log.TraceLog.withTrace(head.cminmsgsId) {
@@ -176,8 +190,7 @@ class MessageProcessor(
                 return
             }
         }
-
-        // 快照消息走专属流程（流程 4：staging→Lua→putGenIfVersion CAS 同事务）
+        // 快照消息走专属流程（流程 4：staging → SQL 批处理/域内差删/CAS 同事务）
         val handler = handlers.registry.dispatcherFor(decoded)
         if (handler == null) {
             // U10（N21）：未注册 ≠ 报文非法——写 FAILED（可重放），绝不写终态
@@ -193,23 +206,35 @@ class MessageProcessor(
             return
         }
 
-        val decision = handler.decide(redis.hgetAllFlightInfo(), decoded)   // 纯函数
-
-        if (props.phase == PipelineProps.Phase.A) {
-            // 阶段 A：主泵线程先写 Redis（幂等），先于事件创建（I2）；TODO: redisApply(flightChanges)
+        val candidateFlids = extractCandidateFlids(decoded)
+        val flightView = if (candidateFlids.isNotEmpty()) {
+            flightSchd.findByFlids(candidateFlids)
+        } else {
+            emptyMap()
         }
-        // 事务 2（自有 PG 内原子）：事件 + SUCCEEDED（实装后 @Transactional）；
-        // CMINMSGS 回填 = 共享信箱外部副作用（最终一致，ACM2-12）；
-        // 静态主数据（refUpserts）同自有 PG 但弱事务独立提交（21 类 REF_MASTER）。
+
+        val decision = handler.decide(flightView, decoded)   // 纯函数
+
+        // 事务 2（自有 PG 单事务原子）：
+        // upsert FLIGHT_SCHD(decision.flightChanges) + insert MSG_EVENT + PROC_STATE → SUCCEEDED
         val events = buildList {
             decision.msgNotifies.forEach { add(MsgEvent(target = Targets.KAFKA_MSG, payloadJson = it.payloadJson)) }
             decision.schdPush.forEach { add(MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = it.flid, payloadJson = it.fltrJson)) }
         }
-        msgEvents.insertAll(events)
+
+        txManager.inTransaction {
+            if (decision.flightChanges.isNotEmpty()) {
+                flightSchd.upsertIncremental(decision.flightChanges)
+            }
+            if (events.isNotEmpty()) {
+                msgEvents.insertAll(events)
+            }
+            procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED)
+        }
+
+        // 提交后：backfill CMINMSGS（共享信箱外部副作用，补偿链路）
         inbox.backfillOnSuccess(head.cminmsgsId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn)
-        procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED)
-        log.info("SUCCEEDED id={} events={}", head.cminmsgsId, events.size)
-        // 阶段 B：flightState.apply(decision.flightChanges) 进入同一事务；投影事件（ES/REDIS）追加。
+        log.info("SUCCEEDED id={} events={} flightChanges={}", head.cminmsgsId, events.size, decision.flightChanges.size)
     }
 }
 

src/main/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlow.kt

@@ -4,59 +4,114 @@ import com.gzzn.omms.msgexchange.domain.DecodedMessage
 import com.gzzn.omms.msgexchange.domain.ErrorClass
 import com.gzzn.omms.msgexchange.domain.ProcState
 import com.gzzn.omms.msgexchange.domain.ProcStatus
+import com.gzzn.omms.msgexchange.domain.MsgEvent
+import com.gzzn.omms.msgexchange.domain.MsgKind
+import com.gzzn.omms.msgexchange.domain.Targets
+import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository
+import com.gzzn.omms.msgexchange.infra.persistence.FlightSchdRepository
+import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository
+import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager
 import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository
-import com.gzzn.omms.msgexchange.infra.persistence.RefDataRepository
-import com.gzzn.omms.msgexchange.infra.redis.FlightRedisClient
-import com.gzzn.omms.msgexchange.infra.redis.RedisScript
+import com.gzzn.omms.msgexchange.infra.persistence.ReqTrackRepository
 import com.gzzn.omms.msgexchange.infra.retry.ProcFailure
 import jakarta.inject.Singleton
 
 /**
- * ACMA-8 流程 4：日计划快照（generation，主泵内执行）。
- * staging（内存瞬态，崩溃从 raw 整包重放）→ Lua 原子“覆盖+按代差删”（I4/I5）
- * → putGenIfVersion CAS + SUCCEEDED（重放幂等，版本不二次自增——恢复协议细节见 U09）。
+ * staging（内存瞬态，崩溃从 raw 整包重放）→ PG 单事务原子“覆盖+按代差删+SQL CAS 推进+事件入队+SUCCEEDED”（I4/I5，ACM2-28 定案）。
  * U10/T07 占位安全化 + U08 统一失败迁移（ProcFailure）：staging 未实装 → FAILED(UNSUPPORTED)+退避
  * （可重放，绝不写终态）；CAS 冲突 → FAILED(INFRA)+退避；达上限统一 DEAD(EXHAUSTED)。
  */
 @Singleton
 class SnapshotFlow(
     private val procState: ProcStateRepository,
-    private val refData: RefDataRepository,
-    private val redis: FlightRedisClient,
+    private val flightSchd: FlightSchdRepository,
+    private val msgEvents: MsgEventRepository,
+    private val reqTrack: ReqTrackRepository,
     private val procFailure: ProcFailure,
+    private val txManager: PipelineTransactionManager,
+    private val inbox: CminmsgInboxRepository,
 ) {
     private val log = org.slf4j.LoggerFactory.getLogger(SnapshotFlow::class.java)
 
     fun publishSnapshot(head: ProcState, msg: DecodedMessage) {
-        // 1) staging：流式解析 + 整包校验（TODO(阶段2): 流式 codec；千级 FLTR 为 MB 级，内存瞬态）
-        val staged = StageResult.stagingOf(msg) // 骨架：TODO 解析 FLTR 集与重组（KEEP 现役 MAFL/登机桥规则）
+        // 1) staging：流式解析 + 整包校验（内存瞬态，崩溃从 raw 整包重放）
+        val staged = StageResult.stagingOf(msg)
         if (staged is StageResult.Invalid) {
             log.warn("staging not implemented -> FAILED(UNSUPPORTED) id={} reason={}", head.cminmsgsId, staged.reason)
             procFailure.fail(head, ErrorClass.UNSUPPORTED, staged.reason)   // U10：未实装 → 可重放，非终态
             return
         }
-        val normalized = (staged as StageResult.Ok).flights            // (flid, payloadJson)
+        val ok = staged as StageResult.Ok
+        val normalized = ok.flights            // (flid, payloadJson)
+        val day = ok.day
 
-        // 2) 发布：Lua 原子覆盖 + 按代差删（删除集 = gen.flids − 新代，ADFT 自动存活）
-        val day = staged.day
-        val gen = refData.getGen(day)
-        val newFlids = normalized.map { it.first }
-        val delFields = gen?.flids?.minus(newFlids.toSet()) ?: emptyList()
-        redis.eval(RedisScript.SNAPSHOT_REPLACE, setPairs = normalized, delFields = delFields)
-
-        // 3) 事务：putGen CAS + SUCCEEDED（实装后同 @Transactional；CAS 失败=并发，串行泵下不应发生→告警）
+        // 内存与超大包熔断防御（ACM2-28 评论 4 项 3）
+        if (normalized.size > MAX_FLIGHTS_PER_SNAPSHOT) {
+            log.error("snapshot flights exceed limit -> DEAD(MALFORMED) id={} size={}", head.cminmsgsId, normalized.size)
+            procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED,
+                lastError = "flights-exceed-limit:${normalized.size}")
+            return
+        }
+
+        // 2) 准备代元数据与差集（删除集 = gen.flids − 新代，ADFT 自动存活）
+        val gen = flightSchd.getGen(day)
         val expected = gen?.version ?: 0L
-        if (!refData.putGenIfVersion(day, expected, RefDataRepository.GenMeta(newFlids, expected + 1))) {
-            // 重放路径：version 已是目标值 → no-op 视为成功
-            val again = refData.getGen(day)
-            if (again == null || again.version != expected + 1) {
-                log.warn("gen CAS conflict -> FAILED(INFRA) id={}", head.cminmsgsId)
-                procFailure.fail(head, ErrorClass.INFRA, "gen-cas-conflict")   // N06/N28：带退避，禁止紧循环
-                return
+        val newFlids = normalized.map { it.first }.toSet()
+        val delFields = gen?.flids?.minus(newFlids) ?: emptySet()
+        val newVersion = expected + 1L
+
+        // 3) 自有 PG 单事务原子提交（ACM2-28 定案：覆盖新代 + 域内差删 + SQL CAS + 事件 + SUCCEEDED）
+        try {
+            txManager.inTransaction {
+                // 覆盖新代全量：强行声明 FDAY 归属（JDBC batch 批处理）
+                flightSchd.upsertSnapshotBatch(day, normalized)
+
+                // 按代差删域化：仅删除 FDAY = day 且在 delFields 中的记录（ADFT 与跨代已迁移行存活）
+                if (delFields.isNotEmpty()) {
+                    flightSchd.deleteDiffByDay(day, delFields)
+                }
+
+                // SQL CAS 版本推进（防双写断言）
+                val casSuccess = flightSchd.putGenIfVersion(
+                    day = day,
+                    expected = expected,
+                    newGen = FlightSchdRepository.GenMeta(fday = day, version = newVersion, flids = newFlids),
+                )
+                if (!casSuccess) {
+                    // 重放路径：version 已是目标值 → no-op 视为成功，否则为 CAS 冲突
+                    val again = flightSchd.getGen(day)
+                    if (again == null || again.version != newVersion) {
+                        throw CasConflictException("gen-cas-conflict: expected=$expected current=${again?.version}")
+                    }
+                }
+
+                // RESP 匹配 REQ_TRACK -> DONE
+                val schdKind = msg.kind as? MsgKind.Schd
+                if (schdKind?.subtype == MsgKind.SchdSubtype.RESP) {
+                    reqTrack.findOpenByKind("SCHD")?.let { req ->
+                        reqTrack.markDone(req.reqId)
+                    }
+                }
+
+                // 构造并批量写入 schd 投递事件
+                val events = normalized.map { (flid, json) ->
+                    MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = flid, payloadJson = json)
+                }
+                if (events.isNotEmpty()) {
+                    msgEvents.insertAll(events)
+                }
+
+                // 终态置 SUCCEEDED
+                procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED)
             }
+
+            // 提交后：backfill CMINMSGS（外部副作用，补偿保障）
+            inbox.backfillOnSuccess(head.cminmsgsId, msg.meta.sndr, msg.meta.type, msg.meta.styp, msg.meta.seqn)
+            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")
         }
-        procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED)
-        log.info("snapshot SUCCEEDED id={} day={} flights={}", head.cminmsgsId, day, normalized.size)
     }
 
     /** staging 结果（骨架）。 */
@@ -65,8 +120,12 @@ class SnapshotFlow(
         data class Invalid(val reason: String) : StageResult
 
         companion object {
+            var parser: ((DecodedMessage) -> StageResult)? = null
+
             fun stagingOf(msg: DecodedMessage): StageResult =
-                Invalid("staging-not-implemented(${msg.typeTag})")   // TODO(阶段2)
+                parser?.invoke(msg) ?: Invalid("staging-not-implemented(${msg.typeTag})")   // TODO(阶段2)
         }
     }
 }
+class CasConflictException(message: String) : RuntimeException(message)
+const val MAX_FLIGHTS_PER_SNAPSHOT = 10000