feat(outbound): REQ_TRACK/COUTMSGS/schd-sync 与 RESP 守卫（ACM2-92/86）

出站请求状态机与编码；SCHD-RESP 须匹配未过期开放请求；EROR 结案匹配 SENT。 Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-21 15:50:14 +08:00
parent b545fd2b58
commit 54b383fa9d
23 changed files with 843 additions and 26 deletions
@@ -0,0 +1,137 @@
+package com.gzzn.omms.msgexchange.processing
+
+import com.gzzn.omms.msgexchange.codec.XmlCodec
+import com.gzzn.omms.msgexchange.config.OperationDayProps
+import com.gzzn.omms.msgexchange.config.PipelineProps
+import com.gzzn.omms.msgexchange.domain.OutboundRequestKeys
+import com.gzzn.omms.msgexchange.infra.persistence.CoutmsgOutboxRepository
+import com.gzzn.omms.msgexchange.infra.persistence.ReqTrackRepository
+import jakarta.inject.Singleton
+import org.slf4j.LoggerFactory
+import java.time.Clock
+import java.time.LocalDate
+import java.time.ZoneId
+import java.time.format.DateTimeFormatter
+
+/**
+ * 出站 REQ_TRACK 协调：登记、COUTMSGS 落信、超时与 EROR 失败（`G-REQ-TRACK`、`C-4`）。
+ */
+@Singleton
+class OutboundRequestService(
+    private val reqTrack: ReqTrackRepository,
+    private val outbox: CoutmsgOutboxRepository,
+    private val codec: XmlCodec,
+    private val props: PipelineProps,
+    operationDayProps: OperationDayProps,
+    private val clock: Clock,
+) {
+    private val log = LoggerFactory.getLogger(OutboundRequestService::class.java)
+    private val zone: ZoneId = operationDayProps.zoneId()
+    private val cutoffHour = operationDayProps.cutoffHour.coerceIn(0, 23)
+    private val dttmFmt = DateTimeFormatter.ofPattern("yyyyMMddHHmmss")
+
+    sealed interface RegisterOutcome {
+        data class Registered(val reqId: Long) : RegisterOutcome
+        data object OpenExists : RegisterOutcome
+    }
+
+    fun currentOperationDay(): LocalDate {
+        val airport = clock.withZone(zone)
+        val now = java.time.LocalDateTime.now(airport)
+        val day = now.toLocalDate()
+        return if (now.hour < cutoffHour) day.minusDays(1) else day
+    }
+
+    fun hasOpenRequest(reqType: String, operationDay: LocalDate = currentOperationDay()): Boolean =
+        reqTrack.findLatest(
+            reqType,
+            operationDay,
+            OutboundRequestKeys.SENDER,
+            listOf(ReqTrackRepository.ReqState.PENDING, ReqTrackRepository.ReqState.SENT),
+        ) != null
+
+    fun registerRqfdSync(): RegisterOutcome {
+        val day = currentOperationDay()
+        if (hasOpenRequest(OutboundRequestKeys.RQFD_REQ_TYPE, day)) return RegisterOutcome.OpenExists
+        val reqId = reqTrack.insert(OutboundRequestKeys.RQFD_REQ_TYPE, day, OutboundRequestKeys.SENDER)
+        dispatchPending(limit = 1)
+        return RegisterOutcome.Registered(reqId)
+    }
+
+    fun hasOpenSentRqfd(operationDay: LocalDate = currentOperationDay()): Boolean =
+        reqTrack.findLatest(
+            OutboundRequestKeys.RQFD_REQ_TYPE,
+            operationDay,
+            OutboundRequestKeys.SENDER,
+            listOf(ReqTrackRepository.ReqState.SENT),
+        ) != null
+
+    fun completeRqfdResponse(operationDay: LocalDate = currentOperationDay()): Boolean =
+        reqTrack.completeLatest(OutboundRequestKeys.RQFD_REQ_TYPE, operationDay, OutboundRequestKeys.SENDER)
+
+    fun dispatchPending(limit: Int = 10): Int {
+        var sent = 0
+        reqTrack.listPendingDispatch(limit).forEach { req ->
+            if (dispatchOne(req)) sent++
+        }
+        return sent
+    }
+
+    fun expireTimedOut(): Int {
+        val cutoff = clock.instant().minus(props.outbound.responseTimeout)
+        return reqTrack.expireSentOlderThan(cutoff)
+    }
+
+    fun failFromEror(seqs: Long, typs: String, stys: String): Boolean {
+        val reqType = OutboundRequestKeys.reqTypeFromEror(typs, stys) ?: return false
+        val day = currentOperationDay()
+        val open = reqTrack.findOpenSentBySeqn(reqType, day, OutboundRequestKeys.SENDER, seqs) ?: return false
+        reqTrack.markFailed(open.reqId)
+        log.warn(
+            "outbound request FAILED from EROR reqId={} type={} seqn={} typs={} stys={}",
+            open.reqId, reqType, seqs, typs, stys,
+        )
+        return true
+    }
+
+    private fun dispatchOne(req: ReqTrackRepository.Req): Boolean {
+        if (req.state != ReqTrackRepository.ReqState.PENDING || req.writeUncertain) return false
+        val seqn = outbox.nextOutboundSeqn()
+        val dttm = beijingMetaDttm()
+        val xml = when (req.reqType) {
+            OutboundRequestKeys.RQFD_REQ_TYPE -> codec.encodeOutboundRqfd(seqn, dttm)
+            OutboundRequestKeys.RQRD_REQ_TYPE -> codec.encodeOutboundRqrd("AIRL", seqn, dttm)
+            else -> {
+                log.warn("unknown req_type for dispatch reqId={} type={}", req.reqId, req.reqType)
+                return false
+            }
+        }
+        val routing = when (req.reqType) {
+            OutboundRequestKeys.RQFD_REQ_TYPE -> props.outbound.routingRqfd
+            else -> props.outbound.routingRqrd
+        }
+        val inserted = outbox.insertMessage(xml, routing)
+        return when (inserted.outcome) {
+            CoutmsgOutboxRepository.InsertOutcome.CONFIRMED -> {
+                val id = inserted.coutmsgsId ?: return false
+                reqTrack.linkCoutmsgs(req.reqId, id, seqn)
+                reqTrack.markSent(req.reqId, clock.instant())
+                log.info("outbound COUTMSGS SENT reqId={} coutmsgsId={} seqn={}", req.reqId, id, seqn)
+                true
+            }
+            CoutmsgOutboxRepository.InsertOutcome.AMBIGUOUS -> {
+                reqTrack.markWriteUncertain(req.reqId)
+                log.error(
+                    "outbound COUTMSGS write AMBIGUOUS reqId={} seqn={} — recorded, no blind resend",
+                    req.reqId, seqn,
+                )
+                false
+            }
+        }
+    }
+
+    private fun beijingMetaDttm(): Long {
+        val ldt = java.time.LocalDateTime.ofInstant(clock.instant(), zone)
+        return dttmFmt.format(ldt).toLong()
+    }
+}
@@ -1,5 +1,6 @@
 package com.gzzn.omms.msgexchange.processing

+import com.gzzn.omms.msgexchange.codec.ErorPayload
 import com.gzzn.omms.msgexchange.codec.FlopPayload
 import com.gzzn.omms.msgexchange.codec.ScheduleBody
 import com.gzzn.omms.msgexchange.codec.XmlCodec
@@ -123,6 +124,7 @@ class MessageProcessor(
    private val flopProcessor: FlopProcessor,
    private val fdelProcessor: FdelProcessor,
    private val adftProcessor: AdftProcessor,
+    private val outbound: OutboundRequestService,
    private val procFailure: ProcFailure,
    private val props: PipelineProps,
    private val clock: Clock,
@@ -201,8 +203,24 @@ class MessageProcessor(
                val body = decoded.body as? ScheduleBody
                if (body == null) return deadMalformed(head, "missing-schd-body")
                when (kind.subtype) {
-                    MsgKind.SchdSubtype.DNLD, MsgKind.SchdSubtype.RESP ->
+                    MsgKind.SchdSubtype.DNLD ->
                        scheduleProcessor.applyScheduleRecords(head, decoded)
+                    MsgKind.SchdSubtype.RESP -> {
+                        if (!outbound.hasOpenSentRqfd()) {
+                            log.info("SCHD-RESP without open RQFD -> SKIPPED msgId={} [G-RESP-GUARD]", head.msgId)
+                            procState.markTerminal(
+                                head.msgId, ProcStatus.SKIPPED,
+                                lastError = "resp-guard:no-open-req",
+                                now = clock.instant(),
+                            )
+                            return
+                        }
+                        val respResult = scheduleProcessor.applyScheduleRecords(head, decoded)
+                        if (respResult is ApplyResult.Succeeded || respResult is ApplyResult.ReplaySkipped) {
+                            outbound.completeRqfdResponse()
+                        }
+                        respResult
+                    }
                    MsgKind.SchdSubtype.ADFT -> {
                        val record = body.records.singleOrNull()
                        if (record == null) return deadMalformed(head, "adft-needs-single-fltr")
@@ -227,9 +245,20 @@ class MessageProcessor(
                return procFailure.fail(head, ErrorClass.UNSUPPORTED, "refdata-pending:${kind.type}")
            }
            MsgKind.Eror -> {
-                // US-09 AC3：须匹配出站请求；REQ_TRACK 协调器落地前（ACM2-92）先按可重试失败留队。
-                log.warn("eror without handler -> FAILED(UNSUPPORTED) msgId={}", head.msgId)
-                return procFailure.fail(head, ErrorClass.UNSUPPORTED, "eror-pending")
+                val payload = decoded.body as? ErorPayload
+                    ?: return deadMalformed(head, "missing-eror-body")
+                val matched = outbound.failFromEror(payload.seqs, payload.typs, payload.stys)
+                if (!matched) {
+                    log.info("EROR without matching open outbound -> SKIPPED msgId={}", head.msgId)
+                    procState.markTerminal(
+                        head.msgId, ProcStatus.SKIPPED,
+                        lastError = "eror:no-matching-req",
+                        now = clock.instant(),
+                    )
+                } else {
+                    procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED, now = clock.instant())
+                }
+                return
            }
            is MsgKind.Unsupported -> {
                // 合法但不支持的类型：跳过留档记 SKIPPED（US-03 AC2；markTerminal 同一条 UPDATE