From 2175c352f6064a23bea10e91fc37b6a3ecfb7f58 Mon Sep 17 00:00:00 2001 From: windyboy Date: Sun, 6 Sep 2026 18:19:04 +0800 Subject: [PATCH] =?UTF-8?q?fix(processing):=20WP1=20=E6=A0=B8=E5=BF=83?= =?UTF-8?q?=E8=AF=AD=E4=B9=89=EF=BC=88ACM2-10=20U06/U08/U10/U11=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - U06/N03：Dispatcher 逐条循环显式排除 KAFKA_SCHD（唯一出口 flushSchd 批量聚合）， 修复 schd 事件被无条件 markSent 吞掉、flush 永远 claim 不到的整链死路；flush 失败整批 保持 PENDING、lastFlush 仅成功后推进（N24） - U08 部分：Pump/Dispatcher loop 异常隔离（单次异常不静默死亡，U12 补告警日志）； processOne FAILED 入口 attempts 毒丸升级 DEAD（防退避到期后无限重试）； N28：backoffFor(attempt≤0) 下界保护不再抛异常 - U10/N21：未注册 handler → FAILED(UNSUPPORTED)+退避（绝不写终态）；占位安全化—— HistorySweep.pickHistory 占位改空集（T07 修订）、SnapshotFlow staging 未实装改 FAILED(UNSUPPORTED)+退避而非 DEAD（N05），CAS 冲突 FAILED(INFRA)+退避（N06 紧循环修复） - U11/T06：decode 失败差异化——MALFORMED→DEAD 不重试；CODEC_ERROR→FAILED 可重放 - 新增 ErrorClass.UNSUPPORTED；单测：DispatcherTickTest(3)/MessageProcessorTest(6)/ PipelinePropsTest(2)，19 个测试全绿 --- .../nextgen/config/PipelineProps.kt | 7 +- .../nextgen/delivery/Dispatcher.kt | 43 +++- .../msgexchange/nextgen/domain/ProcState.kt | 2 +- .../msgexchange/nextgen/jobs/JobExecutor.kt | 8 +- .../msgexchange/nextgen/processing/Pump.kt | 88 +++++-- .../nextgen/processing/SnapshotFlow.kt | 28 ++- .../db/migration/V2.0.0__aux_tables.sql | 2 +- .../nextgen/config/PipelinePropsTest.kt | 27 ++ .../nextgen/delivery/DispatcherTickTest.kt | 133 ++++++++++ .../processing/MessageProcessorTest.kt | 236 ++++++++++++++++++ 10 files changed, 525 insertions(+), 49 deletions(-) create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelinePropsTest.kt create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/DispatcherTickTest.kt create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/MessageProcessorTest.kt diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelineProps.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelineProps.kt index 4631b52..dcc8cc4 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelineProps.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelineProps.kt @@ -29,8 +29,11 @@ class PipelineProps { var backoffCapMs: Long = 60_000 var headDeadline: Duration = Duration.ofMinutes(10) // 最坏 HOL 上界（毒丸升级） - fun backoffFor(attempt: Int): Long = - backoffMs.drop(attempt - 1).firstOrNull()?.coerceAtMost(backoffCapMs) ?: backoffCapMs + /** N28：attempt ≤ 0（如 FAILED 未递增 attempts 的行）不得抛异常，取下界=首档退避。 */ + fun backoffFor(attempt: Int): Long { + val index = (attempt - 1).coerceAtLeast(0) + return backoffMs.getOrNull(index)?.coerceAtMost(backoffCapMs) ?: backoffCapMs + } } @ConfigurationProperties("schd") diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt index 1735331..442f36a 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt @@ -24,7 +24,8 @@ interface DeliveryPort { /** * ACMA-8 流程 3：投递调度——每 target 严格 FIFO（I1 双层同策略）； - * 阶段 B（定案 2）：ES 投递成功后同线程同步 enqueue REDIS:flightInfo 删除事件。 + * N03（U06）：KAFKA_SCHD 不走逐条循环（唯一出口是 flushSchd 批量聚合）， + * 逐条循环显式排除，避免 schd 事件被无条件 markSent 吞掉、flush 永远 claim 不到。 */ @Singleton class Dispatcher( @@ -39,17 +40,24 @@ class Dispatcher( fun loop() { while (running) { - tick() - Thread.sleep(200) + try { + tick() + } catch (e: Exception) { + // U08/R02：投递循环不得因单次异常静默死亡；U12 补 ERROR 日志与告警出口。 + sleepQuietly(props.pipeline.pollInterval) + } + // N18：轮询间隔取参数表（下限 50ms，避免退避节律被吞） + sleepQuietly(props.pipeline.pollInterval.coerceAtLeast(Duration.ofMillis(50))) } } internal fun tick() { val targets = if (props.phase == PipelineProps.Phase.A) Targets.phaseA else Targets.phaseB for (t in targets) { + if (t == Targets.KAFKA_SCHD) continue // N03：schd 唯一出口 flushSchd val head = msgEvents.headUnsent(t) ?: continue if (head.state == EventStatus.PENDING && head.nextAttemptAt != null && head.nextAttemptAt > Instant.now()) { - // 队头退避未到期：等待，不跳过（保序）；毒丸升级由实装补 attempts/deadline 判定 + // 队头退避未到期：等待，不跳过（保序）；超时升级归 U13（WP2） continue } try { @@ -72,15 +80,22 @@ class Dispatcher( } } } - if (Duration.between(lastFlush, Instant.now()) >= props.schd.flushPeriod) { - lastFlush = Instant.now() - flushSchd() + if (flushDue()) { + try { + flushSchd() + } catch (e: Exception) { + // U08/N24：批发送失败不得终止 loop——事件保持 PENDING，下个 flush 周期整批重试； + // 毒丸/告警随 U12/U13 补齐。lastFlush 仅在成功后推进（见 flushSchd）。 + } } } + private fun flushDue(): Boolean = + Duration.between(lastFlush, Instant.now()) >= props.schd.flushPeriod + private fun deliver(target: String, e: MsgEvent) = when (target) { Targets.KAFKA_MSG -> port.sendKafka("msg", e.payloadJson) - Targets.KAFKA_SCHD -> Unit // schd 走 flushSchd 批量路径 + Targets.KAFKA_SCHD -> error("KAFKA_SCHD must go through flushSchd (N03)") Targets.ES_FLIGHT_HTS -> port.indexFlightHts(e.payloadJson) Targets.REDIS_FLIGHT_INFO -> port.projectRedis(e.payloadJson) else -> error("unknown target $target") @@ -91,9 +106,17 @@ class Dispatcher( /** 流程 3 flushSchd：批上限 + 每 FLID 最新一态聚合（topic "schd"，wire=FLTR JSON 数组）。 */ internal fun flushSchd() { val pending = msgEvents.claimBatch(Targets.KAFKA_SCHD, limit = props.schd.flushLimit) - if (pending.isEmpty()) return + if (pending.isEmpty()) { + lastFlush = Instant.now() + return + } val payload = SchdAggregation.latestPerFlight(pending).joinToString(",", "[", "]") - port.sendKafka("schd", payload) // 失败整批退避（at-least-once） + port.sendKafka("schd", payload) // 失败整批退避（at-least-once）：异常上抛由 tick 隔离，事件保持 PENDING msgEvents.markAllSent(pending.mapNotNull { it.eventId }) + lastFlush = Instant.now() // N24：仅在成功后推进 + } + + private fun sleepQuietly(d: Duration) { + if (!d.isNegative && !d.isZero) Thread.sleep(d.toMillis().coerceAtLeast(1)) } } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/ProcState.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/ProcState.kt index 8e841d7..c27deac 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/ProcState.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/ProcState.kt @@ -5,7 +5,7 @@ package com.gzzn.omms.msgexchange.nextgen.domain */ enum class ProcStatus { PENDING, FAILED, SUCCEEDED, SKIPPED, DEAD } -enum class ErrorClass { MALFORMED, CODEC_ERROR, EXHAUSTED, INFRA } +enum class ErrorClass { MALFORMED, CODEC_ERROR, EXHAUSTED, INFRA, UNSUPPORTED } data class ProcState( val cminmsgsId: Long, diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/jobs/JobExecutor.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/jobs/JobExecutor.kt index 75075d1..7dea6e7 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/jobs/JobExecutor.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/jobs/JobExecutor.kt @@ -38,14 +38,16 @@ class HistorySweepJob( ) { fun run() { val all = redis.hgetAllFlightInfo() - val history = pickHistory(all) // TODO(阶段2): 沿用现役判史规则（KEEP） - val success = emptyList() // TODO(阶段2): esFlightHts.saveSync(history) + val history = pickHistory(all) // U10/T07：saveSync 接线前判史为空集——禁止“全量可删”默认 + val success = emptyList() // TODO(阶段2): esFlightHts.saveSync(history)（返回成功集后再接线删除） if (success.isNotEmpty()) { redis.eval(RedisScript.BATCH_DELETE, delFields = success) } } - private fun pickHistory(all: Map): Map = all // TODO(阶段2) + /** U10/T07（修订）：接入 ES success 集之前的门禁——占位默认返回空集； + * 现役五条判史规则（SODT 3 天 / CNCL 1 小时 / 备降 / 离港 / 到港）golden 通过后才允许接线。 */ + private fun pickHistory(all: Map): Map = emptyMap() // TODO(阶段2) } /** 流程 5 runJob(ARCHIVE)：3:00 归档——1 天前且仅终态可迁（矩阵 #12）。 */ diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt index f89c8f0..bc4179f 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt @@ -1,8 +1,10 @@ package com.gzzn.omms.msgexchange.nextgen.processing import com.gzzn.omms.msgexchange.nextgen.config.PipelineProps +import com.gzzn.omms.msgexchange.nextgen.domain.DecodedMessage import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent +import com.gzzn.omms.msgexchange.nextgen.domain.ProcState import com.gzzn.omms.msgexchange.nextgen.domain.ProcStatus import com.gzzn.omms.msgexchange.nextgen.domain.Targets import com.gzzn.omms.msgexchange.nextgen.infra.persistence.CminmsgInboxRepository @@ -33,7 +35,13 @@ class Pump( fun loop() { while (running) { - tick() + try { + tick() + } catch (t: Throwable) { + // U08/R02：主泵不得因单次异常静默死亡——任一 codec/DB/Redis 抖动不得终止队列； + // 具体失败落 FAILED/DEAD 与告警见 MessageProcessor/U12。 + sleepQuietly(props.pipeline.pollInterval) + } } } @@ -48,25 +56,25 @@ class Pump( head == null -> sleepQuietly(props.pipeline.pollInterval) head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > Instant.now() -> if (poisoned(head)) { - // 毒丸出队：DLQ + 告警，队列继续（I1） - procState.update( - head.cminmsgsId, ProcStatus.DEAD, - errorClass = ErrorClass.EXHAUSTED, - lastError = head.lastError ?: "head-deadline-exceeded", - ) + upgradeDead(head, head.lastError ?: "head-deadline-exceeded") } else { sleepQuietly(Duration.between(Instant.now(), head.nextAttemptAt)) } - else -> processor.processOne(head.cminmsgsId) + // PENDING、或 FAILED 退避已到期：交处理入口（入口再做 attempts 毒丸守卫） + else -> processor.processOne(head) } } - private fun poisoned(head: com.gzzn.omms.msgexchange.nextgen.domain.ProcState): Boolean = + private fun poisoned(head: ProcState): Boolean = head.attempts >= props.pipeline.maxAttempts || Duration.between(head.updatedAt, Instant.now()) > props.pipeline.headDeadline + private fun upgradeDead(head: ProcState, reason: String) { + procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.EXHAUSTED, lastError = reason) + } + /** JOB 与队头消息的先后由入队时间近似；实装以统一序号列保证（阶段 1 后续）。 */ - private fun jobBefore(job: PumpJobRepository.Job, head: com.gzzn.omms.msgexchange.nextgen.domain.ProcState?) = + private fun jobBefore(job: PumpJobRepository.Job, head: ProcState?) = head == null || head.state == ProcStatus.FAILED private fun execute(job: PumpJobRepository.Job) { @@ -84,7 +92,12 @@ class Pump( } } -/** ACMA-8 流程 2 processOne：解码→绑定→纯函数决策→提交。 */ +/** + * ACMA-8 流程 2 processOne：解码→绑定→纯函数决策→提交。 + * U08/U10/U11（N06/N21/T06）：统一失败治理——「未实装/未注册/CODEC_ERROR/staging 未实装」 + * 一律 FAILED（可重放，带 attempts+nextAttemptAt，绝不直接写终态）；仅 MALFORMED（报文非法） + * 与「重试耗尽」写 DEAD 终态。attempts ≥ maxAttempts 的 FAILED 在入口即升级 DEAD（防退避到期后无限重试）。 + */ @Singleton class MessageProcessor( private val inbox: CminmsgInboxRepository, @@ -96,28 +109,38 @@ class MessageProcessor( private val snapshotFlow: SnapshotFlow, private val props: PipelineProps, ) { - fun processOne(cminmsgsId: Long) { - val raw = inbox.rawOf(cminmsgsId) ?: run { - procState.update(cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = "raw-missing") + fun processOne(head: ProcState) { + // U08 入口守卫：FAILED 行退避到期后再处理前，先判 attempts 毒丸（与 tick poisoned 同值） + if (head.state == ProcStatus.FAILED && head.attempts >= props.pipeline.maxAttempts) { + procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.EXHAUSTED, + lastError = head.lastError ?: "max-attempts") + return + } + + val raw = inbox.rawOf(head.cminmsgsId) ?: run { + 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.nextgen.codec.DecodeResult.Ok -> r.message is com.gzzn.omms.msgexchange.nextgen.codec.DecodeResult.Err -> { - // MALFORMED 不重试；CODEC_ERROR 可一键重放（errorClass 入库） - procState.update(cminmsgsId, ProcStatus.DEAD, errorClass = r.failure.errorClass, lastError = r.failure.detail) + // T06（U11）：MALFORMED（报文非法）→ DEAD 不重试；CODEC_ERROR（可随 codec 修复重放）→ FAILED 退避 + if (r.failure.errorClass == ErrorClass.MALFORMED) { + procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, + lastError = r.failure.detail) + } else { + failWithBackoff(head, r.failure.errorClass, r.failure.detail) + } return } } - val state = procState.headUnfinished() - // I3：identity 仅首次绑定；FAILED 重试不重绑（绑定失败 = 另一条同键消息 → SKIPPED） - // TODO(阶段1后续): 以行内 IDENTITY_KEY 判定而非 head 复查（含 FAILED 持久化字段读取） - if (state != null && state.cminmsgsId == cminmsgsId && state.identityKey == null) { + // I3：identity 仅首次绑定（head.identityKey == null）；FAILED 重试不重绑（N25：入参传递 head，不再二次查询） + if (head.identityKey == null) { val identity = Identity.of(decoded, props.identity) - if (!procState.tryBindIdentity(cminmsgsId, identity)) { + if (!procState.tryBindIdentity(head.cminmsgsId, identity)) { val owner = procState.ownerOfIdentity(identity) ?: -1L - procState.update(cminmsgsId, ProcStatus.SKIPPED, lastError = "duplicate-of:$owner") + procState.update(head.cminmsgsId, ProcStatus.SKIPPED, lastError = "duplicate-of:$owner") return } } @@ -125,14 +148,15 @@ class MessageProcessor( // 快照消息走专属流程（流程 4：staging→Lua→putGenIfVersion CAS 同事务） val handler = handlers.registry.dispatcherFor(decoded) if (handler == null) { - procState.update(cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = "no-handler:${decoded.typeTag}") + // U10（N21）：未注册 ≠ 报文非法——写 FAILED（可重放），绝不写终态 + failWithBackoff(head, ErrorClass.UNSUPPORTED, "no-handler:${decoded.typeTag}") return } val schdKind = decoded.kind as? com.gzzn.omms.msgexchange.nextgen.domain.MsgKind.Schd if (schdKind != null && schdKind.subtype == com.gzzn.omms.msgexchange.nextgen.domain.MsgKind.SchdSubtype.DNLD ) { - snapshotFlow.publishSnapshot(cminmsgsId, decoded) + snapshotFlow.publishSnapshot(head, decoded) return } @@ -147,10 +171,22 @@ class MessageProcessor( decision.schdPush.forEach { add(MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = it.flid, payloadJson = it.fltrJson)) } } msgEvents.insertAll(events) - inbox.backfillOnSuccess(cminmsgsId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn) - procState.update(cminmsgsId, ProcStatus.SUCCEEDED) + inbox.backfillOnSuccess(head.cminmsgsId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn) + procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED) // 阶段 B：flightState.apply(decision.flightChanges) 进入同一事务；投影事件（ES/REDIS）追加。 } + + /** 统一失败路径：FAILED + attempts+1 + nextAttemptAt=backoff（N28 下界由 backoffFor 保证）。 */ + private fun failWithBackoff(head: ProcState, ec: ErrorClass, reason: String) { + val attempts = head.attempts + 1 + procState.update( + head.cminmsgsId, ProcStatus.FAILED, + attempts = attempts, + nextAttemptAt = Instant.now().plusMillis(props.pipeline.backoffFor(attempts)), + errorClass = ec, + lastError = reason, + ) + } } /** 延迟装配占位（阶段 1 后续以 Micronaut Bean 替换直连构造）。 */ diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/SnapshotFlow.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/SnapshotFlow.kt index 92fb6fc..a31595e 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/SnapshotFlow.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/SnapshotFlow.kt @@ -1,31 +1,36 @@ package com.gzzn.omms.msgexchange.nextgen.processing +import com.gzzn.omms.msgexchange.nextgen.config.PipelineProps import com.gzzn.omms.msgexchange.nextgen.domain.DecodedMessage import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass +import com.gzzn.omms.msgexchange.nextgen.domain.ProcState import com.gzzn.omms.msgexchange.nextgen.domain.ProcStatus import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ProcStateRepository import com.gzzn.omms.msgexchange.nextgen.infra.persistence.RefDataRepository import com.gzzn.omms.msgexchange.nextgen.infra.redis.FlightRedisClient import com.gzzn.omms.msgexchange.nextgen.infra.redis.RedisScript import jakarta.inject.Singleton +import java.time.Instant /** * ACMA-8 流程 4：日计划快照（generation，主泵内执行）。 * staging（内存瞬态，崩溃从 raw 整包重放）→ Lua 原子“覆盖+按代差删”（I4/I5） - * → putGenIfVersion CAS + SUCCEEDED 同一 MySQL 事务（重放幂等，版本不二次自增）。 - * 整包失败 = 现役等价（KEEP）；跳坏行进 quarantine = CONFIRM（矩阵 #9）。 + * → putGenIfVersion CAS + SUCCEEDED（重放幂等，版本不二次自增——恢复协议细节见 U09）。 + * U10/T07 占位安全化：staging 未实装 → FAILED(UNSUPPORTED)+退避（可重放），绝不写 DEAD 终态； + * CAS 冲突 → FAILED(INFRA)+退避（不再无 nextAttemptAt 紧循环，N06/N28）。 */ @Singleton class SnapshotFlow( private val procState: ProcStateRepository, private val refData: RefDataRepository, private val redis: FlightRedisClient, + private val props: PipelineProps, ) { - fun publishSnapshot(cminmsgsId: Long, msg: DecodedMessage) { + 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) { - procState.update(cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = staged.reason) + failWithBackoff(head, ErrorClass.UNSUPPORTED, staged.reason) // U10：未实装 → 可重放，非终态 return } val normalized = (staged as StageResult.Ok).flights // (flid, payloadJson) @@ -43,11 +48,22 @@ class SnapshotFlow( // 重放路径：version 已是目标值 → no-op 视为成功 val again = refData.getGen(day) if (again == null || again.version != expected + 1) { - procState.update(cminmsgsId, ProcStatus.FAILED, lastError = "gen-cas-conflict") + failWithBackoff(head, ErrorClass.INFRA, "gen-cas-conflict") // N06/N28：带退避，禁止紧循环 return } } - procState.update(cminmsgsId, ProcStatus.SUCCEEDED) + procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED) + } + + private fun failWithBackoff(head: ProcState, ec: ErrorClass, reason: String) { + val attempts = head.attempts + 1 + procState.update( + head.cminmsgsId, ProcStatus.FAILED, + attempts = attempts, + nextAttemptAt = Instant.now().plusMillis(props.pipeline.backoffFor(attempts)), + errorClass = ec, + lastError = reason, + ) } /** staging 结果（骨架）。 */ diff --git a/src/main/resources/db/migration/V2.0.0__aux_tables.sql b/src/main/resources/db/migration/V2.0.0__aux_tables.sql index 9319cbe..5bb45bd 100644 --- a/src/main/resources/db/migration/V2.0.0__aux_tables.sql +++ b/src/main/resources/db/migration/V2.0.0__aux_tables.sql @@ -15,7 +15,7 @@ CREATE TABLE PROC_STATE ( IDENTITY_KEY VARCHAR(200) NULL, -- SNDR|TYPE|STYP|SEQN，decode 后首次绑定（I3） ATTEMPTS INT NOT NULL DEFAULT 0, NEXT_ATTEMPT_AT TIMESTAMP NULL, - ERROR_CLASS VARCHAR(20) NULL, -- MALFORMED/CODEC_ERROR/EXHAUSTED/INFRA + ERROR_CLASS VARCHAR(20) NULL, -- MALFORMED/CODEC_ERROR/EXHAUSTED/INFRA/UNSUPPORTED LAST_ERROR VARCHAR(1000) NULL, UPDATED_AT TIMESTAMP NOT NULL, UNIQUE KEY UK_PROC_IDENTITY (IDENTITY_KEY), diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelinePropsTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelinePropsTest.kt new file mode 100644 index 0000000..5be2651 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelinePropsTest.kt @@ -0,0 +1,27 @@ +package com.gzzn.omms.msgexchange.nextgen.config + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test + +/** + * U08/N28：backoffFor 必须对 attempt ≤ 0（FAILED 行未递增 attempts）给出首档退避而非抛异常。 + */ +class PipelinePropsTest { + + private val pipeline = PipelineProps().pipeline + + @Test + fun `backoff follows table then caps`() { + assertEquals(1000, pipeline.backoffFor(1)) + assertEquals(2000, pipeline.backoffFor(2)) + assertEquals(16000, pipeline.backoffFor(5)) + assertEquals(60_000, pipeline.backoffFor(6)) // 表外 → 封顶 + assertEquals(60_000, pipeline.backoffFor(99)) + } + + @Test + fun `non-positive attempt never throws and falls back to first slot`() { + assertEquals(1000, pipeline.backoffFor(0)) + assertEquals(1000, pipeline.backoffFor(-1)) + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/DispatcherTickTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/DispatcherTickTest.kt new file mode 100644 index 0000000..42e0bbe --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/DispatcherTickTest.kt @@ -0,0 +1,133 @@ +package com.gzzn.omms.msgexchange.nextgen.delivery + +import com.gzzn.omms.msgexchange.nextgen.config.PipelineProps +import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass +import com.gzzn.omms.msgexchange.nextgen.domain.EventStatus +import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent +import com.gzzn.omms.msgexchange.nextgen.domain.Targets +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.MsgEventRepository +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import java.time.Duration +import java.time.Instant +import kotlin.test.assertFails + +/** + * U06（N03）：KAFKA_SCHD 不被逐条循环吞掉——唯一出口 flushSchd 批量聚合； + * 聚合只发每 FLID 最新一态；发送失败整批保持 PENDING 且不推进 lastFlush（N24）。 + */ +class DispatcherTickTest { + + private class FakeRepo : MsgEventRepository { + val events = mutableListOf() + + fun enqueue(e: MsgEvent) { events += e } + + override fun insertAll(events: List) = events.map { it.eventId ?: 0L } + + override fun headUnsent(target: String): MsgEvent? = + events.filter { it.target == target && it.state != EventStatus.SENT && it.state != EventStatus.DEAD } + .minByOrNull { it.eventId ?: Long.MAX_VALUE } + + override fun claimBatch(target: String, limit: Int): List = + events.filter { it.target == target && it.state == EventStatus.PENDING } + .sortedBy { it.eventId ?: Long.MAX_VALUE } + .take(limit) + + override fun markSent(eventId: Long) { + val i = events.indexOfFirst { it.eventId == eventId } + if (i >= 0) events[i] = events[i].copy(state = EventStatus.SENT) + } + + override fun markAllSent(eventIds: List) = eventIds.forEach(::markSent) + + override fun scheduleRetry(eventId: Long, nextAttemptAt: Instant, attempts: Int) { + val i = events.indexOfFirst { it.eventId == eventId } + if (i >= 0) events[i] = events[i].copy(state = EventStatus.PENDING, attempts = attempts, nextAttemptAt = nextAttemptAt) + } + + override fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String) { + val i = events.indexOfFirst { it.eventId == eventId } + if (i >= 0) events[i] = events[i].copy(state = EventStatus.DEAD, errorClass = errorClass, lastError = lastError) + } + + override fun insertSync(events: List) = Unit + } + + private class FakePort : DeliveryPort { + val sent = mutableListOf>() + var failTopic: String? = null + + override fun sendKafka(topic: String, payloadJson: String) { + if (failTopic == topic) throw RuntimeException("send-fail:$topic") + sent += topic to payloadJson + } + + override fun indexFlightHts(payloadJson: String) = Unit + override fun projectRedis(payloadJson: String) = Unit + } + + private fun props(): PipelineProps = PipelineProps() // schd.flush-period 默认 3s + + private fun ev(id: Long, target: String, key: String?, payload: String) = + MsgEvent(eventId = id, target = target, partitionKey = key, payloadJson = payload, state = EventStatus.PENDING) + + @Test + fun `schd flushed once via batch path, per-target tick never consumes it`() { + val repo = FakeRepo() + val port = FakePort() + val d = Dispatcher(repo, port, props()) + repo.enqueue(ev(1, Targets.KAFKA_MSG, null, """{"msg":1}""")) + repo.enqueue(ev(2, Targets.KAFKA_SCHD, "F1", """{"FLID":"F1","v":1}""")) + + d.tick() // lastFlush=EPOCH → 首 tick 即到 flush 周期 + + // KAFKA:msg 逐条投递；schd 经 flushSchd 聚合发出一次（而非逐条 markSent 吞掉） + assertEquals(1, port.sent.count { it.first == "msg" }) + assertEquals(1, port.sent.count { it.first == "schd" }) + assertEquals(EventStatus.SENT, repo.events.first { it.eventId == 2L }.state) + + // 新的 schd 事件：flush 周期未到 → 逐条循环不得消费它（仍 PENDING、未发） + repo.enqueue(ev(4, Targets.KAFKA_SCHD, "F1", """{"FLID":"F1","v":4}""")) + d.tick() + assertEquals(EventStatus.PENDING, repo.events.first { it.eventId == 4L }.state) + assertEquals(1, port.sent.count { it.first == "schd" }) // 第二次 tick 未额外发送 + } + + @Test + fun `flushSchd aggregates latest per flight and marks all sent`() { + val repo = FakeRepo() + val port = FakePort() + val d = Dispatcher(repo, port, props()) + repo.enqueue(ev(1, Targets.KAFKA_SCHD, "F1", "old")) + repo.enqueue(ev(2, Targets.KAFKA_SCHD, "F2", "only")) + repo.enqueue(ev(3, Targets.KAFKA_SCHD, "F1", "new")) + + d.flushSchd() + + assertEquals(1, port.sent.count { it.first == "schd" }) + val payload = port.sent.first { it.first == "schd" }.second + assertTrue(payload.startsWith("[") && payload.endsWith("]")) + assertTrue(payload.contains("new") && payload.contains("only") && !payload.contains("old")) + assertEquals(2, payload.removeSurrounding("[", "]").split(",").size) // 同 FLID 只发最新（矩阵 #7） + assertTrue(repo.events.all { it.state == EventStatus.SENT }) + } + + @Test + fun `flush failure keeps batch PENDING and retries after next due`() { + val repo = FakeRepo() + val port = FakePort().apply { failTopic = "schd" } + val d = Dispatcher(repo, port, props()) + repo.enqueue(ev(1, Targets.KAFKA_SCHD, "F1", """{"v":"1"}""")) + + assertFails { d.flushSchd() } // 发送失败 → 异常上抛（tick 侧隔离），事件保持 PENDING + assertEquals(EventStatus.PENDING, repo.events.single().state) + assertEquals(0, port.sent.size) + + port.failTopic = null + d.flushSchd() // lastFlush 未推进 → 下个周期整批重试成功 + assertEquals(EventStatus.SENT, repo.events.single().state) + assertEquals(1, port.sent.count { it.first == "schd" }) + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/MessageProcessorTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/MessageProcessorTest.kt new file mode 100644 index 0000000..8b0d26d --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/MessageProcessorTest.kt @@ -0,0 +1,236 @@ +package com.gzzn.omms.msgexchange.nextgen.processing + +import com.gzzn.omms.msgexchange.nextgen.codec.DecodeFailure +import com.gzzn.omms.msgexchange.nextgen.codec.DecodeResult +import com.gzzn.omms.msgexchange.nextgen.codec.XmlCodec +import com.gzzn.omms.msgexchange.nextgen.config.PipelineProps +import com.gzzn.omms.msgexchange.nextgen.domain.DecodedMessage +import com.gzzn.omms.msgexchange.nextgen.domain.Decision +import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass +import com.gzzn.omms.msgexchange.nextgen.domain.EventStatus +import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent +import com.gzzn.omms.msgexchange.nextgen.domain.MsgKind +import com.gzzn.omms.msgexchange.nextgen.domain.MetaFields +import com.gzzn.omms.msgexchange.nextgen.domain.NotifyPayload +import com.gzzn.omms.msgexchange.nextgen.domain.ProcState +import com.gzzn.omms.msgexchange.nextgen.domain.ProcStatus +import com.gzzn.omms.msgexchange.nextgen.domain.SchdPush +import com.gzzn.omms.msgexchange.nextgen.domain.Targets +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.CminmsgInboxRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.MsgEventRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ProcStateRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.PumpJobRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.RefDataRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ReqTrackRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.FlightStateRepository +import com.gzzn.omms.msgexchange.nextgen.infra.redis.FlightRedisClient +import com.gzzn.omms.msgexchange.nextgen.infra.redis.RedisScript +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import java.time.Instant + +/** + * U10/U11/U08（N21/T06/N06/N28）：processOne 失败语义—— + * 未注册 handler / CODEC_ERROR / staging 未实装 → FAILED（可重放，带退避，绝不写终态）； + * MALFORMED → DEAD（不重试）；FAILED attempts≥maxAttempts 入口升级 DEAD；成功路径事件+回填+SUCCEEDED。 + */ +class MessageProcessorTest { + + // ---------- fakes ---------- + private class FakeProcState : ProcStateRepository { + var record = mutableMapOf() + val bound = mutableMapOf() // identityKey -> owner + + override fun insert(cminmsgsId: Long, state: ProcStatus) { record[cminmsgsId] = ProcState(cminmsgsId, state) } + + override fun headUnfinished(): ProcState? = null + + override fun tryBindIdentity(cminmsgsId: Long, identityKey: String): Boolean { + val owner = bound[identityKey] + if (owner != null && owner != cminmsgsId) return false + bound[identityKey] = cminmsgsId + record[cminmsgsId] = record[cminmsgsId]?.copy(identityKey = identityKey) ?: ProcState(cminmsgsId, ProcStatus.PENDING, identityKey = identityKey) + return true + } + + override fun ownerOfIdentity(identityKey: String): Long? = bound[identityKey] + + override fun update( + cminmsgsId: Long, state: ProcStatus, nextAttemptAt: Instant?, attempts: Int?, + errorClass: ErrorClass?, lastError: String?, + ) { + val old = record[cminmsgsId] ?: ProcState(cminmsgsId, state) + record[cminmsgsId] = old.copy( + state = state, + nextAttemptAt = nextAttemptAt ?: old.nextAttemptAt, + attempts = attempts ?: old.attempts, + errorClass = errorClass ?: old.errorClass, + lastError = lastError ?: old.lastError, + ) + } + + fun state(id: Long): ProcState = record.getValue(id) + } + + private class FakeInbox : CminmsgInboxRepository { + var raws = mutableMapOf() + val backfilled = mutableListOf>() + + override fun insertRaw(rawXml: String): Long = 0 + + override fun rawOf(cminmsgsId: Long): String? = raws[cminmsgsId] + + override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) { + backfilled += listOf(cminmsgsId, sndr, type, styp, seqn) + } + } + + private class FakeEvents : MsgEventRepository { + val inserted = mutableListOf>() + + override fun insertAll(events: List): List { inserted += events; return events.indices.map { it.toLong() + 1 } } + + override fun headUnsent(target: String): MsgEvent? = null + override fun claimBatch(target: String, limit: Int): List = emptyList() + override fun markSent(eventId: Long) = Unit + override fun markAllSent(eventIds: List) = Unit + override fun scheduleRetry(eventId: Long, nextAttemptAt: Instant, attempts: Int) = Unit + override fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String) = Unit + override fun insertSync(events: List) = Unit + } + + private class FakeCodec : XmlCodec { + var result: DecodeResult = DecodeResult.Ok( + DecodedMessage( + meta = MetaFields(sndr = "AODB", type = "FLOP", styp = "DELY", seqn = 1L, dttm = 20260906120000L), + kind = MsgKind.Flop("DELY"), + rawXml = "", + ), + ) + override fun decode(rawXml: String): DecodeResult = result + override fun encodeRqrd(kind: String, rangeJson: String): String = "" + } + + private object FakeRedis : FlightRedisClient { + override fun eval(script: RedisScript, setPairs: List>, delFields: List) = Unit + override fun hgetAllFlightInfo(): Map = emptyMap() + } + + private class FakeRefData : RefDataRepository { + override fun getGen(day: String): RefDataRepository.GenMeta? = null + override fun putGenIfVersion(day: String, expected: Long, new: RefDataRepository.GenMeta): Boolean = true + override fun upsertAll(rows: List>) = Unit + } + + // ---------- helpers ---------- + private fun msg(seqn: String) = DecodedMessage( + meta = MetaFields(sndr = "AODB", type = "FLOP", styp = "DELY", seqn = seqn.toLong(), dttm = 20260906120000L), + kind = MsgKind.Flop("DELY"), + rawXml = "", + ) + + private fun head(id: Long = 1, status: ProcStatus = ProcStatus.PENDING, attempts: Int = 0, identityKey: String? = null) = + ProcState(id, status, identityKey = identityKey, attempts = attempts) + + private fun deliveHandler() = object : Handler { + override val kind: MsgKind = MsgKind.Flop("DELY") + override fun decide(flightView: Map, msg: DecodedMessage): Decision = Decision( + msgNotifies = listOf(NotifyPayload("""{"n":1}""")), + schdPush = listOf(SchdPush(flid = "F1", fltrJson = """{"FLID":"F1","v":2}""")), + ) + } + + private fun processor( + procState: FakeProcState, inbox: FakeInbox, events: FakeEvents, codec: FakeCodec, + registry: HandlerRegistry = HandlerRegistry(emptyList()), + ): MessageProcessor { + val props = PipelineProps() + val snapshot = SnapshotFlow(procState, FakeRefData(), FakeRedis, props) + return MessageProcessor(inbox, procState, events, CodecHolder(codec), HandlerHolder(registry), FakeRedis, snapshot, props) + } + + // ---------- tests ---------- + @Test + fun `no handler is FAILED UNSUPPORTED with backoff - never terminal`() { + val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents() + ps.record[1] = head(); inbox.raws[1] = "" + processor(ps, inbox, ev, FakeCodec()).processOne(ps.state(1)) + + val s = ps.state(1) + assertEquals(ProcStatus.FAILED, s.state) + assertEquals(ErrorClass.UNSUPPORTED, s.errorClass) + assertEquals(1, s.attempts) + assertNotNull(s.nextAttemptAt) + assertTrue(s.lastError?.startsWith("no-handler:FLOP-DELY") == true) + } + + @Test + fun `CODEC_ERROR decode failure is FAILED retryable`() { + val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents() + ps.record[1] = head(); inbox.raws[1] = "" + val codec = FakeCodec().apply { result = DecodeResult.Err(DecodeFailure(ErrorClass.CODEC_ERROR, "boom")) } + processor(ps, inbox, ev, codec).processOne(ps.state(1)) + + val s = ps.state(1) + assertEquals(ProcStatus.FAILED, s.state) + assertEquals(ErrorClass.CODEC_ERROR, s.errorClass) + assertNotNull(s.nextAttemptAt) + } + + @Test + fun `MALFORMED decode failure stays DEAD terminal`() { + val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents() + ps.record[1] = head(); inbox.raws[1] = "" + val codec = FakeCodec().apply { result = DecodeResult.Err(DecodeFailure(ErrorClass.MALFORMED, "bad-xml")) } + processor(ps, inbox, ev, codec).processOne(ps.state(1)) + + val s = ps.state(1) + assertEquals(ProcStatus.DEAD, s.state) + assertEquals(ErrorClass.MALFORMED, s.errorClass) + } + + @Test + fun `failed head at max attempts upgrades to DEAD at entry`() { + val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents() + ps.record[1] = head(status = ProcStatus.FAILED, attempts = 5, identityKey = "k") + processor(ps, inbox, ev, FakeCodec()).processOne(ps.state(1)) + + val s = ps.state(1) + assertEquals(ProcStatus.DEAD, s.state) + assertEquals(ErrorClass.EXHAUSTED, s.errorClass) + assertTrue(inbox.raws.isEmpty() || inbox.backfilled.isEmpty()) // 入口即升级，未进入处理 + } + + @Test + fun `success path inserts events backfills and marks SUCCEEDED`() { + val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents() + ps.record[1] = head(); inbox.raws[1] = "" + val registry = HandlerRegistry(listOf(deliveHandler())) + processor(ps, inbox, ev, FakeCodec(), registry).processOne(ps.state(1)) + + assertEquals(ProcStatus.SUCCEEDED, ps.state(1).state) + assertEquals(1, ev.inserted.size) + val events = ev.inserted.single() + assertEquals(listOf(Targets.KAFKA_MSG, Targets.KAFKA_SCHD), events.map { it.target }) + assertEquals("F1", events.first { it.target == Targets.KAFKA_SCHD }.partitionKey) + assertEquals(listOf(1L, "AODB", "FLOP", "DELY", 1L), inbox.backfilled.single()) + } + + @Test + fun `duplicate identity leads to SKIPPED`() { + val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents() + ps.record[1] = head(); ps.record[2] = head(id = 2) + inbox.raws[1] = "" + ps.bound["AODB|FLOP|DELY|1"] = 2L // 另一条消息已持有该键 + + processor(ps, inbox, ev, FakeCodec()).processOne(ps.state(1)) + + val s = ps.state(1) + assertEquals(ProcStatus.SKIPPED, s.state) + assertTrue(s.lastError == "duplicate-of:2") + assertTrue(ev.inserted.isEmpty()) + } +}