cdia/msgexchange-v2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(processing): Wave1 — IgnoreRules/ROUT截断/Q9文档/schd FLTR 数组（ACM2-94/88/81/96）

Browse Source 
REGN/RSTA/EROR 退出忽略清单；ROUT/ERUT 升序留 4 且不落 SCAT/SCDT；C-1/Q9 统一库方清除；flushSchd 聚合成无 key 的 FLTR 数组。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
windyboy
2026-09-21 14:18:40 +08:00
co-authored by Cursor
parent d298fcc48f
commit 23b74ed554
18 changed files with 268 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodec.kt
+2
View File
@@ -68,6 +68,7 @@ class JacksonXmlCodec : XmlCodec {
MsgKind.Fdel -> msg.flop?.let(SisWireMapper::flopPayload)
// 参考数据类别报文的正文结构随 13 类各异（ACM2-93 实装 ReferenceDataProcessor 时绑定）
is MsgKind.RefData -> null
MsgKind.Eror -> null
is MsgKind.Unsupported -> null
}
@@ -106,6 +107,7 @@ class JacksonXmlCodec : XmlCodec {
else -> MsgKind.Unsupported("FLOP-$styp")
}
in REF_DATA_TYPES -> MsgKind.RefData(type)
"EROR" -> MsgKind.Eror
else -> MsgKind.Unsupported("$type-$styp")
}
src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt
+27 -18
View File
@@ -18,8 +18,10 @@ interface DeliveryPort {
/** 发一条变化通知，消息 key 是 FLID（航班实例 ID）。 */
fun sendKafka(topic: String, key: String, payloadJson: String)
/** 发一条完整状态，消息 key 是 FLID。适配层负责把 target 映射成 topic：KAFKA:msg 对应 "msg"，KAFKA:schd 对应 "schd"。 */
fun sendKafkaSchd(topic: String, key: String, payloadJson: String)
/**
* 发一整批 schd 状态：`payloadJson` 是 `SCHD.FLTR` JSON 数组（`C-9`），**不设 message key**。
*/
fun sendKafkaSchd(topic: String, payloadJson: String)
/** 发一条删除通知：key 是 FLID、value 为空；下游按"整态里这个键没了"理解成删除。 */
fun sendKafkaNull(topic: String, key: String)
@@ -33,8 +35,8 @@ interface DeliveryPort {
*
* KAFKA_MSG 按 `EVENT_ID` 顺序发，保序以 FLID 为单位（D2）：某航班的队头失败或退避未到期
* 只暂停该 FLID，其他航班照常推进。KAFKA_SCHD 走 flushSchd：outbox 每个 FLID 只保留
* 一行（写入侧单行 upsert），发出后按读取时刻的代次做条件确认；删除通知发 value 为空的 tombstone。
* 两个主题之间不保证先后顺序。
* 一行（写入侧单行 upsert），聚合成一条无 key 的 `SCHD.FLTR` JSON 数组发出（`C-9`）；
* 删航班只走 msg。两个主题之间不保证先后顺序。
*
* 失败处理：某 FLID 队首的重试时间没到就跳过该 FLID；发送失败重试次数加一并推后退避，
* 次数用尽转 DEAD 当死信。见 docs/implementation.md「Kafka 与读取」。
@@ -46,6 +48,7 @@ class Dispatcher(
private val props: PipelineProps,
private val scheduler: FailureScheduler,
private val counters: com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters,
private val mapper: com.fasterxml.jackson.databind.ObjectMapper,
) {
private val log = org.slf4j.LoggerFactory.getLogger(Dispatcher::class.java)
@@ -152,7 +155,10 @@ class Dispatcher(
}
}
/** 发 KAFKA_SCHD：每个 FLID 只有一行待发事件，删除通知发空 value，成功后按代次条件确认。 */
/**
* 发 `KAFKA:schd`：本 tick 窗口内待发 UPSERT 聚成一条 `SCHD.FLTR` JSON 数组、不设 key（`C-9`）。
* 空窗口不发。schd 侧 TOMBSTONE 不发送（删航班只走 msg）。
*/
internal fun flushSchd() {
val batch = try {
msgEvents.mergePendingSchd(scheduler.now(), props.schd.flushLimit)
@@ -160,25 +166,28 @@ class Dispatcher(
log.warn("mergePendingSchd failed: {}", e.message)
return
}
if (batch.isEmpty()) {
val upserts = batch.filter { it.eventType == EventType.UPSERT }
if (upserts.isEmpty()) {
lastFlush = scheduler.now()
return
}
val failures = mutableListOf<Pair<MsgEvent, String>>()
for (e in batch) {
val fltrs: List<Any> = upserts.map { e ->
try {
when (e.eventType) {
EventType.TOMBSTONE -> port.sendKafkaNull("schd", e.partitionKey)
EventType.UPSERT -> port.sendKafkaSchd("schd", e.partitionKey, e.payloadJson)
}
// 条件确认：读取时刻的代次（EVENT_ID + STATE_VERSION）被新写入覆盖时不标记，留待下一轮重发。
e.eventId?.let { msgEvents.markSentIfVersion(it, e.stateVersion, scheduler.now()) }
} catch (ex: Exception) {
// 首次失败就带上真实原因；不能退回字面量把根因抹掉
failures += e to (ex.message ?: ex.javaClass.simpleName)
mapper.readValue(e.payloadJson, Any::class.java)
} catch (_: Exception) {
e.payloadJson
}
}
failures.forEach { (e, error) -> retryOrDead(e, error) }
val arrayJson = mapper.writeValueAsString(fltrs)
try {
port.sendKafkaSchd("schd", arrayJson)
for (e in upserts) {
e.eventId?.let { msgEvents.markSentIfVersion(it, e.stateVersion, scheduler.now()) }
}
} catch (ex: Exception) {
val error = ex.message ?: ex.javaClass.simpleName
upserts.forEach { e -> retryOrDead(e, error) }
}
lastFlush = scheduler.now()
}
src/main/kotlin/com/gzzn/omms/msgexchange/domain/DecodedMessage.kt
+7 -1
View File
@@ -14,7 +14,7 @@ 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 单独成类，不走这里
@@ -26,6 +26,11 @@ sealed interface MsgKind {
*/
data class RefData(val type: String) : MsgKind
/**
* AODB 错误回报（US-09 AC3）：须匹配出站请求并标 FAILED；协调器落地前保持可重试失败并告警。
*/
data object Eror : MsgKind
/** 已识别但还没有对应处理器的类型：跳过留档记 `SKIPPED(unsupported)`，不作为可重试失败占住队头。 */
data class Unsupported(val tag: String) : MsgKind
@@ -45,6 +50,7 @@ data class DecodedMessage(
is MsgKind.Flop -> "FLOP-${k.subtype}"
MsgKind.Fdel -> "FDEL"
is MsgKind.RefData -> "REF-${k.type}"
MsgKind.Eror -> "EROR"
is MsgKind.Unsupported -> k.tag
}
}
src/main/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngine.kt
+24 -2
View File
@@ -22,6 +22,28 @@ object FlightStateEngine {
"GTDT", "CKDT", "CLDT", "PSDT", "CHDT", "DELY", "ABTM", "CHOT", "ROUT", "ERUT",
)
private val ROUTE_COLLECTION_KEYS = setOf("ROUT", "ERUT")
private const val ROUTE_POINT_LIMIT = 4
/**
* US-05 AC3：ROUT/ERUT 各自最多 4 条；按上游序号（RTNO / 读回的 SOURCE_SEQ）升序保留，
* 不写入当前态的 SCAT/SCDT。
*/
fun normalizeRouteCollection(items: List<Map<String, String>>): List<Map<String, String>> =
items.withIndex()
.sortedWith(compareBy({ (_, row) -> routeSeqKey(row) }, { (idx, _) -> idx }))
.map { it.value }
.take(ROUTE_POINT_LIMIT)
.map { row -> row.filterKeys { it != "SCAT" && it != "SCDT" } }
private fun routeSeqKey(row: Map<String, String>): Int {
val raw = row["RTNO"] ?: row["SOURCE_SEQ"]
return raw?.trim()?.toIntOrNull() ?: Int.MAX_VALUE
}
private fun normalizeCollection(key: String, items: List<Map<String, String>>): List<Map<String, String>> =
if (key in ROUTE_COLLECTION_KEYS) normalizeRouteCollection(items) else items
/**
* 校验一包日计划：报头声明的记录数（RECS）要在 0–9999 之间、且等于实际收到的条数；
* 每条记录的 FLID 必须是 1–12 位数字，快照内不能有重复 FLID，每条记录的运营日都要
@@ -99,7 +121,7 @@ object FlightStateEngine {
putAll(current.collections) // 报文没带的集合保持原值
}
record.collections.forEach { (key, items) ->
if (key in COLLECTION_KEYS) put(key, items) // 报文带了整个集合就整体替换
if (key in COLLECTION_KEYS) put(key, normalizeCollection(key, items))
}
}
return FlightSnapshot(
@@ -124,7 +146,7 @@ object FlightStateEngine {
val collections = buildMap {
putAll(current.collections)
change.collections.forEach { (key, items) ->
if (key in COLLECTION_KEYS) put(key, items)
if (key in COLLECTION_KEYS) put(key, normalizeCollection(key, items))
}
}
return current.copy(
src/main/kotlin/com/gzzn/omms/msgexchange/infra/kafka/KafkaDeliveryPort.kt
+3 -2
View File
@@ -29,8 +29,9 @@ class KafkaDeliveryPort(
producer.send(ProducerRecord(topic, key, payloadJson)).get()
}
override fun sendKafkaSchd(topic: String, key: String, payloadJson: String) {
producer.send(ProducerRecord(topic, key, payloadJson)).get()
override fun sendKafkaSchd(topic: String, payloadJson: String) {
// C-9：不设 message key
producer.send(ProducerRecord<String, String>(topic, null, payloadJson)).get()
}
override fun sendKafkaNull(topic: String, key: String) {
src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt
+23 -5
View File
@@ -8,6 +8,7 @@ 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.domain.flight.FlightStateEngine
import com.gzzn.omms.msgexchange.domain.flight.FlightMainRow
import com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot
import com.gzzn.omms.msgexchange.domain.flight.FlightState
@@ -679,7 +680,12 @@ class JdbcFlightStateRepository(
val collections = linkedMapOf<String, List<Map<String, String>>>()
COLLECTIONS.forEach { (key, spec) ->
collections[key] = loadDetails(flid, key, spec)
val loaded = loadDetails(flid, key, spec)
collections[key] = if (key == "ROUT" || key == "ERUT") {
FlightStateEngine.normalizeRouteCollection(loaded)
} else {
loaded
}
}
return FlightSnapshot(
flid = main.flid,
@@ -812,8 +818,20 @@ class JdbcFlightStateRepository(
private fun replaceDetails(snapshot: FlightSnapshot, now: Instant) {
val flid = snapshot.flid
COLLECTIONS.forEach { (key, spec) ->
ds.update("DELETE FROM ${spec.table} WHERE flid = ?", { ps -> ps.setString(1, flid) })
val items = snapshot.collections[key].orEmpty()
if (spec.routeKind != null) {
ds.update(
"DELETE FROM ${spec.table} WHERE flid = ? AND route_kind = ?",
{ ps ->
ps.setString(1, flid)
ps.setString(2, spec.routeKind)
},
)
} else {
ds.update("DELETE FROM ${spec.table} WHERE flid = ?", { ps -> ps.setString(1, flid) })
}
val items = snapshot.collections[key].orEmpty().let { raw ->
if (key == "ROUT" || key == "ERUT") FlightStateEngine.normalizeRouteCollection(raw) else raw
}
items.forEachIndexed { ordinal, item ->
val cols = mutableListOf("flid", "ordinal", "source_seq")
val vals = mutableListOf<Any?>(flid)
@@ -915,8 +933,8 @@ class JdbcFlightStateRepository(
"DELY" to DetailSpec("flight_delay", listOf("code", "strt", "dura", "remc"), null),
"ABTM" to DetailSpec("flight_bridge_op", listOf("abdg", "abop", "aotm"), "ASNO"),
"CHOT" to DetailSpec("flight_chock_op", listOf("chid", "chst", "chtm"), "CSNO"),
"ROUT" to DetailSpec("flight_route_point", listOf("apcd", "scat", "scdt"), "RTNO", routeKind = "ROUT"),
"ERUT" to DetailSpec("flight_route_point", listOf("apcd", "scat", "scdt"), "RTNO", routeKind = "ERUT"),
"ROUT" to DetailSpec("flight_route_point", listOf("apcd"), "RTNO", routeKind = "ROUT"),
"ERUT" to DetailSpec("flight_route_point", listOf("apcd"), "RTNO", routeKind = "ERUT"),
)
}
}
src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubAdapters.kt
+3 -3
View File
@@ -13,7 +13,7 @@ import jakarta.inject.Singleton
@Requires(property = "msgx.stubs", value = "true")
@Singleton
class StubDeliveryPort : DeliveryPort {
data class Sent(val topic: String, val key: String, val payload: String?)
data class Sent(val topic: String, val key: String?, val payload: String?)
val sent = mutableListOf<Sent>()
val tombstones: List<Sent> get() = sent.filter { it.payload == null }
@@ -24,8 +24,8 @@ class StubDeliveryPort : DeliveryPort {
sent += Sent(topic, key, payloadJson)
}
override fun sendKafkaSchd(topic: String, key: String, payloadJson: String) {
sent += Sent(topic, key, payloadJson)
override fun sendKafkaSchd(topic: String, payloadJson: String) {
sent += Sent(topic, null, payloadJson)
}
override fun sendKafkaNull(topic: String, key: String) {
src/main/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreRules.kt
+3 -7
View File
@@ -1,18 +1,14 @@
package com.gzzn.omms.msgexchange.processing
/**
* US-04 忽略清单：基线规则为 `LDM-*`、`REGN-*`、`RSTA-*`、`EROR-*`。
* 忽略清单：仅保留文档未要求处理的 `LDM-*`。
*
* 匹配按 `TYPE` 前缀做（`TYPE-*` 表示该 TYPE 下所有 STYP 均忽略）；
* `MetaFields.type` 已由 codec 统一大写，这里直接用大写常量比较。
* 返回命中的规则标签（如 `LDM-*`），未命中返回 null。
* `REGN`/`RSTA` 走参考数据（US-13）；`EROR` 走出站错误回报（US-09）——不得在此跳过。
* 匹配按 `TYPE` 精确相等；返回命中规则标签，未命中返回 null。
*/
object IgnoreRules {
private val rules: List<Pair<String, String>> = listOf(
"LDM" to "LDM-*",
"REGN" to "REGN-*",
"RSTA" to "RSTA-*",
"EROR" to "EROR-*",
)
fun match(type: String): String? {
src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt
+5
View File
@@ -251,6 +251,11 @@ class MessageProcessor(
log.warn("refdata without handler -> FAILED(UNSUPPORTED) msgId={} type={}", head.msgId, kind.type)
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")
}
is MsgKind.Unsupported -> {
// 合法但不支持的类型：跳过留档记 SKIPPED（US-03 AC2；markTerminal 同一条 UPDATE
// 登记回填意图），不当可重试失败占住队头