feat(ingress): 实现 JDBC 信箱轮询、入站持久化与对拍测试 (U05)

2026-09-07 15:11:33 +08:00
parent dc68f1e1f8
commit c7b4b527ef
51 changed files with 1005 additions and 235 deletions
@@ -0,0 +1,34 @@
+package com.gzzn.omms.msgexchange.processing
+
+import com.gzzn.omms.msgexchange.domain.DecodedMessage
+import com.gzzn.omms.msgexchange.domain.Decision
+import com.gzzn.omms.msgexchange.domain.MsgKind
+
+/**
+ * ACMA-8 流程 2：Handler = 纯函数（状态与报文进，Decision 出，不碰 Redis/Kafka）。
+ * 32 个 Handler（flop 29 + schd 3）翻译属阶段 2/3，逐条对照 ACMA-4 基线与兼容矩阵 KEEP/FIX。
+ */
+interface Handler {
+    val kind: MsgKind
+
+    /** 在线状态以只读快照传入（阶段 A 读 Redis 权威、阶段 B 读 FLIGHT_STATE，由调用方装配）。 */
+    fun decide(flightView: Map<String, String>, msg: DecodedMessage): Decision
+}
+
+/**
+ * sealed 穷尽分派（ACMA-6 选型：取代 legacy 反射 get{TYPE}()）。
+ */
+class HandlerRegistry(handlers: List<Handler>) {
+    private val byKind: Map<String, Handler> = handlers.associateBy { keyOf(it.kind) }
+
+    fun dispatcherFor(msg: DecodedMessage): Handler? = byKind[keyOf(msg.kind)]
+
+    companion object {
+        fun keyOf(kind: MsgKind): String = when (kind) {
+            is MsgKind.Schd -> "SCHD-${kind.subtype.name}"
+            is MsgKind.Flop -> "FLOP-${kind.subtype}"
+        }
+    }
+}
+
+// TODO(阶段2/3): 注册 3+29 Handler；本阶段仅含骨架（阶段 2 最小纵向链路先做 1 SCHD(DNLD) + 1 FLOP）。
@@ -0,0 +1,24 @@
+package com.gzzn.omms.msgexchange.processing
+
+import com.gzzn.omms.msgexchange.config.PipelineProps
+import com.gzzn.omms.msgexchange.domain.DecodedMessage
+import java.time.LocalDate
+
+/**
+ * ACMA-8 I3：幂等键 = SNDR|TYPE|STYP|SEQN。
+ * 计算集中此处唯一入口；“是否含日边界”可配置且默认关闭（CONFIRM 矩阵 #11，
+ * SEQN 重置作用域确认前不改语义——上线后不改幂等键）。
+ */
+object Identity {
+    fun of(
+        msg: DecodedMessage,
+        includeDayBoundary: Boolean,
+        day: LocalDate = LocalDate.now(),
+    ): String {
+        val base = "${msg.meta.sndr}|${msg.meta.type}|${msg.meta.styp}|${msg.meta.seqn}"
+        return if (includeDayBoundary) "$base|${day}" else base
+    }
+
+    fun of(msg: DecodedMessage, props: PipelineProps.Identity): String =
+        of(msg, props.includeDayBoundary)
+}
@@ -0,0 +1,218 @@
+package com.gzzn.omms.msgexchange.processing
+
+import com.gzzn.omms.msgexchange.config.PipelineProps
+import com.gzzn.omms.msgexchange.domain.DecodedMessage
+import com.gzzn.omms.msgexchange.domain.ErrorClass
+import com.gzzn.omms.msgexchange.domain.MsgEvent
+import com.gzzn.omms.msgexchange.domain.ProcState
+import com.gzzn.omms.msgexchange.domain.ProcStatus
+import com.gzzn.omms.msgexchange.domain.Targets
+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.retry.ProcFailure
+import com.gzzn.omms.msgexchange.jobs.JobExecutor
+import jakarta.inject.Singleton
+import java.time.Duration
+import java.time.Instant
+
+/**
+ * ACMA-8 流程 2：处理主泵——严格 FIFO + HOL + 毒丸 DEAD 升级（I1）；
+ * 阶段 A 全部 Redis 写在本线程（I5），且先于事件创建（I2 happens-before）。
+ */
+@Singleton
+class Pump(
+    private val procState: ProcStateRepository,
+    private val pumpJobs: PumpJobRepository,
+    private val inbox: CminmsgInboxRepository,
+    private val processor: MessageProcessor,
+    private val jobExecutor: JobExecutor,
+    private val props: PipelineProps,
+) {
+    private val log = org.slf4j.LoggerFactory.getLogger(Pump::class.java)
+
+    @Volatile
+    private var running = true
+
+    /** 优雅停机：loop 在当前 tick 收尾后退出；线程中断由 Runner 负责。 */
+    fun stop() {
+        running = false
+    }
+
+    fun loop() {
+        while (running) {
+            try {
+                tick()
+            } catch (e: InterruptedException) {
+                Thread.currentThread().interrupt()
+                return
+            } catch (e: Exception) {
+                // U08：loop 只作最后防线——失败状态迁移已在 processOne/execute 的边界内完成；
+                // 致命 Error 不在此捕获（任其终止进程，保证“异常必可见”）。
+                sleepQuietly(props.pipeline.pollInterval)
+            }
+        }
+    }
+
+    internal fun tick() {
+        val job = pumpJobs.headQueued()
+        val head = procState.headUnfinished()
+
+        // 统一 FIFO：JOB 与消息同队列语义（定时任务产物不绕过队头顺序，决策 1）。
+        val nextJob = job?.takeIf { jobBefore(job, head) }
+        when {
+            nextJob != null -> execute(nextJob)
+            head == null -> sleepQuietly(props.pipeline.pollInterval)
+            head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > Instant.now() ->
+                if (poisoned(head)) {
+                    log.error("poison -> DEAD id={} attempts={} lastError={}", head.cminmsgsId, head.attempts, head.lastError)
+                    procState.update(head.cminmsgsId, ProcStatus.DEAD,
+                        errorClass = ErrorClass.EXHAUSTED, lastError = head.lastError ?: "head-deadline-exceeded")
+                } else {
+                    sleepQuietly(Duration.between(Instant.now(), head.nextAttemptAt))
+                }
+            // PENDING、或 FAILED 退避已到期：交处理入口（内部有边界化失败迁移与 attempts 守卫）
+            else -> processor.processOne(head)
+        }
+    }
+
+    private fun poisoned(head: ProcState): Boolean =
+        head.attempts >= props.pipeline.maxAttempts ||
+            Duration.between(head.updatedAt, Instant.now()) > props.pipeline.headDeadline
+
+    /** JOB 与队头消息的先后由入队时间近似；实装以统一序号列保证（阶段 1 后续）。 */
+    private fun jobBefore(job: PumpJobRepository.Job, head: ProcState?) =
+        head == null || head.state == ProcStatus.FAILED
+
+    private fun execute(job: PumpJobRepository.Job) {
+        pumpJobs.markRunning(job.jobId)
+        try {
+            jobExecutor.execute(job)
+            pumpJobs.markDone(job.jobId)
+        } catch (e: Exception) {
+            pumpJobs.markFailed(job.jobId, e.message ?: "unknown")
+        }
+    }
+
+    private fun sleepQuietly(d: Duration) {
+        if (!d.isNegative && !d.isZero) Thread.sleep(d.toMillis().coerceAtLeast(1))
+    }
+}
+
+/**
+ * ACMA-8 流程 2 processOne：解码→绑定→纯函数决策→提交。
+ * U08/U10/U11：失败状态迁移全部在“持有具体 head”的边界内完成——
+ * 任何意外异常 → FAILED(INFRA)+退避（ProcFailure）；attempts 达上限 → DEAD(EXHAUSTED)；
+ * MALFORMED 直接 DEAD；InterruptedException 恢复中断位并上抛（不被普通恢复逻辑吞掉）。
+ */
+@Singleton
+class MessageProcessor(
+    private val inbox: CminmsgInboxRepository,
+    private val procState: ProcStateRepository,
+    private val msgEvents: MsgEventRepository,
+    private val codecHolder: CodecHolder,
+    private val handlers: HandlerHolder,
+    private val redis: FlightRedisClient,
+    private val snapshotFlow: SnapshotFlow,
+    private val procFailure: ProcFailure,
+    private val props: PipelineProps,
+) {
+    private val log = org.slf4j.LoggerFactory.getLogger(MessageProcessor::class.java)
+
+    fun processOne(head: ProcState) {
+        com.gzzn.omms.msgexchange.infra.log.TraceLog.withTrace(head.cminmsgsId) {
+            try {
+                processInternal(head)
+            } catch (e: InterruptedException) {
+                Thread.currentThread().interrupt()
+                throw e
+            } catch (e: Exception) {
+                // U08：边界化——异常归于本条 head，写 FAILED/DEAD，而不是穿出杀 pump
+                log.warn("processOne unexpected failure id={} ec=INFRA msg={}", head.cminmsgsId, e.message ?: e.javaClass.simpleName)
+                procFailure.fail(head, ErrorClass.INFRA, e.message ?: e.javaClass.simpleName)
+            }
+        }
+    }
+
+    private fun processInternal(head: ProcState) {
+        // 守卫：手工/遗留 FAILED 行若 attempts 已达上限，直接终态（防止退避到期后无限重试）
+        if (head.state == ProcStatus.FAILED && procFailure.scheduler.exhausted(head.attempts)) {
+            log.error("head exhausted at entry -> DEAD id={} attempts={}", head.cminmsgsId, head.attempts)
+            procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.EXHAUSTED,
+                lastError = head.lastError ?: "max-attempts")
+            return
+        }
+
+        val raw = inbox.rawOf(head.cminmsgsId) ?: run {
+            log.error("raw missing -> DEAD(MALFORMED) id={}", head.cminmsgsId)
+            procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = "raw-missing")
+            return
+        }
+        val decoded = when (val r = codecHolder.codec.decode(raw)) {
+            is com.gzzn.omms.msgexchange.codec.DecodeResult.Ok -> r.message
+            is com.gzzn.omms.msgexchange.codec.DecodeResult.Err -> {
+                // T06（U11）：MALFORMED（报文非法）→ DEAD 不重试；CODEC_ERROR（可随 codec 修复重放）→ FAILED 退避
+                if (r.failure.errorClass == ErrorClass.MALFORMED) {
+                    log.error("decode MALFORMED -> DEAD id={} detail={}", head.cminmsgsId, r.failure.detail)
+                    procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED,
+                        lastError = r.failure.detail)
+                } else {
+                    log.warn("decode {} -> FAILED id={} detail={}", r.failure.errorClass, head.cminmsgsId, r.failure.detail)
+                    procFailure.fail(head, r.failure.errorClass, r.failure.detail)
+                }
+                return
+            }
+        }
+
+        // I3：identity 仅首次绑定（head.identityKey == null）；FAILED 重试不重绑（N25：入参传递 head，不再二次查询）
+        if (head.identityKey == null) {
+            val identity = Identity.of(decoded, props.identity)
+            if (!procState.tryBindIdentity(head.cminmsgsId, identity)) {
+                val owner = procState.ownerOfIdentity(identity) ?: -1L
+                log.info("duplicate-of:{} -> SKIPPED id={}", owner, head.cminmsgsId)
+                procState.update(head.cminmsgsId, ProcStatus.SKIPPED, lastError = "duplicate-of:$owner")
+                return
+            }
+        }
+
+        // 快照消息走专属流程（流程 4：staging→Lua→putGenIfVersion CAS 同事务）
+        val handler = handlers.registry.dispatcherFor(decoded)
+        if (handler == null) {
+            // U10（N21）：未注册 ≠ 报文非法——写 FAILED（可重放），绝不写终态
+            log.warn("no-handler:{} -> FAILED(UNSUPPORTED) id={}", decoded.typeTag, head.cminmsgsId)
+            procFailure.fail(head, ErrorClass.UNSUPPORTED, "no-handler:${decoded.typeTag}")
+            return
+        }
+        val schdKind = decoded.kind as? com.gzzn.omms.msgexchange.domain.MsgKind.Schd
+        if (schdKind != null &&
+            schdKind.subtype == com.gzzn.omms.msgexchange.domain.MsgKind.SchdSubtype.DNLD
+        ) {
+            snapshotFlow.publishSnapshot(head, decoded)
+            return
+        }
+
+        val decision = handler.decide(redis.hgetAllFlightInfo(), decoded)   // 纯函数
+
+        if (props.phase == PipelineProps.Phase.A) {
+            // 阶段 A：主泵线程先写 Redis（幂等），先于事件创建（I2）；TODO: redisApply(flightChanges)
+        }
+        // 事务 2（自有 PG 内原子）：事件 + SUCCEEDED（实装后 @Transactional）；
+        // CMINMSGS 回填 = 共享信箱外部副作用（最终一致，ACM2-12）；
+        // 静态主数据（refUpserts）同自有 PG 但弱事务独立提交（21 类 REF_MASTER）。
+        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)
+        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）追加。
+    }
+}
+
+/** 延迟装配占位（阶段 1 后续以 Micronaut Bean 替换直连构造）。 */
+class CodecHolder(val codec: com.gzzn.omms.msgexchange.codec.XmlCodec)
+class HandlerHolder(val registry: HandlerRegistry)
@@ -0,0 +1,72 @@
+package com.gzzn.omms.msgexchange.processing
+
+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.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.retry.ProcFailure
+import jakarta.inject.Singleton
+
+/**
+ * ACMA-8 流程 4：日计划快照（generation，主泵内执行）。
+ * staging（内存瞬态，崩溃从 raw 整包重放）→ Lua 原子“覆盖+按代差删”（I4/I5）
+ * → putGenIfVersion CAS + SUCCEEDED（重放幂等，版本不二次自增——恢复协议细节见 U09）。
+ * 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 procFailure: ProcFailure,
+) {
+    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/登机桥规则）
+        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)
+
+        // 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 失败=并发，串行泵下不应发生→告警）
+        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
+            }
+        }
+        procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED)
+        log.info("snapshot SUCCEEDED id={} day={} flights={}", head.cminmsgsId, day, normalized.size)
+    }
+
+    /** staging 结果（骨架）。 */
+    sealed interface StageResult {
+        data class Ok(val day: String, val flights: List<Pair<String, String>>) : StageResult
+        data class Invalid(val reason: String) : StageResult
+
+        companion object {
+            fun stagingOf(msg: DecodedMessage): StageResult =
+                Invalid("staging-not-implemented(${msg.typeTag})")   // TODO(阶段2)
+        }
+    }
+}