windyboy
a936238368
Add JacksonXmlCodec (XXE-safe META/FLOP parse), GtdtHandler, and unified PipelineHolderFactory; route DNLD/FLOP through FlightStateEngine persistNextStates.
254 lines
12 KiB
Kotlin
254 lines
12 KiB
Kotlin
package com.gzzn.omms.msgexchange.processing
|
||
|
||
import com.gzzn.omms.msgexchange.config.PipelineProps
|
||
import com.gzzn.omms.msgexchange.domain.DecodedMessage
|
||
import com.gzzn.omms.msgexchange.domain.flight.FlightStateEngine
|
||
import com.gzzn.omms.msgexchange.domain.ErrorClass
|
||
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.infra.persistence.CminmsgInboxRepository
|
||
import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository
|
||
import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository
|
||
import com.gzzn.omms.msgexchange.infra.persistence.PumpJobRepository
|
||
import com.gzzn.omms.msgexchange.infra.persistence.FlightSchdRepository
|
||
import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager
|
||
import com.gzzn.omms.msgexchange.infra.retry.ProcFailure
|
||
import com.gzzn.omms.msgexchange.jobs.JobExecutor
|
||
import jakarta.inject.Singleton
|
||
import java.time.Duration
|
||
import java.time.Instant
|
||
|
||
/**
|
||
* ACMA-8 流程 2：处理主泵——严格 FIFO + HOL + 毒丸 DEAD 升级（I1）；
|
||
* 阶段 A 运营航班 FLIGHT_SCHD 与待发事件、终态同事务原子提交（I2/I5，ACM2-28 定案）。
|
||
*/
|
||
@Singleton
|
||
class Pump(
|
||
private val procState: ProcStateRepository,
|
||
private val pumpJobs: PumpJobRepository,
|
||
private val inbox: CminmsgInboxRepository,
|
||
private val processor: MessageProcessor,
|
||
private val jobExecutor: JobExecutor,
|
||
private val props: PipelineProps,
|
||
) {
|
||
private val log = org.slf4j.LoggerFactory.getLogger(Pump::class.java)
|
||
|
||
@Volatile
|
||
private var running = true
|
||
|
||
/** 优雅停机：loop 在当前 tick 收尾后退出；线程中断由 Runner 负责。 */
|
||
fun stop() {
|
||
running = false
|
||
}
|
||
|
||
fun loop() {
|
||
while (running) {
|
||
try {
|
||
tick()
|
||
} catch (e: InterruptedException) {
|
||
Thread.currentThread().interrupt()
|
||
return
|
||
} catch (e: Exception) {
|
||
// U08：loop 只作最后防线——失败状态迁移已在 processOne/execute 的边界内完成；
|
||
// 致命 Error 不在此捕获（任其终止进程，保证“异常必可见”）。
|
||
sleepQuietly(props.pipeline.pollInterval)
|
||
}
|
||
}
|
||
}
|
||
|
||
internal fun tick() {
|
||
val job = pumpJobs.headQueued()
|
||
val head = procState.headUnfinished()
|
||
|
||
// 统一 FIFO：JOB 与消息同队列语义（定时任务产物不绕过队头顺序，决策 1）。
|
||
val nextJob = job?.takeIf { jobBefore(job, head) }
|
||
when {
|
||
nextJob != null -> execute(nextJob)
|
||
head == null -> sleepQuietly(props.pipeline.pollInterval)
|
||
head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > Instant.now() ->
|
||
if (poisoned(head)) {
|
||
log.error("poison -> DEAD id={} attempts={} lastError={}", head.cminmsgsId, head.attempts, head.lastError)
|
||
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 守卫）
|
||
else -> processor.processOne(head)
|
||
}
|
||
}
|
||
|
||
private fun poisoned(head: ProcState): Boolean =
|
||
head.attempts >= props.pipeline.maxAttempts ||
|
||
Duration.between(head.updatedAt, Instant.now()) > props.pipeline.headDeadline
|
||
|
||
/** JOB 与队头消息的先后由入队时间近似；实装以统一序号列保证（阶段 1 后续）。 */
|
||
private fun jobBefore(job: PumpJobRepository.Job, head: ProcState?) =
|
||
head == null || head.state == ProcStatus.FAILED
|
||
|
||
private fun execute(job: PumpJobRepository.Job) {
|
||
pumpJobs.markRunning(job.jobId)
|
||
try {
|
||
jobExecutor.execute(job)
|
||
pumpJobs.markDone(job.jobId)
|
||
} catch (e: Exception) {
|
||
pumpJobs.markFailed(job.jobId, e.message ?: "unknown")
|
||
}
|
||
}
|
||
|
||
private fun sleepQuietly(d: Duration) {
|
||
if (!d.isNegative && !d.isZero) Thread.sleep(d.toMillis().coerceAtLeast(1))
|
||
}
|
||
}
|
||
|
||
/**
|
||
* ACMA-8 流程 2 processOne：解码→绑定→纯函数决策→提交。
|
||
* U08/U10/U11：失败状态迁移全部在“持有具体 head”的边界内完成——
|
||
* 任何意外异常 → FAILED(INFRA)+退避（ProcFailure）；attempts 达上限 → DEAD(EXHAUSTED)；
|
||
* MALFORMED 直接 DEAD；InterruptedException 恢复中断位并上抛（不被普通恢复逻辑吞掉）。
|
||
*/
|
||
@Singleton
|
||
class MessageProcessor(
|
||
private val inbox: CminmsgInboxRepository,
|
||
private val procState: ProcStateRepository,
|
||
private val msgEvents: MsgEventRepository,
|
||
private val codecHolder: CodecHolder,
|
||
private val handlers: HandlerHolder,
|
||
private val flightSchd: FlightSchdRepository,
|
||
private val snapshotFlow: SnapshotFlow,
|
||
private val procFailure: ProcFailure,
|
||
private val props: PipelineProps,
|
||
private val txManager: PipelineTransactionManager,
|
||
) {
|
||
private val log = org.slf4j.LoggerFactory.getLogger(MessageProcessor::class.java)
|
||
private val flidRegex = Regex("<FLID>(.*?)</FLID>", RegexOption.IGNORE_CASE)
|
||
|
||
private fun extractCandidateFlids(decoded: DecodedMessage): Set<String> {
|
||
val fromXml = flidRegex.findAll(decoded.rawXml).map { it.groupValues[1].trim() }.filter { it.isNotEmpty() }.toSet()
|
||
if (fromXml.isNotEmpty()) return fromXml
|
||
val b = decoded.body
|
||
if (b is Map<*, *>) {
|
||
val flid = b["FLID"] ?: b["flid"]
|
||
if (flid != null) return setOf(flid.toString())
|
||
}
|
||
return emptySet()
|
||
}
|
||
|
||
fun processOne(head: ProcState) {
|
||
com.gzzn.omms.msgexchange.infra.log.TraceLog.withTrace(head.cminmsgsId) {
|
||
try {
|
||
processInternal(head)
|
||
} catch (e: InterruptedException) {
|
||
Thread.currentThread().interrupt()
|
||
throw e
|
||
} catch (e: Exception) {
|
||
// U08：边界化——异常归于本条 head，写 FAILED/DEAD，而不是穿出杀 pump
|
||
log.warn("processOne unexpected failure id={} ec=INFRA msg={}", head.cminmsgsId, e.message ?: e.javaClass.simpleName)
|
||
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)) {
|
||
log.error("head exhausted at entry -> DEAD id={} attempts={}", head.cminmsgsId, head.attempts)
|
||
procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.EXHAUSTED,
|
||
lastError = head.lastError ?: "max-attempts")
|
||
return
|
||
}
|
||
|
||
val raw = inbox.rawOf(head.cminmsgsId) ?: run {
|
||
log.error("raw missing -> DEAD(MALFORMED) id={}", head.cminmsgsId)
|
||
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.codec.DecodeResult.Ok -> r.message
|
||
is com.gzzn.omms.msgexchange.codec.DecodeResult.Err -> {
|
||
// T06（U11）：MALFORMED（报文非法）→ DEAD 不重试；CODEC_ERROR（可随 codec 修复重放）→ FAILED 退避
|
||
if (r.failure.errorClass == ErrorClass.MALFORMED) {
|
||
log.error("decode MALFORMED -> DEAD id={} detail={}", head.cminmsgsId, r.failure.detail)
|
||
procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED,
|
||
lastError = r.failure.detail)
|
||
} else {
|
||
log.warn("decode {} -> FAILED id={} detail={}", r.failure.errorClass, head.cminmsgsId, r.failure.detail)
|
||
procFailure.fail(head, r.failure.errorClass, r.failure.detail)
|
||
}
|
||
return
|
||
}
|
||
}
|
||
|
||
// I3：identity 仅首次绑定（head.identityKey == null）；FAILED 重试不重绑（N25：入参传递 head，不再二次查询）
|
||
if (head.identityKey == null) {
|
||
val identity = Identity.of(decoded, props.identity)
|
||
if (!procState.tryBindIdentity(head.cminmsgsId, identity)) {
|
||
val owner = procState.ownerOfIdentity(identity) ?: -1L
|
||
log.info("duplicate-of:{} -> SKIPPED id={}", owner, head.cminmsgsId)
|
||
procState.update(head.cminmsgsId, ProcStatus.SKIPPED, lastError = "duplicate-of:$owner")
|
||
return
|
||
}
|
||
}
|
||
// 快照消息走专属流程（流程 4：staging → SQL 批处理/域内差删/CAS 同事务）
|
||
val handler = handlers.registry.dispatcherFor(decoded)
|
||
if (handler == null) {
|
||
// U10（N21）：未注册 ≠ 报文非法——写 FAILED（可重放），绝不写终态
|
||
log.warn("no-handler:{} -> FAILED(UNSUPPORTED) id={}", decoded.typeTag, head.cminmsgsId)
|
||
procFailure.fail(head, ErrorClass.UNSUPPORTED, "no-handler:${decoded.typeTag}")
|
||
return
|
||
}
|
||
val schdKind = decoded.kind as? com.gzzn.omms.msgexchange.domain.MsgKind.Schd
|
||
if (schdKind != null &&
|
||
schdKind.subtype == com.gzzn.omms.msgexchange.domain.MsgKind.SchdSubtype.DNLD
|
||
) {
|
||
snapshotFlow.publishSnapshot(head, decoded)
|
||
return
|
||
}
|
||
|
||
val candidateFlids = extractCandidateFlids(decoded)
|
||
val flightView = if (candidateFlids.isNotEmpty()) {
|
||
flightSchd.findByFlids(candidateFlids)
|
||
} else {
|
||
emptyMap()
|
||
}
|
||
|
||
val decision = handler.decide(flightView, decoded) // 纯函数
|
||
|
||
// 事务 2（自有 PG 单事务原子）：
|
||
// upsert FLIGHT_SCHD(decision.flightChanges) + insert MSG_EVENT + PROC_STATE → SUCCEEDED
|
||
val events = buildList {
|
||
decision.msgNotifies.forEach { add(MsgEvent(target = Targets.KAFKA_MSG, payloadJson = it.payloadJson)) }
|
||
decision.schdPush.forEach { add(MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = it.flid, payloadJson = it.payloadJson)) }
|
||
}
|
||
|
||
val messageId = head.cminmsgsId.toString()
|
||
txManager.inTransaction {
|
||
if (decision.flightChanges.isNotEmpty()) {
|
||
val nextStates = decision.flightChanges.map { change ->
|
||
val current = flightSchd.findNextStateByFlid(change.flid)
|
||
val commands = FlightStateEngine.commandsFromFields(change.flid, change.fields, snapshotReplace = false)
|
||
FlightStateEngine.apply(current, commands, messageId, bumpVersion = true)
|
||
}
|
||
flightSchd.persistNextStates(null, nextStates, snapshotReplace = false)
|
||
}
|
||
if (events.isNotEmpty()) {
|
||
msgEvents.insertAll(events)
|
||
}
|
||
procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED)
|
||
}
|
||
|
||
try {
|
||
inbox.backfillOnSuccess(head.cminmsgsId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn)
|
||
} catch (e: Exception) {
|
||
log.error("backfill failed after SUCCEEDED id={} (compensation required)", head.cminmsgsId, e)
|
||
}
|
||
log.info("SUCCEEDED id={} events={} flightChanges={}", head.cminmsgsId, events.size, decision.flightChanges.size)
|
||
}
|
||
}
|
||
|
||
/** 延迟装配占位（阶段 1 后续以 Micronaut Bean 替换直连构造）。 */
|
||
class CodecHolder(val codec: com.gzzn.omms.msgexchange.codec.XmlCodec)
|
||
class HandlerHolder(val registry: HandlerRegistry)
|