From 3c577b4d1e56e1f627efd1d3a039c43a5243547b Mon Sep 17 00:00:00 2001 From: windyboy Date: Mon, 21 Sep 2026 11:41:34 +0800 Subject: [PATCH] =?UTF-8?q?fix(processing):=20=E5=90=88=E6=B3=95=E4=B8=8D?= =?UTF-8?q?=E6=94=AF=E6=8C=81=E7=B1=BB=E5=9E=8B=E6=94=B9=E8=AE=B0=20SKIPPE?= =?UTF-8?q?D=20=E7=BB=88=E6=80=81=EF=BC=8C=E5=8F=82=E8=80=83=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=8D=95=E5=88=97=20RefData=EF=BC=88ACM2-83=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - MsgKind.RefData 承接 SIS:3.1~3.14 类别码，白名单分派对齐 implementation.md - Unsupported 由 FAILED(UNSUPPORTED) 重试改为 SKIPPED(unsupported) 终态（US-03 AC2） - RefData 在 ReferenceDataProcessor 实装前保持可重试失败，不被吞为跳过 - 同步更新 IgnoreBranchTest / PipelineSmokeTest 旧断言，补 codec 分派回归 --- .../omms/msgexchange/codec/JacksonXmlCodec.kt | 10 +++++++ .../omms/msgexchange/domain/DecodedMessage.kt | 11 +++++-- .../gzzn/omms/msgexchange/processing/Pump.kt | 14 +++++++-- .../omms/msgexchange/PipelineSmokeTest.kt | 26 +++++++++------- .../msgexchange/codec/JacksonXmlCodecTest.kt | 18 +++++++++++ .../processing/IgnoreBranchTest.kt | 30 +++++++++++++++++-- 6 files changed, 91 insertions(+), 18 deletions(-) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodec.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodec.kt index fb86627..d6af86b 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodec.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodec.kt @@ -66,6 +66,8 @@ class JacksonXmlCodec : XmlCodec { is MsgKind.Schd -> msg.schd?.let(SisWireMapper::scheduleBody) is MsgKind.Flop -> msg.flop?.let(SisWireMapper::flopPayload) MsgKind.Fdel -> msg.flop?.let(SisWireMapper::flopPayload) + // 参考数据类别报文的正文结构随 13 类各异（ACM2-93 实装 ReferenceDataProcessor 时绑定） + is MsgKind.RefData -> null is MsgKind.Unsupported -> null } @@ -86,6 +88,8 @@ class JacksonXmlCodec : XmlCodec { * 子类型分派必须是**白名单**：未知的 SCHD 子类型不能静默当成全量日计划（DNLD）—— * 那会让一条本该"暂不支持、等人工处置"的报文走最重的整包合并写入路径。 * 空 STYP 按 legacy 约定视为 DNLD（该约定待与 SIS 逐类对拍确认，见 Q3）。 + * 静态参考数据类别码（`implementation.md`「静态参考数据」`SIS:3.1`～`3.14`）单列成 + * [MsgKind.RefData]：它们必须走参考数据处理（US-13），不得混入"不支持"跳过。 */ private fun kindOf(type: String, styp: String): MsgKind = when (type) { "SCHD" -> when (styp) { @@ -95,6 +99,12 @@ class JacksonXmlCodec : XmlCodec { else -> MsgKind.Unsupported("SCHD-$styp") } "FLOP" -> if (styp == "FDEL") MsgKind.Fdel else MsgKind.Flop(styp) + in REF_DATA_TYPES -> MsgKind.RefData(type) else -> MsgKind.Unsupported("$type-$styp") } + + private val REF_DATA_TYPES = setOf( + "COUL", "ARPT", "AIRL", "AIRC", "REGN", "ORGN", "FLTL", + "TLST", "GLST", "SLST", "CLST", "BLST", "CHLT", "RSTA", + ) } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/DecodedMessage.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/DecodedMessage.kt index 5fd7585..40f573e 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/DecodedMessage.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/DecodedMessage.kt @@ -14,13 +14,19 @@ data class MetaFields( val dttm: Long, ) -/** 报文的种类：日计划、运行动态、FDEL（航班终止），或者还没支持的类型。分派时用穷尽 when 保证不漏分支。 */ +/** 报文的种类：日计划、运行动态、FDEL（航班终止）、静态参考数据，或者还没支持的类型。分派时用穷尽 when 保证不漏分支。 */ sealed interface MsgKind { data class Schd(val subtype: SchdSubtype) : MsgKind data class Flop(val subtype: String) : MsgKind // 运行动态，subtype 是报文子类型；FDEL 单独成类，不走这里 data object Fdel : MsgKind - /** 还没有对应处理器的类型：先按 UNSUPPORTED 记失败并退避重试，重试次数用尽转 DEAD 等人工处置。 */ + /** + * 静态参考数据（`implementation.md`「静态参考数据」：`SIS:3.1`～`3.14` 的类别码，3.15 是同类别的 RESP）。 + * 必须走参考数据处理（US-13），不得按"不支持"跳过；处理器落地前保持可重试失败并告警。 + */ + data class RefData(val type: String) : MsgKind + + /** 已识别但还没有对应处理器的类型：跳过留档记 `SKIPPED(unsupported)`，不作为可重试失败占住队头。 */ data class Unsupported(val tag: String) : MsgKind enum class SchdSubtype { RESP, DNLD, ADFT } @@ -38,6 +44,7 @@ data class DecodedMessage( is MsgKind.Schd -> "SCHD-${k.subtype.name}" is MsgKind.Flop -> "FLOP-${k.subtype}" MsgKind.Fdel -> "FDEL" + is MsgKind.RefData -> "REF-${k.type}" is MsgKind.Unsupported -> k.tag } } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt index 5882a64..95243bc 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt @@ -245,10 +245,18 @@ class MessageProcessor( if (payload == null) return deadMalformed(head, "missing-flop-body") flopProcessor.apply(head, decoded, payload) } + is MsgKind.RefData -> { + // 参考数据必须走 US-13（implementation.md「分派」），不得按"不支持"跳过； + // ReferenceDataProcessor 尚未实装（ACM2-93，待 Q22），先按可重试失败留队并告警。 + log.warn("refdata without handler -> FAILED(UNSUPPORTED) msgId={} type={}", head.msgId, kind.type) + return procFailure.fail(head, ErrorClass.UNSUPPORTED, "refdata-pending:${kind.type}") + } is MsgKind.Unsupported -> { - // 还没有对应处理器的报文类型：先按可重试的失败处理，等能力补齐，不直接判死 - log.warn("unsupported type -> FAILED(UNSUPPORTED) msgId={} tag={}", head.msgId, kind.tag) - return procFailure.fail(head, ErrorClass.UNSUPPORTED, "no-handler:${kind.tag}") + // 合法但不支持的类型：跳过留档记 SKIPPED（US-03 AC2；markTerminal 同一条 UPDATE + // 登记回填意图），不当可重试失败占住队头 + log.info("unsupported type -> SKIPPED msgId={} tag={}", head.msgId, kind.tag) + procState.markTerminal(head.msgId, ProcStatus.SKIPPED, lastError = "unsupported:${kind.tag}", now = clock.instant()) + return } } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt index e49f592..60c87af 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt @@ -78,36 +78,42 @@ class PipelineSmokeTest { } companion object { - /** 报文头合法，但类型没有对应处理器，会走"暂不支持、先重试"这条路 */ + /** 报文头合法，但类型没有对应处理器且不属参考数据 → 跳过留档（US-03 AC2） */ val UNSUPPORTED_XML = """ AODB120260908120000XYZQFOO """.trimIndent() + + /** 参考数据类别（SIS:3.2）：ReferenceDataProcessor 未实装（ACM2-93），先按可重试失败留队 */ + val REFDATA_XML = """ + + AODB220260908120000ARPTDNLD + + """.trimIndent() } @Test - fun `accept then pump tick transitions message to FAILED UNSUPPORTED with backoff`() { + fun `accept then pump tick skips unknown type as SKIPPED with backfill intent`() { val receipt = controller.send(UNSUPPORTED_XML) assertNotNull(receipt.body()) // 受理 ID val id = receipt.body()!!.toLong() // 兼容入口只保证"已落信 + 已入队"，**不推进水位**；必须先被收报发现（W 追平）才可领取。 ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce(Instant.now()) - pump.tick() // 解码成功但无 DELY Handler → FAILED(UNSUPPORTED) + pump.tick() // 解码成功但无 XYZQ Handler → SKIPPED(unsupported) val stub = ctx.getBean(StubProcState::class.java) val s = stub.snapshotOf(id) assertNotNull(s) - assertEquals(ProcStatus.FAILED, s!!.state) - assertEquals(ErrorClass.UNSUPPORTED, s.errorClass) - assertEquals(1, s.attempts) - assertNotNull(s.nextAttemptAt) + assertEquals(ProcStatus.SKIPPED, s!!.state) + assertEquals("unsupported:XYZQ-FOO", s.lastError) + assertNotNull(s.backfillNextAt) } @Test fun `replay reopens failed UNSUPPORTED row to PENDING`() { - val receipt = controller.send(UNSUPPORTED_XML) + val receipt = controller.send(REFDATA_XML) val id = receipt.body()!!.toLong() ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce(Instant.now()) pump.tick() @@ -251,8 +257,8 @@ class PipelineSmokeTest { poller.pollOnce(Instant.now()) assertTrue(cursor.cursor.committedUpTo >= high) repeat(3) { pump.tick() } // 依次处理 2、3、高 ID - assertEquals(ProcStatus.FAILED, proc.find(high)!!.state) - assertEquals(ErrorClass.UNSUPPORTED, proc.find(high)!!.errorClass) + assertEquals(ProcStatus.SKIPPED, proc.find(high)!!.state) + assertEquals("unsupported:XYZQ-FOO", proc.find(high)!!.lastError) } @Test diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodecTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodecTest.kt index f8767b2..8f71ec3 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodecTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodecTest.kt @@ -232,4 +232,22 @@ class JacksonXmlCodecTest { assertEquals(MsgKind.Fdel, result.message.kind) assertEquals("121112312", (result.message.body as FlopPayload).flid) } + + @Test + fun `reference data category types dispatch to RefData and unknown types stay Unsupported`() { + fun raw(type: String, styp: String) = """ + AODB920021010090311 + $type$styp + """.trimIndent() + + // SIS:3.1～3.14 类别码单列为 RefData（US-13），任一 STYP 都不得落成 Unsupported + for (type in listOf("COUL", "ARPT", "AIRL", "AIRC", "REGN", "ORGN", "FLTL", "TLST", + "GLST", "SLST", "CLST", "BLST", "CHLT", "RSTA")) { + val msg = (codec.decode(raw(type, "DNLD")) as DecodeResult.Ok).message + assertEquals(MsgKind.RefData(type), msg.kind) + } + + val unknown = (codec.decode(raw("XYZZ", "TEST")) as DecodeResult.Ok).message + assertEquals(MsgKind.Unsupported("XYZZ-TEST"), unknown.kind) + } } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt index 602f7e3..d356de6 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt @@ -30,7 +30,8 @@ import java.time.Instant /** * 忽略类报文（US-04）：命中忽略清单的报文先绑定身份再写 SKIPPED， - * 不产生航班或 outbox 副作用；未命中的合法类型继续按 UNSUPPORTED 处理。 + * 不产生航班或 outbox 副作用；未命中且无处理器的合法类型同样落 SKIPPED(unsupported) + * 终态（US-03 AC2），参考数据类别（US-13）在处理器落地前保持可重试失败。 */ class IgnoreBranchTest { @@ -173,7 +174,7 @@ class IgnoreBranchTest { } @Test - fun `non-ignored unsupported type continues as UNSUPPORTED`() { + fun `non-ignored unsupported type is skipped as terminal with backfill intent`() { val proc = StubProcState() proc.insertIfAbsent(msgId, null) val inbox = StubInbox() @@ -184,10 +185,33 @@ class IgnoreBranchTest { p.processOne(head()) - assertEquals(ProcStatus.FAILED, proc.find(msgId)!!.state) + val row = proc.find(msgId)!! + assertEquals(ProcStatus.SKIPPED, row.state) + assertEquals("unsupported:XYZZ-TEST", row.lastError) + assertNotNull(row.backfillNextAt) assertEquals(0L, counters.ignoredCount()) } + @Test + fun `reference data type without handler stays retryable instead of being swallowed`() { + val proc = StubProcState() + proc.insertIfAbsent(msgId, null) + val inbox = StubInbox() + inbox.raws[msgId] = "" + val msg = DecodedMessage( + meta = MetaFields("AODB", "ARPT", "DNLD", 400L, 1L), + kind = MsgKind.RefData("ARPT"), + rawXml = "", + ) + val p = processor(proc = proc, inbox = inbox, codec = codecReturning(msg)) + + p.processOne(head()) + + val row = proc.find(msgId)!! + assertEquals(ProcStatus.FAILED, row.state) + assertEquals("refdata-pending:ARPT", row.lastError) + } + @Test fun `ignore increments counter`() { val proc = StubProcState()