feat(metrics): 投递失败与死信可观测指标，修 flushSchd 首次失败原因丢失（ACM2-98）

- PipelineCounters 按 target 累计发送失败；Dispatcher.retryOrDead 记账（OPS-2）
- 新增 deadByTarget 查询与 DeliveryDeadSnapshotProvider（TTL 缓存），死信存量按 target 暴露
- /metrics 注册 send_failures.total{target} 与 dead{target}，登记进 reference.md「指标与健康」
- flushSchd 失败改记真实异常原因，不再退回固定 send-failed

docs/reference.md

@@ -95,6 +95,8 @@
 | `msgx.pipeline.codec.srvt_seen.total` | 收到含 `SRVT` 段的消息数 | 非零：核对 `G-SRVT-VIPF` |
 | `msgx.pipeline.codec.vipf_seen.total` | 收到含 `VIPF` 段的消息数 | 非零：核对 `G-SRVT-VIPF` |
 | `msgx.pipeline.processing.ignored.total` | 被 `IgnoreRules` 跳过的消息数 | 清单含 `REGN`、`RSTA`、`EROR`，与 `US-13`、`US-09` 冲突 |
+| `msgx.pipeline.delivery.send_failures.total{target}` | 该投递目标累计发送失败次数（进程内，重启归零） | 持续增长且 `dead` 非零：投递链路故障（`OPS-2`） |
+| `msgx.pipeline.delivery.dead{target}` | 该投递目标当前死信（`DEAD`）行数 | 非零即告警：死信需人工处置（`OPS-2`；状态语义见 [implementation.md](implementation.md)「状态与错误分类」） |
 | `msgx.pipeline.watermark.lag` | 信箱最新编号比已扫描编号大多少 | 改用处理时间筛选后删除（`G-SCAN-PREDICATE`） |
 
 未处理消息的统计使用同一份缓存；统计不可用显示 `NaN`，没有记录可比时年龄与编号差显示 `-1`。作业计数在进程重启后归零；作业已启动却连续三个检查周期没有成功运行时，`/health` 报 `DOWN`。

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

@@ -45,6 +45,7 @@ class Dispatcher(
     private val port: DeliveryPort,
     private val props: PipelineProps,
     private val scheduler: FailureScheduler,
+    private val counters: com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters,
 ) {
     private val log = org.slf4j.LoggerFactory.getLogger(Dispatcher::class.java)
 
@@ -162,7 +163,7 @@ class Dispatcher(
             lastFlush = scheduler.now()
             return
         }
-        val failures = mutableListOf<MsgEvent>()
+        val failures = mutableListOf<Pair<MsgEvent, String>>()
         for (e in batch) {
             try {
                 when (e.eventType) {
@@ -172,15 +173,17 @@ class Dispatcher(
                 // 条件确认：读取时刻的代次（EVENT_ID + STATE_VERSION）被新写入覆盖时不标记，留待下一轮重发。
                 e.eventId?.let { msgEvents.markSentIfVersion(it, e.stateVersion, scheduler.now()) }
             } catch (ex: Exception) {
-                failures.add(e)
+                // 首次失败就带上真实原因；不能退回字面量把根因抹掉
+                failures += e to (ex.message ?: ex.javaClass.simpleName)
             }
         }
-        failures.forEach { retryOrDead(it, it.lastError ?: "send-failed") }
+        failures.forEach { (e, error) -> retryOrDead(e, error) }
         lastFlush = scheduler.now()
     }
 
     /** 一条事件发失败之后怎么走：重试次数加一，到上限就标成 DEAD(EXHAUSTED) 留作死信（次数落库便于追查），否则按退避推到下次再发。 */
     private fun retryOrDead(e: MsgEvent, lastError: String) {
+        counters.sendFailedAdd(e.target)   // OPS-2：发送失败按 target 计数，供 /metrics 告警
         val eventId = e.eventId ?: return
         val attempts = e.attempts + 1
         if (scheduler.exhausted(attempts)) {

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

@@ -0,0 +1,44 @@
+package com.gzzn.omms.msgexchange.infra.health
+
+import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository
+import io.micronaut.context.BeanProvider
+import io.micronaut.context.annotation.Value
+import jakarta.inject.Singleton
+import java.time.Clock
+import java.time.Duration
+import java.time.Instant
+
+/**
+ * 投递死信统计（每个 target 的 `DEAD` 行数）的唯一取数点，带 TTL 缓存。
+ *
+ * 与 [BacklogSnapshotProvider] 同一设计：`/metrics` 抓取可能高频发生，
+ * `GROUP BY target` 聚合不该每次直打数据库；数字最多滞后一个缓存窗口，对告警信号足够。
+ * 缓存窗口沿用 `msgx.health.backlog-cache-ttl-ms`。
+ */
+@Singleton
+class DeliveryDeadSnapshotProvider(
+    private val msgEvents: BeanProvider<MsgEventRepository>,
+    private val clock: Clock,
+    @Value("\${msgx.health.backlog-cache-ttl-ms:30000}") private val ttlMs: Long = 30_000,
+) {
+    @Volatile
+    private var cached: Map<String, Int>? = null
+
+    @Volatile
+    private var cachedAt: Instant? = null
+
+    /** @return target → 死信行数；仓储未绑定时返回 null，由调用方按"未绑定"处理。 */
+    fun snapshot(): Map<String, Int>? {
+        if (!msgEvents.isPresent) return null
+        val now = clock.instant()
+        val at = cachedAt
+        val value = cached
+        if (ttlMs > 0 && at != null && value != null && Duration.between(at, now).toMillis() < ttlMs) {
+            return value
+        }
+        val fresh = msgEvents.get().deadByTarget()
+        cached = fresh
+        cachedAt = now
+        return fresh
+    }
+}

src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineCounters.kt

@@ -16,6 +16,7 @@ class PipelineCounters {
     private val srvtSeen = AtomicLong(0)
     private val vipfSeen = AtomicLong(0)
     private val ignored = AtomicLong(0)
+    private val sendFailures = java.util.concurrent.ConcurrentHashMap<String, AtomicLong>()
 
     /**
      * 入站记录里出现 `SRVT`/`VIPF` 段的条数（`[G-SRVT-VIPF]`）。
@@ -36,4 +37,14 @@ class PipelineCounters {
     fun ignoredAdd(count: Int = 1) { ignored.addAndGet(count.toLong()) }
 
     fun ignoredCount(): Long = ignored.get()
+
+    /**
+     * 按投递目标（`KAFKA:msg`/`KAFKA:schd`）累计的发送失败次数（`OPS-2`）。
+     * 每次记退避或死信时累加；重启归零，跨重启的累计由指标后端聚合。
+     */
+    fun sendFailedAdd(target: String) {
+        sendFailures.computeIfAbsent(target) { AtomicLong(0) }.incrementAndGet()
+    }
+
+    fun sendFailedCount(target: String): Long = sendFailures[target]?.get() ?: 0L
 }

src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetrics.kt

@@ -1,6 +1,7 @@
 package com.gzzn.omms.msgexchange.infra.metrics
 
 import com.gzzn.omms.msgexchange.infra.health.BacklogSnapshotProvider
+import com.gzzn.omms.msgexchange.infra.health.DeliveryDeadSnapshotProvider
 import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository
 import com.gzzn.omms.msgexchange.infra.persistence.InboxCursorRepository
 import io.micronaut.context.BeanProvider
@@ -29,6 +30,8 @@ import java.time.Duration
  * - `msgx.pipeline.codec.srvt_seen.total` / `msgx.pipeline.codec.vipf_seen.total`：入站记录里出现
  *   `SRVT`/`VIPF` 段的条数（尚未落明细表，`[G-SRVT-VIPF]`；> 0 表示真实流量确有该段）
  * - `msgx.pipeline.processing.ignored.total`：命中 US-04 忽略清单的报文条数
+ * - `msgx.pipeline.delivery.send_failures.total{target}`：按 target 的投递发送失败次数（`OPS-2`）
+ * - `msgx.pipeline.delivery.dead{target}`：按 target 的死信（`DEAD`）存量行数（`OPS-2`，非零需人工处置）
  *
  * 取数统一走 [BacklogSnapshotProvider]（30 秒 TTL），因此指标抓取不会打穿数据库。
  * 无法取数时以 `NaN` 上报（Micrometer 的惯例表示"本次无值"），而不是伪造 0。
@@ -44,6 +47,7 @@ import java.time.Duration
 class PipelineMetrics(
     private val registry: MeterRegistry,
     private val backlogs: BacklogSnapshotProvider,
+    private val deadEvents: DeliveryDeadSnapshotProvider,
     private val cursor: BeanProvider<InboxCursorRepository>,
     private val mailbox: BeanProvider<CminmsgInboxRepository>,
     private val activity: JobActivity,
@@ -106,6 +110,24 @@ class PipelineMetrics(
         Gauge.builder("msgx.pipeline.processing.ignored.total", counters) { it.ignoredCount().toDouble() }
             .strongReference(true)
             .register(registry)
+
+        // OPS-2：投递失败可观测。发送失败按 target 计数（进程内，重启归零）；
+        // 死信存量按 target 从库端聚合取（带 TTL 缓存）。非零死信是人工处置信号。
+        for (target in listOf(com.gzzn.omms.msgexchange.domain.Targets.KAFKA_MSG, com.gzzn.omms.msgexchange.domain.Targets.KAFKA_SCHD)) {
+            Gauge.builder("msgx.pipeline.delivery.send_failures.total", counters) { it.sendFailedCount(target).toDouble() }
+                .tag("target", target)
+                .strongReference(true)
+                .register(registry)
+
+            Gauge.builder("msgx.pipeline.delivery.dead", deadEvents) { provider ->
+                val dead = provider.snapshot()
+                // 仓储未绑定按无值（NaN）上报；已绑定但该 target 无死信行是确定的 0
+                if (dead == null) Double.NaN else (dead[target] ?: 0).toDouble()
+            }
+                .tag("target", target)
+                .strongReference(true)
+                .register(registry)
+        }
     }
 
     private fun backlogGauge(name: String, value: (com.gzzn.omms.msgexchange.infra.persistence.Backlog) -> Double) {

src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt

@@ -186,6 +186,9 @@ interface MsgEventRepository {
 
     fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String, attempts: Int? = null)
 
+    /** 死信统计：每个 target 的 `STATE='DEAD'` 行数（投递告警信号，`OPS-2`）。 */
+    fun deadByTarget(): Map<String, Int>
+
     /**
      * 删除已确认投递且超过保留期的事件行：`STATE='SENT' AND SENT_AT < cutoff`。
      * 带 `STATE='SENT'` 条件，避免与 upsertSchd 的并发写入冲突。

src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt

@@ -562,6 +562,13 @@ class JdbcMsgEventRepository(
         )
     }
 
+    override fun deadByTarget(): Map<String, Int> =
+        ds.query(
+            "SELECT target, count(*) AS total FROM msg_event WHERE state = 'DEAD' GROUP BY target",
+            { /* 无绑定参数 */ },
+            { rs -> rs.getString("target") to rs.getInt("total") },
+        ).toMap()
+
     override fun deleteExpiredSent(cutoff: Instant, limit: Int): Int =
         ds.update(
             """

src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt

@@ -304,6 +304,9 @@ class StubMsgEvents : MsgEventRepository {
         )
     }
 
+    override fun deadByTarget(): Map<String, Int> =
+        rows.values.filter { it.state == EventStatus.DEAD }.groupingBy { it.target }.eachCount()
+
     override fun deleteExpiredSent(cutoff: Instant, limit: Int): Int {
         val ids = rows.values
             .filter { it.state == EventStatus.SENT && it.sentAt != null && it.sentAt < cutoff }

src/test/kotlin/com/gzzn/omms/msgexchange/delivery/DispatcherTickTest.kt

@@ -8,6 +8,7 @@ import com.gzzn.omms.msgexchange.domain.EventType
 import com.gzzn.omms.msgexchange.domain.MsgEvent
 import com.gzzn.omms.msgexchange.domain.Targets
 import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository
+import com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters
 import com.gzzn.omms.msgexchange.infra.retry.FailureScheduler
 import com.gzzn.omms.msgexchange.infra.stub.StubDeliveryPort
 import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents
@@ -28,8 +29,12 @@ class DispatcherTickTest {
 
     private val clock = MutableClock(MutableClock.BASE)
 
-    private fun dispatcher(repo: MsgEventRepository, port: DeliveryPort, p: PipelineProps = PipelineProps()): Dispatcher =
-        Dispatcher(repo, port, p, FailureScheduler(p, clock))
+    private fun dispatcher(
+        repo: MsgEventRepository,
+        port: DeliveryPort,
+        p: PipelineProps = PipelineProps(),
+        counters: PipelineCounters = PipelineCounters(),
+    ): Dispatcher = Dispatcher(repo, port, p, FailureScheduler(p, clock), counters)
 
     private fun ev(id: Long, target: String, key: String, payload: String, version: Long = 0) =
         MsgEvent(eventId = id, target = target, partitionKey = key, stateVersion = version, payloadJson = payload, createdAt = MutableClock.BASE)
@@ -123,11 +128,14 @@ class DispatcherTickTest {
     fun `schd send failure schedules backoff and goes DEAD at limit`() {
         val repo = StubMsgEvents()
         val p = PipelineProps()
-        val d = dispatcher(repo, FailingSchdPort(), p)
+        val counters = PipelineCounters()
+        val d = dispatcher(repo, FailingSchdPort(), p, counters)
         repo.insertAll(listOf(ev(1, Targets.KAFKA_SCHD, "F1", """{"v":"1"}""", 1)))
 
         d.flushSchd()
         assertEquals(1, repo.rows[1]!!.attempts)
+        // OPS-2：发送失败按 target 计数，且失败原因不是固定字面量
+        assertEquals(1L, counters.sendFailedCount(Targets.KAFKA_SCHD))
 
         repeat(p.pipeline.maxAttempts - 1) {
             clock.advance(p.pipeline.backoffFor(repo.rows[1]!!.attempts) + 1)
@@ -135,6 +143,7 @@ class DispatcherTickTest {
         }
         val final = repo.rows[1]!!
         assertTrue(final.attempts >= p.pipeline.maxAttempts || final.state.name == "DEAD")
+        assertEquals(0L, counters.sendFailedCount(Targets.KAFKA_MSG))   // msg 目标不受 schd 失败影响
     }
 
     @Test

src/test/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetricsTest.kt

@@ -1,6 +1,10 @@
 package com.gzzn.omms.msgexchange.infra.metrics
 
+import com.gzzn.omms.msgexchange.domain.ErrorClass
+import com.gzzn.omms.msgexchange.domain.MsgEvent
 import com.gzzn.omms.msgexchange.domain.ProcStatus
+import com.gzzn.omms.msgexchange.domain.Targets
+import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents
 import com.gzzn.omms.msgexchange.infra.stub.StubProcState
 import io.micrometer.core.instrument.MeterRegistry
 import io.micronaut.context.ApplicationContext
@@ -74,6 +78,34 @@ class PipelineMetricsTest {
         assertEquals(vipfBefore + 1.0, gauge("msgx.pipeline.codec.vipf_seen.total"), 0.001)
     }
 
+    @Test
+    fun `delivery send-failure and dead-letter gauges are registered per target`() {
+        val counters = ctx.getBean(PipelineCounters::class.java)
+        val events = ctx.getBean(StubMsgEvents::class.java)
+        val failuresBefore = taggedGauge("msgx.pipeline.delivery.send_failures.total", Targets.KAFKA_SCHD)
+
+        counters.sendFailedAdd(Targets.KAFKA_SCHD)
+        val ids = events.insertAll(
+            listOf(
+                MsgEvent(target = Targets.KAFKA_MSG, partitionKey = "F1", stateVersion = 1, payloadJson = "{}", createdAt = Instant.now()),
+            ),
+        )
+        events.markDead(ids.single(), ErrorClass.EXHAUSTED, "boom", attempts = 5)
+
+        assertEquals(
+            failuresBefore + 1.0,
+            taggedGauge("msgx.pipeline.delivery.send_failures.total", Targets.KAFKA_SCHD),
+            0.001,
+        )
+        assertEquals(1.0, taggedGauge("msgx.pipeline.delivery.dead", Targets.KAFKA_MSG), 0.001)
+        assertEquals(0.0, taggedGauge("msgx.pipeline.delivery.dead", Targets.KAFKA_SCHD), 0.001)
+    }
+
     private fun gauge(name: String): Double =
         requireNotNull(registry.find(name).gauge()) { "gauge not registered: $name" }.value()
+
+    private fun taggedGauge(name: String, target: String): Double =
+        requireNotNull(registry.find(name).tag("target", target).gauge()) {
+            "gauge not registered: $name(target=$target)"
+        }.value()
 }