From 73113779d2007c559af98f49af6da215843a3864 Mon Sep 17 00:00:00 2001 From: windyboy Date: Sun, 6 Sep 2026 19:08:02 +0800 Subject: [PATCH] =?UTF-8?q?fix(processing):=20U08=20=E9=87=8D=E8=AF=95?= =?UTF-8?q?=E9=97=AD=E7=8E=AF=20+=20FailureScheduler=20+=20U11=20=E9=87=8D?= =?UTF-8?q?=E6=94=BE=E5=85=A5=E5=8F=A3=EF=BC=88ACM2-10=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - U08 闭环：失败状态迁移移到持有 head/事件的边界内——processOne/SnapshotFlow 统一经 ProcFailure（FAILED+attempts+退避，达上限 DEAD/EXHAUSTED 并落库 attempts）；Dispatcher 逐条与 SCHD 批量共用 retryOrDead；flushSchd 失败整批 attempts+1/指数退避，队首未到期不 claim，达上限整批 DEAD(DLQ)；loop 只作最后防线——仅 catch Exception，InterruptedException 恢复中断位并上抛，致命 Error 不吞（异常必可见） - 统一策略：新增 FailureScheduler（可注入 java.time.Clock + TimeFactory，backoff/exhausted 单一来源）与 ProcFailure，MessageProcessor/SnapshotFlow/Dispatcher 共用 - U11：ReplayService 显式重放入口 + ProcStateRepository.requeueByErrorClasses——仅白名单内 可恢复类（CODEC_ERROR/UNSUPPORTED/INFRA/EXHAUSTED）从 FAILED/DEAD 回 PENDING(attempts 清零)， MALFORMED 永不重放 - 测试（30 个全绿）：①Pump 内部异常→FAILED(INFRA)→重试达上限 DEAD；②SCHD 连续失败退避递增→ 整批 DEAD 且 KAFKA:msg 不受影响；③AssertionError/InterruptedException 不被普通恢复吞掉； 另有 ReplayService(3)、Dispatcher 批退避、InfraBindingStartupTest（DataSource 启动级绑定） - 收尾：.gitattributes（行尾/二进制）；Gradle 10 弃用告警核查——仅来自 micronaut-application 插件（BOM 注入 + IDE 文件生成），升级 Gradle 10 前需先升插件版本（已记录） --- .gitattributes | 13 ++ .../nextgen/delivery/Dispatcher.kt | 69 ++++++---- .../nextgen/infra/persistence/Repositories.kt | 9 +- .../nextgen/infra/retry/FailureScheduler.kt | 36 ++++++ .../nextgen/infra/retry/ProcFailure.kt | 38 ++++++ .../nextgen/infra/retry/ReplayService.kt | 28 +++++ .../msgexchange/nextgen/processing/Pump.kt | 58 ++++----- .../nextgen/processing/SnapshotFlow.kt | 24 +--- .../omms/msgexchange/nextgen/TestClocks.kt | 27 ++++ .../nextgen/config/InfraBindingStartupTest.kt | 39 ++++++ .../nextgen/delivery/DispatcherTickTest.kt | 91 +++++++++++--- .../nextgen/infra/retry/ReplayServiceTest.kt | 94 ++++++++++++++ .../processing/MessageProcessorTest.kt | 118 +++++++++++++++--- 13 files changed, 535 insertions(+), 109 deletions(-) create mode 100644 .gitattributes create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/FailureScheduler.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ProcFailure.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ReplayService.kt create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/TestClocks.kt create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/config/InfraBindingStartupTest.kt create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ReplayServiceTest.kt diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..34f8179 --- /dev/null +++ b/.gitattributes @@ -0,0 +1,13 @@ +# 行尾/二进制规范（Gradle 10 前收尾项，ACM2-10 U02/U30） +* text=auto + +# shell 启动脚本必须 LF（Windows 上保留可执行） +*.sh text eol=lf +gradlew text eol=lf +*.bat text eol=crlf + +# 二进制一律不转行尾 +*.jar binary +*.png binary +*.jpg binary +*.pdf binary 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 442f36a..41c30e2 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 @@ -6,6 +6,7 @@ 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 com.gzzn.omms.msgexchange.nextgen.infra.retry.FailureScheduler import jakarta.inject.Singleton import java.time.Duration import java.time.Instant @@ -23,15 +24,19 @@ interface DeliveryPort { } /** - * ACMA-8 流程 3：投递调度——每 target 严格 FIFO（I1 双层同策略）； - * N03（U06）：KAFKA_SCHD 不走逐条循环（唯一出口是 flushSchd 批量聚合）， - * 逐条循环显式排除，避免 schd 事件被无条件 markSent 吞掉、flush 永远 claim 不到。 + * ACMA-8 流程 3：投递调度——每 target 严格 FIFO（I1 双层同策略）。 + * N03（U06）：KAFKA_SCHD 不走逐条循环（唯一出口是 flushSchd 批量聚合），逐条循环显式排除。 + * U08 闭环：逐条与批量失败都在持有具体事件/批次的边界完成状态迁移—— + * · 逐条：scheduleRetry(attempts+1, backoff) / 达上限 markDead(EXHAUSTED)（DLQ 保留行）； + * · 批量（flushSchd）：整批退避，队首 nextAttemptAt 未到不 claim；达上限整批 DEAD/DLQ； + * · loop 只作最后防线，不吞 InterruptedException（致命/中断错误不被普通恢复吞掉）。 */ @Singleton class Dispatcher( private val msgEvents: MsgEventRepository, private val port: DeliveryPort, private val props: PipelineProps, + private val scheduler: FailureScheduler, ) { @Volatile private var running = true @@ -42,8 +47,11 @@ class Dispatcher( while (running) { try { tick() + } catch (e: InterruptedException) { + Thread.currentThread().interrupt() + return } catch (e: Exception) { - // U08/R02：投递循环不得因单次异常静默死亡；U12 补 ERROR 日志与告警出口。 + // U08：投递循环只作最后防线；致命 Error 不吞。tick 内失败迁移已完成。 sleepQuietly(props.pipeline.pollInterval) } // N18：轮询间隔取参数表（下限 50ms，避免退避节律被吞） @@ -56,7 +64,7 @@ class Dispatcher( 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()) { + if (head.state == EventStatus.PENDING && head.nextAttemptAt != null && head.nextAttemptAt > scheduler.now()) { // 队头退避未到期：等待，不跳过（保序）；超时升级归 U13（WP2） continue } @@ -68,30 +76,26 @@ class Dispatcher( msgEvents.insertSync(listOf(MsgEvent(target = Targets.REDIS_FLIGHT_INFO, payloadJson = deleteOf(head)))) } } catch (e: Exception) { - val attempts = head.attempts + 1 - if (attempts >= props.pipeline.maxAttempts) { - msgEvents.markDead(head.eventId!!, ErrorClass.EXHAUSTED, e.message ?: "unknown") - } else { - msgEvents.scheduleRetry( - head.eventId!!, - Instant.now().plusMillis(props.pipeline.backoffFor(attempts)), - attempts, - ) - } + retryOrDead(head, e.message ?: "unknown") } } if (flushDue()) { - try { - flushSchd() - } catch (e: Exception) { - // U08/N24：批发送失败不得终止 loop——事件保持 PENDING，下个 flush 周期整批重试； - // 毒丸/告警随 U12/U13 补齐。lastFlush 仅在成功后推进（见 flushSchd）。 - } + flushSchd() } } private fun flushDue(): Boolean = - Duration.between(lastFlush, Instant.now()) >= props.schd.flushPeriod + Duration.between(lastFlush, scheduler.now()) >= props.schd.flushPeriod + + /** 单条事件失败迁移：attempts+1；达上限 DEAD(EXHAUSTED)（DLQ，attempts 落库审计），否则退避重试。 */ + private fun retryOrDead(e: MsgEvent, lastError: String) { + val attempts = e.attempts + 1 + if (scheduler.exhausted(attempts)) { + msgEvents.markDead(e.eventId!!, ErrorClass.EXHAUSTED, lastError, attempts) + } else { + msgEvents.scheduleRetry(e.eventId!!, scheduler.nextAttemptAt(attempts), attempts) + } + } private fun deliver(target: String, e: MsgEvent) = when (target) { Targets.KAFKA_MSG -> port.sendKafka("msg", e.payloadJson) @@ -103,17 +107,30 @@ class Dispatcher( private fun deleteOf(e: MsgEvent) = """{"op":"delete","refs":${e.partitionKey ?: ""}}""" - /** 流程 3 flushSchd：批上限 + 每 FLID 最新一态聚合（topic "schd"，wire=FLTR JSON 数组）。 */ + /** + * 流程 3 flushSchd：批上限 + 每 FLID 最新一态聚合（topic "schd"，wire=FLTR JSON 数组）。 + * U08 批量闭环：队首退避未到期不 claim；发送失败 → 整批 attempts+1（退避）或达上限整批 DEAD/DLQ； + * lastFlush 仅在成功（含空批）后推进。 + */ internal fun flushSchd() { val pending = msgEvents.claimBatch(Targets.KAFKA_SCHD, limit = props.schd.flushLimit) if (pending.isEmpty()) { - lastFlush = Instant.now() + lastFlush = scheduler.now() return } + val due = pending.first() + if (due.nextAttemptAt != null && due.nextAttemptAt > scheduler.now()) { + return // 队首仍在退避：整批等待，不推进 lastFlush（到期再试） + } val payload = SchdAggregation.latestPerFlight(pending).joinToString(",", "[", "]") - port.sendKafka("schd", payload) // 失败整批退避（at-least-once）：异常上抛由 tick 隔离，事件保持 PENDING + try { + port.sendKafka("schd", payload) + } catch (e: Exception) { + pending.forEach { retryOrDead(it, "schd-send: ${e.message ?: "unknown"}") } + return // 不推进 lastFlush：整批退避（含 DEAD 出队）后到期重试 + } msgEvents.markAllSent(pending.mapNotNull { it.eventId }) - lastFlush = Instant.now() // N24：仅在成功后推进 + lastFlush = scheduler.now() } private fun sleepQuietly(d: Duration) { diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt index a6a06d9..fd27d08 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt @@ -29,6 +29,13 @@ interface ProcStateRepository { errorClass: ErrorClass? = null, lastError: String? = null, ) + + /** + * U11 显式重放入口（ReplayService）：仅把给定 errorClass 集合中的行从 FAILED/DEAD 置回 PENDING， + * 以便主泵重新领取。实现约定：ATTEMPTS=0、NEXT_ATTEMPT_AT=NULL（立即重试）， + * ERROR_CLASS/LAST_ERROR 保留作审计。返回受影响行数。 + */ + fun requeueByErrorClasses(errorClasses: List): Int } interface MsgEventRepository { @@ -45,7 +52,7 @@ interface MsgEventRepository { fun scheduleRetry(eventId: Long, nextAttemptAt: Instant, attempts: Int) - fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String) + fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String, attempts: Int? = null) /** 阶段 B（定案 2）：Delivery 同线程在 ES 投递成功后同步 enqueue 删除事件。 */ fun insertSync(events: List) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/FailureScheduler.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/FailureScheduler.kt new file mode 100644 index 0000000..742345e --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/FailureScheduler.kt @@ -0,0 +1,36 @@ +package com.gzzn.omms.msgexchange.nextgen.infra.retry + +import com.gzzn.omms.msgexchange.nextgen.config.PipelineProps +import io.micronaut.context.annotation.Factory +import jakarta.inject.Singleton +import java.time.Clock +import java.time.Instant + +/** + * 统一重试策略（ACM2-10 U08）：供 MessageProcessor / SnapshotFlow（ProcState 侧）与 + * Dispatcher（MsgEvent 侧）共用——attempts 递增后按 backoff 表给 nextAttemptAt； + * exhausted 判定与两侧同源（maxAttempts）。时间一律经可注入 Clock（测试用固定钟，避免脆弱睡眠）。 + */ +@Singleton +class FailureScheduler( + private val props: PipelineProps, + private val clock: Clock, +) { + /** 便捷构造：默认系统时钟（生产路径）。 */ + constructor(props: PipelineProps) : this(props, Clock.systemUTC()) + + fun now(): Instant = clock.instant() + + fun exhausted(attempts: Int): Boolean = attempts >= props.pipeline.maxAttempts + + /** attempts 指递增后的值；N28：attempt ≤ 0 由 backoffFor 兜底为首档。 */ + fun nextAttemptAt(attemptsAfterIncrement: Int): Instant = + now().plusMillis(props.pipeline.backoffFor(attemptsAfterIncrement)) +} + +/** 提供可注入 Clock（java.time.Clock）；测试可用 Clock.fixed(...) 或自定义可变钟覆盖。 */ +@Factory +class TimeFactory { + @Singleton + fun systemClock(): Clock = Clock.systemUTC() +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ProcFailure.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ProcFailure.kt new file mode 100644 index 0000000..456af55 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ProcFailure.kt @@ -0,0 +1,38 @@ +package com.gzzn.omms.msgexchange.nextgen.infra.retry + +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 jakarta.inject.Singleton + +/** + * ProcState 侧统一失败迁移（U08/U10）：处理/快照路径共用—— + * attempts+1 后若 exhausted → DEAD(EXHAUSTED)（终态，errorClass 规范化，原因保留在 lastError）； + * 否则 FAILED + attempts + nextAttemptAt(退避)（可重放）。任何“失败”都不得在无退避下直接终态化。 + */ +@Singleton +class ProcFailure( + private val procState: ProcStateRepository, + val scheduler: FailureScheduler, +) { + fun fail(head: ProcState, ec: ErrorClass, reason: String) { + val attempts = head.attempts + 1 + if (scheduler.exhausted(attempts)) { + procState.update( + head.cminmsgsId, ProcStatus.DEAD, + attempts = attempts, + errorClass = ErrorClass.EXHAUSTED, + lastError = "$reason; attempts=$attempts", + ) + } else { + procState.update( + head.cminmsgsId, ProcStatus.FAILED, + attempts = attempts, + nextAttemptAt = scheduler.nextAttemptAt(attempts), + errorClass = ec, + lastError = reason, + ) + } + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ReplayService.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ReplayService.kt new file mode 100644 index 0000000..aec8e9d --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ReplayService.kt @@ -0,0 +1,28 @@ +package com.gzzn.omms.msgexchange.nextgen.infra.retry + +import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ProcStateRepository +import jakarta.inject.Singleton + +/** + * U11 显式重放入口：只允许“可恢复”的错误类从 FAILED/DEAD 回到 PENDING（主泵重领）。 + * 不可恢复类（MALFORMED——报文非法，重放必再失败）与未知类一律不在白名单内。 + */ +@Singleton +class ReplayService( + private val procState: ProcStateRepository, +) { + /** 可恢复错误类：codec 修复可重放 / 未实装补齐可重放 / 基础设施抖动可重放 / 重试耗尽后人工复核可重放。 */ + val replayableErrorClasses: Set = + setOf(ErrorClass.CODEC_ERROR, ErrorClass.UNSUPPORTED, ErrorClass.INFRA, ErrorClass.EXHAUSTED) + + /** 只重放白名单内的类；请求含 MALFORMED 等非法类时静默忽略该类。 */ + fun replay(requested: Collection): Int { + val allowed = requested.filter { it in replayableErrorClasses } + if (allowed.isEmpty()) return 0 + return procState.requeueByErrorClasses(allowed) + } + + /** 默认入口：重放全部可恢复类。 */ + fun replayAll(): Int = replay(replayableErrorClasses) +} 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 bc4179f..41d1040 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 @@ -12,6 +12,7 @@ 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.redis.FlightRedisClient +import com.gzzn.omms.msgexchange.nextgen.infra.retry.ProcFailure import com.gzzn.omms.msgexchange.nextgen.jobs.JobExecutor import jakarta.inject.Singleton import java.time.Duration @@ -37,9 +38,12 @@ class Pump( while (running) { try { tick() - } catch (t: Throwable) { - // U08/R02：主泵不得因单次异常静默死亡——任一 codec/DB/Redis 抖动不得终止队列； - // 具体失败落 FAILED/DEAD 与告警见 MessageProcessor/U12。 + } catch (e: InterruptedException) { + Thread.currentThread().interrupt() + return + } catch (e: Exception) { + // U08：loop 只作最后防线——失败状态迁移已在 processOne/execute 的边界内完成； + // 致命 Error 不在此捕获（任其终止进程，保证“异常必可见”）。 sleepQuietly(props.pipeline.pollInterval) } } @@ -56,11 +60,12 @@ class Pump( head == null -> sleepQuietly(props.pipeline.pollInterval) head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > Instant.now() -> if (poisoned(head)) { - upgradeDead(head, head.lastError ?: "head-deadline-exceeded") + 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 毒丸守卫） + // PENDING、或 FAILED 退避已到期：交处理入口（内部有边界化失败迁移与 attempts 守卫） else -> processor.processOne(head) } } @@ -69,10 +74,6 @@ class Pump( 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: ProcState?) = head == null || head.state == ProcStatus.FAILED @@ -94,9 +95,9 @@ class Pump( /** * ACMA-8 流程 2 processOne：解码→绑定→纯函数决策→提交。 - * U08/U10/U11（N06/N21/T06）：统一失败治理——「未实装/未注册/CODEC_ERROR/staging 未实装」 - * 一律 FAILED（可重放，带 attempts+nextAttemptAt，绝不直接写终态）；仅 MALFORMED（报文非法） - * 与「重试耗尽」写 DEAD 终态。attempts ≥ maxAttempts 的 FAILED 在入口即升级 DEAD（防退避到期后无限重试）。 + * U08/U10/U11：失败状态迁移全部在“持有具体 head”的边界内完成—— + * 任何意外异常 → FAILED(INFRA)+退避（ProcFailure）；attempts 达上限 → DEAD(EXHAUSTED)； + * MALFORMED 直接 DEAD；InterruptedException 恢复中断位并上抛（不被普通恢复逻辑吞掉）。 */ @Singleton class MessageProcessor( @@ -107,11 +108,24 @@ class MessageProcessor( private val handlers: HandlerHolder, private val redis: FlightRedisClient, private val snapshotFlow: SnapshotFlow, + private val procFailure: ProcFailure, private val props: PipelineProps, ) { fun processOne(head: ProcState) { - // U08 入口守卫：FAILED 行退避到期后再处理前，先判 attempts 毒丸（与 tick poisoned 同值） - if (head.state == ProcStatus.FAILED && head.attempts >= props.pipeline.maxAttempts) { + try { + processInternal(head) + } catch (e: InterruptedException) { + Thread.currentThread().interrupt() + throw e + } catch (e: Exception) { + // U08：边界化——异常归于本条 head，写 FAILED/DEAD，而不是穿出杀 pump + 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)) { procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.EXHAUSTED, lastError = head.lastError ?: "max-attempts") return @@ -129,7 +143,7 @@ class MessageProcessor( procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = r.failure.detail) } else { - failWithBackoff(head, r.failure.errorClass, r.failure.detail) + procFailure.fail(head, r.failure.errorClass, r.failure.detail) } return } @@ -149,7 +163,7 @@ class MessageProcessor( val handler = handlers.registry.dispatcherFor(decoded) if (handler == null) { // U10（N21）：未注册 ≠ 报文非法——写 FAILED（可重放），绝不写终态 - failWithBackoff(head, ErrorClass.UNSUPPORTED, "no-handler:${decoded.typeTag}") + procFailure.fail(head, ErrorClass.UNSUPPORTED, "no-handler:${decoded.typeTag}") return } val schdKind = decoded.kind as? com.gzzn.omms.msgexchange.nextgen.domain.MsgKind.Schd @@ -175,18 +189,6 @@ class MessageProcessor( 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 a31595e..fa58d4b 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,6 +1,5 @@ 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 @@ -9,28 +8,28 @@ 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 com.gzzn.omms.msgexchange.nextgen.infra.retry.ProcFailure import jakarta.inject.Singleton -import java.time.Instant /** * ACMA-8 流程 4：日计划快照（generation，主泵内执行）。 * staging（内存瞬态，崩溃从 raw 整包重放）→ Lua 原子“覆盖+按代差删”（I4/I5） * → putGenIfVersion CAS + SUCCEEDED（重放幂等，版本不二次自增——恢复协议细节见 U09）。 - * U10/T07 占位安全化：staging 未实装 → FAILED(UNSUPPORTED)+退避（可重放），绝不写 DEAD 终态； - * CAS 冲突 → FAILED(INFRA)+退避（不再无 nextAttemptAt 紧循环，N06/N28）。 + * 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 props: PipelineProps, + private val procFailure: ProcFailure, ) { 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) { - failWithBackoff(head, ErrorClass.UNSUPPORTED, staged.reason) // U10：未实装 → 可重放，非终态 + procFailure.fail(head, ErrorClass.UNSUPPORTED, staged.reason) // U10：未实装 → 可重放，非终态 return } val normalized = (staged as StageResult.Ok).flights // (flid, payloadJson) @@ -48,24 +47,13 @@ class SnapshotFlow( // 重放路径：version 已是目标值 → no-op 视为成功 val again = refData.getGen(day) if (again == null || again.version != expected + 1) { - failWithBackoff(head, ErrorClass.INFRA, "gen-cas-conflict") // N06/N28：带退避，禁止紧循环 + procFailure.fail(head, ErrorClass.INFRA, "gen-cas-conflict") // N06/N28：带退避，禁止紧循环 return } } 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 结果（骨架）。 */ sealed interface StageResult { data class Ok(val day: String, val flights: List>) : StageResult diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/TestClocks.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/TestClocks.kt new file mode 100644 index 0000000..9729b4a --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/TestClocks.kt @@ -0,0 +1,27 @@ +package com.gzzn.omms.msgexchange.nextgen + +import java.time.Clock +import java.time.Instant +import java.time.ZoneId +import java.time.ZoneOffset + +/** 可手动推进的可注入时钟（测试用），替代真实时间避免脆弱睡眠测试。 */ +internal class MutableClock( + initial: Instant, + private val zone: ZoneId = ZoneOffset.UTC, +) : Clock() { + + var instant: Instant = initial + + override fun instant(): Instant = instant + + override fun getZone(): ZoneId = zone + + override fun withZone(zoneId: ZoneId): Clock = MutableClock(instant, zoneId) + + fun advance(millis: Long) { instant = instant.plusMillis(millis) } + + companion object { + val BASE: Instant = Instant.parse("2026-09-06T02:00:00Z") + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/config/InfraBindingStartupTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/config/InfraBindingStartupTest.kt new file mode 100644 index 0000000..25d5ee8 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/config/InfraBindingStartupTest.kt @@ -0,0 +1,39 @@ +package com.gzzn.omms.msgexchange.nextgen.config + +import io.micronaut.test.extensions.junit5.annotation.MicronautTest +import jakarta.inject.Inject +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Test +import java.sql.Connection +import javax.sql.DataSource + +/** + * U03（N02/R09/R01a 验收的“启动级”版本）：基础设施配置在真实上下文启动时绑定生效—— + * 注入 DataSource 并建立真实连接（H2 内存，application-test.yml），证明 datasources.default.* + * 键位与驱动解析正确（而非“看似配置实则未生效”）。Pump 等业务 bean 懒加载，不依赖实仓储。 + */ +@MicronautTest +class InfraBindingStartupTest { + + @Inject + lateinit var dataSource: DataSource + + @Inject + lateinit var props: PipelineProps + + @Test + fun `datasource binds and yields a live connection`() { + assertNotNull(dataSource) + dataSource.connection.use { c: Connection -> + assertNotNull(c.metaData) + c.createStatement().use { it.execute("SELECT 1") } + } + } + + @Test + fun `registration disabled by test profile`() { + assertNotNull(props) + // eureka 注册在 application-test.yml 经 msgx.register-eureka=false 关闭——启动级验证绑定可达 + assertNotNull(dataSource) + } +} 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 index 42e0bbe..e5efc64 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/DispatcherTickTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/DispatcherTickTest.kt @@ -1,21 +1,24 @@ package com.gzzn.omms.msgexchange.nextgen.delivery +import com.gzzn.omms.msgexchange.nextgen.MutableClock 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 com.gzzn.omms.msgexchange.nextgen.infra.retry.FailureScheduler 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.Duration import java.time.Instant -import kotlin.test.assertFails /** - * U06（N03）：KAFKA_SCHD 不被逐条循环吞掉——唯一出口 flushSchd 批量聚合； - * 聚合只发每 FLID 最新一态；发送失败整批保持 PENDING 且不推进 lastFlush（N24）。 + * U06（N03）+ U08 闭环：schd 出口只走 flushSchd； + * 批发送失败 → 整批 attempts 递增 + 指数退避（队首未到期不 claim），达上限整批 DEAD/DLQ； + * 时间经可注入 Clock（MutableClock），不依赖真实睡眠。 */ class DispatcherTickTest { @@ -47,9 +50,10 @@ class DispatcherTickTest { if (i >= 0) events[i] = events[i].copy(state = EventStatus.PENDING, attempts = attempts, nextAttemptAt = nextAttemptAt) } - override fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String) { + override fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String, attempts: Int?) { val i = events.indexOfFirst { it.eventId == eventId } - if (i >= 0) events[i] = events[i].copy(state = EventStatus.DEAD, errorClass = errorClass, lastError = lastError) + if (i >= 0) events[i] = events[i].copy(state = EventStatus.DEAD, errorClass = errorClass, lastError = lastError, + attempts = attempts ?: events[i].attempts) } override fun insertSync(events: List) = Unit @@ -68,7 +72,10 @@ class DispatcherTickTest { override fun projectRedis(payloadJson: String) = Unit } - private fun props(): PipelineProps = PipelineProps() // schd.flush-period 默认 3s + private val clock = MutableClock(MutableClock.BASE) + + private fun dispatcher(repo: FakeRepo, port: FakePort, p: PipelineProps = PipelineProps()): Dispatcher = + Dispatcher(repo, port, p, FailureScheduler(p, clock)) private fun ev(id: Long, target: String, key: String?, payload: String) = MsgEvent(eventId = id, target = target, partitionKey = key, payloadJson = payload, state = EventStatus.PENDING) @@ -77,57 +84,101 @@ class DispatcherTickTest { 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()) + val d = dispatcher(repo, port) 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() + d.tick() // 时钟未推进 → flush 周期未到；逐条循环不得消费 schd assertEquals(EventStatus.PENDING, repo.events.first { it.eventId == 4L }.state) - assertEquals(1, port.sent.count { it.first == "schd" }) // 第二次 tick 未额外发送 + assertEquals(1, port.sent.count { it.first == "schd" }) } @Test fun `flushSchd aggregates latest per flight and marks all sent`() { val repo = FakeRepo() val port = FakePort() - val d = Dispatcher(repo, port, props()) + val d = dispatcher(repo, port) 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`() { + fun `flush failure schedules per-event backoff and retries after due`() { val repo = FakeRepo() val port = FakePort().apply { failTopic = "schd" } - val d = Dispatcher(repo, port, props()) + val d = dispatcher(repo, port) repo.enqueue(ev(1, Targets.KAFKA_SCHD, "F1", """{"v":"1"}""")) - assertFails { d.flushSchd() } // 发送失败 → 异常上抛（tick 侧隔离），事件保持 PENDING - assertEquals(EventStatus.PENDING, repo.events.single().state) + d.flushSchd() + val e1 = repo.events.single() + assertEquals(EventStatus.PENDING, e1.state) + assertEquals(1, e1.attempts) + assertNotNull(e1.nextAttemptAt) // 指数退避落库（attempts=1 → 首档） assertEquals(0, port.sent.size) + clock.advance(1000) // 退避到期 port.failTopic = null - d.flushSchd() // lastFlush 未推进 → 下个周期整批重试成功 + d.flushSchd() assertEquals(EventStatus.SENT, repo.events.single().state) assertEquals(1, port.sent.count { it.first == "schd" }) } + + @Test + fun `repeated batch failures backoff grows and whole batch goes DEAD DLQ at limit`() { + val props = PipelineProps().apply { pipeline.maxAttempts = 2 } + 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"}""")) + repo.enqueue(ev(2, Targets.KAFKA_SCHD, "F2", """{"v":"2"}""")) + + d.flushSchd() // 失败#1：attempts=1，退避 backoff[1]=1000ms + val after1 = repo.events.map { it.attempts to it.state } + assertEquals(listOf(1 to EventStatus.PENDING, 1 to EventStatus.PENDING), after1) + assertTrue(repo.events.all { it.nextAttemptAt == MutableClock.BASE.plusMillis(1000) }) + + d.flushSchd() // 时钟未推进 → 队首未到期，不 claim（不推进 lastFlush） + assertEquals(1, repo.events.first { it.eventId == 1L }.attempts) + + clock.advance(1000) + d.flushSchd() // 失败#2：attempts=2 == maxAttempts → 整批 DEAD + assertTrue(repo.events.all { it.state == EventStatus.DEAD && it.errorClass == ErrorClass.EXHAUSTED }) + assertTrue(repo.events.all { it.attempts == 2 }) + assertEquals(0, port.sent.size) + + // DEAD 行不再参与 claim：空批 → 无新发送、无异常 + d.flushSchd() + assertEquals(0, port.sent.size) + } + + @Test + fun `KAFKA msg delivery unaffected while schd batch is retrying`() { + val props = PipelineProps().apply { pipeline.maxAttempts = 2 } + 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"}""")) + repo.enqueue(ev(2, Targets.KAFKA_MSG, null, """{"m":1}""")) + + d.tick() + + assertEquals(EventStatus.SENT, repo.events.first { it.eventId == 2L }.state) // msg 正常 + assertEquals(1, repo.events.first { it.eventId == 1L }.attempts) // schd 进退避 + assertEquals(EventStatus.PENDING, repo.events.first { it.eventId == 1L }.state) + } } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ReplayServiceTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ReplayServiceTest.kt new file mode 100644 index 0000000..383e9a6 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ReplayServiceTest.kt @@ -0,0 +1,94 @@ +package com.gzzn.omms.msgexchange.nextgen.infra.retry + +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 org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Test +import java.time.Instant + +/** + * U11：显式重放入口只允许可恢复错误从 FAILED/DEAD 返回 PENDING（attempts 清零、立即重试）； + * MALFORMED（报文非法）永不被重放。 + */ +class ReplayServiceTest { + + private class FakeRepo : ProcStateRepository { + val rows = linkedMapOf() + val requeueCalls = mutableListOf>() + + fun seed(id: Long, status: ProcStatus, ec: ErrorClass?) { + rows[id] = ProcState(id, status, identityKey = "k$id", attempts = 3, errorClass = ec, lastError = "x") + } + + override fun insert(cminmsgsId: Long, state: ProcStatus) = Unit + override fun headUnfinished(): ProcState? = null + override fun tryBindIdentity(cminmsgsId: Long, identityKey: String): Boolean = true + override fun ownerOfIdentity(identityKey: String): Long? = null + override fun update( + cminmsgsId: Long, state: ProcStatus, nextAttemptAt: Instant?, attempts: Int?, + errorClass: ErrorClass?, lastError: String?, + ) = Unit + + override fun requeueByErrorClasses(errorClasses: List): Int { + requeueCalls += errorClasses + var n = 0 + rows.keys.toList().forEach { id -> + val s = rows[id]!! + if (s.errorClass != null && s.errorClass in errorClasses && + (s.state == ProcStatus.FAILED || s.state == ProcStatus.DEAD)) { + rows[id] = s.copy(state = ProcStatus.PENDING, attempts = 0, nextAttemptAt = null) + n++ + } + } + return n + } + } + + @Test + fun `only recoverable classes are requeued, MALFORMED stays terminal`() { + val repo = FakeRepo() + repo.seed(1, ProcStatus.DEAD, ErrorClass.CODEC_ERROR) + repo.seed(2, ProcStatus.DEAD, ErrorClass.MALFORMED) + repo.seed(3, ProcStatus.FAILED, ErrorClass.UNSUPPORTED) + + val n = ReplayService(repo).replay(listOf(ErrorClass.CODEC_ERROR, ErrorClass.MALFORMED, ErrorClass.UNSUPPORTED)) + + assertEquals(2, n) + assertEquals(listOf(listOf(ErrorClass.CODEC_ERROR, ErrorClass.UNSUPPORTED)), repo.requeueCalls) + assertEquals(ProcStatus.PENDING, repo.rows[1]!!.state) + assertEquals(0, repo.rows[1]!!.attempts) + assertNull(repo.rows[1]!!.nextAttemptAt) + assertEquals(ProcStatus.DEAD, repo.rows[2]!!.state) // MALFORMED 永不被重放 + assertEquals(ProcStatus.PENDING, repo.rows[3]!!.state) + } + + @Test + fun `requesting only MALFORMED is a no-op`() { + val repo = FakeRepo() + repo.seed(2, ProcStatus.DEAD, ErrorClass.MALFORMED) + + val n = ReplayService(repo).replay(listOf(ErrorClass.MALFORMED)) + + assertEquals(0, n) + assertNull(repo.requeueCalls.lastOrNull()) // 白名单过滤后为空 → 不触达仓储 + assertEquals(ProcStatus.DEAD, repo.rows[2]!!.state) + } + + @Test + fun `replayAll reopens every recoverable class including EXHAUSTED`() { + val repo = FakeRepo() + repo.seed(4, ProcStatus.FAILED, ErrorClass.INFRA) + repo.seed(5, ProcStatus.DEAD, ErrorClass.EXHAUSTED) + repo.seed(6, ProcStatus.DEAD, ErrorClass.MALFORMED) + + val n = ReplayService(repo).replayAll() + + assertEquals(2, n) + assertEquals(ProcStatus.PENDING, repo.rows[4]!!.state) + assertEquals(ProcStatus.PENDING, repo.rows[5]!!.state) + assertEquals(ProcStatus.DEAD, repo.rows[6]!!.state) + } +} 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 index 8b0d26d..6df2f64 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/MessageProcessorTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/MessageProcessorTest.kt @@ -1,5 +1,6 @@ package com.gzzn.omms.msgexchange.nextgen.processing +import com.gzzn.omms.msgexchange.nextgen.MutableClock import com.gzzn.omms.msgexchange.nextgen.codec.DecodeFailure import com.gzzn.omms.msgexchange.nextgen.codec.DecodeResult import com.gzzn.omms.msgexchange.nextgen.codec.XmlCodec @@ -25,24 +26,27 @@ 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 com.gzzn.omms.msgexchange.nextgen.infra.retry.FailureScheduler +import com.gzzn.omms.msgexchange.nextgen.infra.retry.ProcFailure 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.assertThrows 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。 + * U08/U10/U11：processOne 失败闭环—— + * ① 意外异常在边界内写 FAILED(INFRA)+退避，重试达上限转 DEAD(EXHAUSTED)（Pump 内部异常不杀泵）； + * ② 未注册/CODEC_ERROR/staging 未实装 → FAILED（绝不直接终态）；MALFORMED → DEAD； + * ③ InterruptedException/致命 Error 不被普通恢复逻辑吞掉。 */ class MessageProcessorTest { - // ---------- fakes ---------- private class FakeProcState : ProcStateRepository { var record = mutableMapOf() - val bound = mutableMapOf() // identityKey -> owner + val bound = mutableMapOf() override fun insert(cminmsgsId: Long, state: ProcStatus) { record[cminmsgsId] = ProcState(cminmsgsId, state) } @@ -72,16 +76,33 @@ class MessageProcessorTest { ) } + override fun requeueByErrorClasses(errorClasses: List): Int { + var n = 0 + record.keys.toList().forEach { id -> + val s = record[id]!! + if (s.errorClass != null && s.errorClass in errorClasses && + (s.state == ProcStatus.FAILED || s.state == ProcStatus.DEAD)) { + record[id] = s.copy(state = ProcStatus.PENDING, attempts = 0, nextAttemptAt = null) + n++ + } + } + return n + } + fun state(id: Long): ProcState = record.getValue(id) } private class FakeInbox : CminmsgInboxRepository { var raws = mutableMapOf() + var throwOnRawOf: Throwable? = null val backfilled = mutableListOf>() override fun insertRaw(rawXml: String): Long = 0 - override fun rawOf(cminmsgsId: Long): String? = raws[cminmsgsId] + override fun rawOf(cminmsgsId: Long): String? { + throwOnRawOf?.let { throw it } + return raws[cminmsgsId] + } override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) { backfilled += listOf(cminmsgsId, sndr, type, styp, seqn) @@ -98,7 +119,7 @@ class MessageProcessorTest { 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 markDead(eventId: Long, errorClass: ErrorClass, lastError: String, attempts: Int?) = Unit override fun insertSync(events: List) = Unit } @@ -126,6 +147,8 @@ class MessageProcessorTest { } // ---------- helpers ---------- + private val clock = MutableClock(MutableClock.BASE) + private fun msg(seqn: String) = DecodedMessage( meta = MetaFields(sndr = "AODB", type = "FLOP", styp = "DELY", seqn = seqn.toLong(), dttm = 20260906120000L), kind = MsgKind.Flop("DELY"), @@ -148,8 +171,10 @@ class MessageProcessorTest { 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) + val scheduler = FailureScheduler(props, clock) + val procFailure = ProcFailure(procState, scheduler) + val snapshot = SnapshotFlow(procState, FakeRefData(), FakeRedis, procFailure) + return MessageProcessor(inbox, procState, events, CodecHolder(codec), HandlerHolder(registry), FakeRedis, snapshot, procFailure, props) } // ---------- tests ---------- @@ -177,6 +202,7 @@ class MessageProcessorTest { val s = ps.state(1) assertEquals(ProcStatus.FAILED, s.state) assertEquals(ErrorClass.CODEC_ERROR, s.errorClass) + assertEquals(1, s.attempts) assertNotNull(s.nextAttemptAt) } @@ -193,15 +219,13 @@ class MessageProcessorTest { } @Test - fun `failed head at max attempts upgrades to DEAD at entry`() { + fun `legacy 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()) // 入口即升级，未进入处理 + assertEquals(ProcStatus.DEAD, ps.state(1).state) + assertEquals(ErrorClass.EXHAUSTED, ps.state(1).errorClass) } @Test @@ -212,7 +236,6 @@ class MessageProcessorTest { 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) @@ -222,7 +245,7 @@ class MessageProcessorTest { @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) + ps.record[1] = head() inbox.raws[1] = "" ps.bound["AODB|FLOP|DELY|1"] = 2L // 另一条消息已持有该键 @@ -233,4 +256,67 @@ class MessageProcessorTest { assertTrue(s.lastError == "duplicate-of:2") assertTrue(ev.inserted.isEmpty()) } + + // ---------- U08 边界化（评审要求①③） ---------- + @Test + fun `unexpected exception maps to FAILED INFRA with backoff at boundary`() { + val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents() + ps.record[1] = head() + inbox.throwOnRawOf = RuntimeException("db-down") + + processor(ps, inbox, ev, FakeCodec()).processOne(ps.state(1)) + + val s = ps.state(1) + assertEquals(ProcStatus.FAILED, s.state) + assertEquals(ErrorClass.INFRA, s.errorClass) + assertEquals(1, s.attempts) + assertNotNull(s.nextAttemptAt) + assertTrue(s.lastError == "db-down") + } + + @Test + fun `repeated unexpected exceptions escalate to DEAD EXHAUSTED at max attempts`() { + val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents() + ps.record[1] = head() + inbox.throwOnRawOf = RuntimeException("db-down") + val p = processor(ps, inbox, ev, FakeCodec()) + + var attempts = 0 + while (ps.state(1).state != ProcStatus.DEAD && attempts < 10) { + p.processOne(ps.state(1)) + clock.advance(60_000) // 推进时钟：让退避/毒丸时间语义确定 + attempts++ + } + + val s = ps.state(1) + assertEquals(ProcStatus.DEAD, s.state, "should exhaust within maxAttempts(=5)") + assertEquals(ErrorClass.EXHAUSTED, s.errorClass) + assertTrue(s.lastError?.contains("attempts=5") == true) + assertEquals(5, s.attempts) + } + + @Test + fun `fatal Error is not swallowed by exception recovery`() { + val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents() + ps.record[1] = head() + inbox.throwOnRawOf = AssertionError("fatal") + + assertThrows(AssertionError::class.java) { + processor(ps, inbox, ev, FakeCodec()).processOne(ps.state(1)) + } + // 未被普通恢复逻辑改写成 FAILED（Error 不落入 catch Exception） + assertEquals(ProcStatus.PENDING, ps.state(1).state) + } + + @Test + fun `InterruptedException restores flag and propagates`() { + val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents() + ps.record[1] = head() + inbox.throwOnRawOf = InterruptedException("stop") + + assertThrows(InterruptedException::class.java) { + processor(ps, inbox, ev, FakeCodec()).processOne(ps.state(1)) + } + assertTrue(Thread.interrupted(), "interrupt flag must be restored") + } }