feat(observability): U12 可观测底座（ACM2-10 U12/R05/N31b）

- logback：logstash 通道先加 AsyncAppender（queueSize 4096、neverBlock、discardingThreshold 0）
  ——logstash 不可达不阻塞业务线程（N31b 顺序约束：先降级再补日志）；MDC 增加 traceId/target
- 结构化日志 + MDC traceId（TraceLog.withTrace=cminmsgsId/eventId）：
  InboxService 收报 INFO；MessageProcessor SUCCEEDED/SKIPPED INFO、FAILED/解码失败 WARN、
  DEAD(MALFORMED/毒丸/EXHAUSTED) ERROR；SnapshotFlow staging/CAS WARN、快照成功 INFO；
  Dispatcher 发送 DEBUG、逐条/整批重试 WARN、DEAD(DLQ) ERROR、flushSchd 批次 INFO；
  ReplayService 重放 INFO/非法类 WARN；PipelineLifecycle 起停 INFO
- 自定义健康指示器（infra/health）：redis-flight-store / kafka-delivery——BeanProvider 可选解析，
  缺 bean（未 stub 也未实装）报 DOWN 而非启动失败；/health 聚合
- 保留：micrometer 队列深度/投递延迟 gauge 依赖版本选型（micronaut-micrometer 与平台 BOM 5.1.3
  对齐待锁），随数据层/真实 client 批次补；U13 毒丸超时告警与 DLQ 巡检入口仍属 WP2

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineLifecycle.kt

@@ -20,6 +20,7 @@ class PipelineLifecycle(
     private val pump: Pump,
     private val dispatcher: Dispatcher,
 ) {
+    private val log = org.slf4j.LoggerFactory.getLogger(PipelineLifecycle::class.java)
     private val threads = mutableListOf<Thread>()
 
     @Volatile
@@ -35,6 +36,7 @@ class PipelineLifecycle(
         started = true
         threads += spawn("msgx-pump", pump::loop)
         threads += spawn("msgx-dispatcher", dispatcher::loop)
+        log.info("pipeline loops started (pump, dispatcher)")
     }
 
     private fun spawn(name: String, body: () -> Unit): Thread =
@@ -46,5 +48,6 @@ class PipelineLifecycle(
         dispatcher.stop()
         threads.forEach { it.interrupt() }          // 解除 Thread.sleep 阻塞，加速退出
         threads.forEach { runCatching { it.join(3000) } }
+        log.info("pipeline loops stopped")
     }
 }

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt

@@ -38,6 +38,8 @@ class Dispatcher(
     private val props: PipelineProps,
     private val scheduler: FailureScheduler,
 ) {
+    private val log = org.slf4j.LoggerFactory.getLogger(Dispatcher::class.java)
+
     @Volatile
     private var running = true
 
@@ -76,6 +78,7 @@ class Dispatcher(
             try {
                 deliver(t, head)
                 msgEvents.markSent(head.eventId!!)
+                log.debug("sent target={} eventId={}", t, head.eventId)
                 if (t == Targets.ES_FLIGHT_HTS && props.phase == PipelineProps.Phase.B) {
                     // 定案 2：ES 投递成功 → 同线程同步 enqueue 删除事件（不轮询 ack）
                     msgEvents.insertSync(listOf(MsgEvent(target = Targets.REDIS_FLIGHT_INFO, payloadJson = deleteOf(head))))
@@ -96,8 +99,10 @@ class Dispatcher(
     private fun retryOrDead(e: MsgEvent, lastError: String) {
         val attempts = e.attempts + 1
         if (scheduler.exhausted(attempts)) {
+            log.error("event DEAD(DLQ) eventId={} attempts={} lastError={}", e.eventId, attempts, lastError)
             msgEvents.markDead(e.eventId!!, ErrorClass.EXHAUSTED, lastError, attempts)
         } else {
+            log.warn("event retry scheduled eventId={} attempts={} nextAttemptAt={}", e.eventId, attempts, scheduler.nextAttemptAt(attempts))
             msgEvents.scheduleRetry(e.eventId!!, scheduler.nextAttemptAt(attempts), attempts)
         }
     }
@@ -128,13 +133,16 @@ class Dispatcher(
             return                                   // 队首仍在退避：整批等待，不推进 lastFlush（到期再试）
         }
         val payload = SchdAggregation.latestPerFlight(pending).joinToString(",", "[", "]")
+        log.debug("flushSchd batch size={} payloadLen={}", pending.size, payload.length)
         try {
             port.sendKafka("schd", payload)
         } catch (e: Exception) {
+            log.warn("flushSchd send failed batch={} -> batch retryOrDead: {}", pending.size, e.message)
             pending.forEach { retryOrDead(it, "schd-send: ${e.message ?: "unknown"}") }
             return                                   // 不推进 lastFlush：整批退避（含 DEAD 出队）后到期重试
         }
         msgEvents.markAllSent(pending.mapNotNull { it.eventId })
+        log.info("flushSchd sent batch={}", pending.size)
         lastFlush = scheduler.now()
     }
 

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/health/HealthIndicators.kt

@@ -0,0 +1,48 @@
+package com.gzzn.omms.msgexchange.nextgen.infra.health
+
+import com.gzzn.omms.msgexchange.nextgen.delivery.DeliveryPort
+import com.gzzn.omms.msgexchange.nextgen.infra.redis.FlightRedisClient
+import io.micronaut.context.BeanProvider
+import io.micronaut.core.async.publisher.Publishers
+import io.micronaut.health.HealthStatus
+import io.micronaut.management.health.indicator.HealthIndicator
+import io.micronaut.management.health.indicator.HealthResult
+import jakarta.inject.Singleton
+import org.reactivestreams.Publisher
+
+/**
+ * U12（R05）：阶段 A 关键依赖的自定义健康指示器——
+ * Redis（flightInfo 权威存储）与 Kafka（投递端口）。依赖经 BeanProvider 可选解析：
+ * 缺 bean（如未用 stub 也未实装）时指示 DOWN 而非启动失败。
+ */
+@Singleton
+class FlightRedisHealthIndicator(
+    private val redis: BeanProvider<FlightRedisClient>,
+) : HealthIndicator {
+
+    override fun getResult(): Publisher<HealthResult> =
+        Publishers.just(resultOf("redis-flight-store", "flight store", redis.isPresent,
+            redis.isPresent.takeIf { it }?.let { redis.get()::class.java.simpleName }))
+}
+
+@Singleton
+class KafkaDeliveryHealthIndicator(
+    private val port: BeanProvider<DeliveryPort>,
+) : HealthIndicator {
+
+    override fun getResult(): Publisher<HealthResult> =
+        Publishers.just(resultOf("kafka-delivery", "delivery port", port.isPresent,
+            port.isPresent.takeIf { it }?.let { port.get()::class.java.simpleName }))
+}
+
+private fun resultOf(name: String, what: String, up: Boolean, impl: String?): HealthResult {
+    val builder = HealthResult.builder(name)
+        .status(if (up) HealthStatus.UP else HealthStatus.DOWN)
+        .details(
+            mapOf(
+                "message" to if (up) "$what present" else "$what bean missing (stub off, impl pending)",
+                "implementation" to (impl ?: "none"),
+            ),
+        )
+    return builder.build()
+}

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/log/TraceLog.kt

@@ -0,0 +1,18 @@
+package com.gzzn.omms.msgexchange.nextgen.infra.log
+
+import org.slf4j.MDC
+
+/**
+ * U12（R05）：处理路径入口写入 MDC traceId（=cminmsgsId/eventId），
+ * 使一条消息全链路日志可串（logback %X{traceId} + logstash includeMdcKeyName）。
+ */
+object TraceLog {
+    fun <T> withTrace(id: Any, body: () -> T): T {
+        MDC.put("traceId", id.toString())
+        return try {
+            body()
+        } finally {
+            MDC.remove("traceId")
+        }
+    }
+}

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ReplayService.kt

@@ -12,6 +12,7 @@ import jakarta.inject.Singleton
 class ReplayService(
     private val procState: ProcStateRepository,
 ) {
+    private val log = org.slf4j.LoggerFactory.getLogger(ReplayService::class.java)
     /** 可恢复错误类：codec 修复可重放 / 未实装补齐可重放 / 基础设施抖动可重放 / 重试耗尽后人工复核可重放。 */
     val replayableErrorClasses: Set<ErrorClass> =
         setOf(ErrorClass.CODEC_ERROR, ErrorClass.UNSUPPORTED, ErrorClass.INFRA, ErrorClass.EXHAUSTED)
@@ -19,8 +20,13 @@ class ReplayService(
     /** 只重放白名单内的类；请求含 MALFORMED 等非法类时静默忽略该类。 */
     fun replay(requested: Collection<ErrorClass>): Int {
         val allowed = requested.filter { it in replayableErrorClasses }
-        if (allowed.isEmpty()) return 0
-        return procState.requeueByErrorClasses(allowed)
+        if (allowed.isEmpty()) {
+            log.warn("replay requested only non-replayable classes: {}", requested)
+            return 0
+        }
+        val n = procState.requeueByErrorClasses(allowed)
+        log.info("replayed rows={} classes={}", n, allowed)
+        return n
     }
 
     /** 默认入口：重放全部可恢复类。 */

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt

@@ -13,12 +13,15 @@ class InboxService(
     // TODO(阶段1后续): 事务边界（@Transactional）随 Micronaut Data 实装补齐；
     // pump 唤醒仅加速，崩溃后主泵 1s 轮询兜底。
 ) {
+    private val log = org.slf4j.LoggerFactory.getLogger(InboxService::class.java)
+
     data class Receipt(val cminmsgsId: Long, val receivedAt: Instant)
 
     fun accept(rawXml: String): Receipt {
         val id = inbox.insertRaw(rawXml)
         procState.insert(id)                 // 同事务（实装后）；接收层无唯一约束（I3）
         wakePump()
+        log.info("accepted cminmsgsId={} (tx1)", id)
         return Receipt(id, Instant.now())
     }
 

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt

@@ -31,6 +31,8 @@ class Pump(
     private val jobExecutor: JobExecutor,
     private val props: PipelineProps,
 ) {
+    private val log = org.slf4j.LoggerFactory.getLogger(Pump::class.java)
+
     @Volatile
     private var running = true
 
@@ -65,6 +67,7 @@ class Pump(
             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 {
@@ -116,7 +119,10 @@ class MessageProcessor(
     private val procFailure: ProcFailure,
     private val props: PipelineProps,
 ) {
+    private val log = org.slf4j.LoggerFactory.getLogger(MessageProcessor::class.java)
+
     fun processOne(head: ProcState) {
+        com.gzzn.omms.msgexchange.nextgen.infra.log.TraceLog.withTrace(head.cminmsgsId) {
             try {
                 processInternal(head)
             } catch (e: InterruptedException) {
@@ -124,19 +130,23 @@ class MessageProcessor(
                 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
         }
@@ -145,9 +155,11 @@ class MessageProcessor(
             is com.gzzn.omms.msgexchange.nextgen.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
@@ -159,6 +171,7 @@ class MessageProcessor(
             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
             }
@@ -168,6 +181,7 @@ class MessageProcessor(
         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
         }
@@ -192,6 +206,7 @@ class MessageProcessor(
         msgEvents.insertAll(events)
         inbox.backfillOnSuccess(head.cminmsgsId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn)
         procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED)
+        log.info("SUCCEEDED id={} events={}", head.cminmsgsId, events.size)
         // 阶段 B：flightState.apply(decision.flightChanges) 进入同一事务；投影事件（ES/REDIS）追加。
     }
 }

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/SnapshotFlow.kt

@@ -25,10 +25,13 @@ class SnapshotFlow(
     private val redis: FlightRedisClient,
     private val procFailure: ProcFailure,
 ) {
+    private val log = org.slf4j.LoggerFactory.getLogger(SnapshotFlow::class.java)
+
     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) {
+            log.warn("staging not implemented -> FAILED(UNSUPPORTED) id={} reason={}", head.cminmsgsId, staged.reason)
             procFailure.fail(head, ErrorClass.UNSUPPORTED, staged.reason)   // U10：未实装 → 可重放，非终态
             return
         }
@@ -47,11 +50,13 @@ class SnapshotFlow(
             // 重放路径：version 已是目标值 → no-op 视为成功
             val again = refData.getGen(day)
             if (again == null || again.version != expected + 1) {
+                log.warn("gen CAS conflict -> FAILED(INFRA) id={}", head.cminmsgsId)
                 procFailure.fail(head, ErrorClass.INFRA, "gen-cas-conflict")   // N06/N28：带退避，禁止紧循环
                 return
             }
         }
         procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED)
+        log.info("snapshot SUCCEEDED id={} day={} flights={}", head.cminmsgsId, day, normalized.size)
     }
 
     /** staging 结果（骨架）。 */

src/main/resources/logback.xml

@@ -1,6 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!-- ACMA-8/ACMA-6：logstash TCP JSON 通道保留（字段兼容现网）+ MDC traceId -->
-<!-- U03/N13：logback 直接读环境变量（前缀 MSGX_LOGSTASH_*），与 application.yml 不再双定义 -->
+<!-- U12/N31b（顺序约束）：先加 Async + neverBlock 降级，再补热路径日志——
+     否则 logstash 不可达时同步 TCP 发送会反向阻塞业务线程 -->
 <configuration>
     <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
         <encoder>
@@ -8,17 +9,27 @@
         </encoder>
     </appender>
 
-    <appender name="LOGSTASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
+    <appender name="LOGSTASH_SYNC" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
         <destination>${MSGX_LOGSTASH_HOST:-127.0.0.1}:${MSGX_LOGSTASH_PORT:-5044}</destination>
         <encoder class="net.logstash.logback.encoder.LogstashEncoder">
             <includeMdcKeyName>traceId</includeMdcKeyName>
+            <includeMdcKeyName>target</includeMdcKeyName>
         </encoder>
         <reconnectionDelay>10 seconds</reconnectionDelay>
     </appender>
 
+    <!-- U12：异步 + neverBlock——队列满即丢弃（降级优先保业务线程），不阻塞热路径 -->
+    <appender name="ASYNC_LOGSTASH" class="ch.qos.logback.classic.AsyncAppender">
+        <queueSize>4096</queueSize>
+        <discardingThreshold>0</discardingThreshold>
+        <neverBlock>true</neverBlock>
+        <includeCallerData>false</includeCallerData>
+        <appender-ref ref="LOGSTASH_SYNC"/>
+    </appender>
+
     <root level="INFO">
         <appender-ref ref="STDOUT"/>
-        <appender-ref ref="LOGSTASH"/>
+        <appender-ref ref="ASYNC_LOGSTASH"/>
     </root>
 
     <logger name="com.gzzn.omms.msgexchange.nextgen" level="DEBUG"/>