fix(pipeline): 修复取报-处理-回填链路的超时、吞吐、回填与 FIFO 问题

外部调用（ACM2-33）
- 共享 MySQL 与自有 PG 全部补有界超时：驱动 connect/socket 超时 + Hikari 池超时
  （注意 Micronaut 的 Hikari 项是毫秒数，不是 Duration 字面量）
- 删除主泵/毒丸路径的 inline 回填：跨库写不再占用 FIFO 关键路径，回填统一由扫描驱动
- 游标自愈：save() 改为「先 UPDATE、缺行 INSERT」，缺行只记一次 ERROR
- Pump/Dispatcher 外层 catch 补 error 日志与失败计数

投递吞吐（ACM2-34）
- Dispatcher 改批量领取；队头退避未到期或首条发送失败即停止本轮（保持目标内保序）
- 有活不再 sleep；删除不可达的 state==SENT 死条件；markSent 移入分支；
  superseded 清理加轮数上限与空 eventId 保护

回填闭环（ACM2-36，V4）
- MISSING（运行时确认行不存在）立即放弃自动重试并告警；暂时性故障达上限后停止自动重试
- 新增 reopen 人工恢复入口；放弃 ≠ 标记已确认（BACKFILL_AT 仍为空，清除前提不成立）
- 扫描改 (BACKFILL_ATTEMPTS, MSG_ID) 公平轮转并排除放弃行，消除全局回填饥饿
- V4 只加字段与必要索引，不按年龄做任何存量推断

切流播种（ACM2-35，V5）
- cutover-watermark 四模式（min/zero/max/显式 ID），默认不播种、代码不做默认选择
- 升级实例拒绝重新播种（SEEDED_AT 为 NULL ≠ 从未消费）；播种与水位同语句落库
- 非法取值由启动自检挡下

错误分类与入口契约（ACM2-37）
- 未知 SCHD 子类型改为 UNSUPPORTED，不再静默当全量日计划合并
- ADFT 运营日冲突改走 ProtocolViolation → DEAD(PROTOCOL)
- 兼容入口 receivedAt 缺失回退到注入 Clock；MessageLifecycleGate 强制注入 + 装配断言
- FIFO：主泵只领取 msgId ≤ W，兼容入口登记的行在水位追平前不被领取

时间源与可观测（ACM2-38 / ACM2-41 阶段 0）
- 仓储/处理器/作业全部经注入 Clock；移除 markTerminal/markBackfilled 的 Instant.now() 默认值
- 退避表档位与 max-attempts 对齐并加启动自检
- Micrometer 7 个 gauge（@Context 急切注册）+ /health 与 /metrics 共用积压快照缓存
- 只读迟到检测：监视被放行的空洞 ID 是否后来真的出现，只计数告警、不补入队

Plane: ACM2-33 ACM2-34 ACM2-35 ACM2-36 ACM2-37 ACM2-38 ACM2-41
Tests: 88 → 120（1 skipped 需真实 PG）

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

@@ -11,6 +11,7 @@ import com.gzzn.omms.msgexchange.infra.retry.FailureScheduler
 import jakarta.inject.Singleton
 import java.time.Duration
 import java.time.Instant
+import java.util.concurrent.atomic.AtomicLong
 
 /** 对外投递的出口：同步等 Kafka 确认，语义是至少发一次（可能重复，但不会丢）。以后 ES 投影也从这里加。 */
 interface DeliveryPort {
@@ -46,6 +47,9 @@ class Dispatcher(
 ) {
     private val log = org.slf4j.LoggerFactory.getLogger(Dispatcher::class.java)
 
+    /** 连续失败计数：仅用于日志/排障。 */
+    private val tickFailures = AtomicLong(0)
+
     @Volatile
     private var running = true
 
@@ -60,35 +64,77 @@ class Dispatcher(
         while (running) {
             try {
                 tick()
+                tickFailures.set(0)
             } catch (e: InterruptedException) {
                 Thread.currentThread().interrupt()
                 return
             } catch (e: Exception) {
+                // tick 整体失败（DB 断连等）必须在别处留痕，否则只剩"投递不动"没有原因。
+                val failures = tickFailures.incrementAndGet()
+                log.error("dispatcher tick failed (failure #{})", failures, e)
                 sleepQuietly(props.pipeline.pollInterval)
             }
         }
     }
 
     internal fun tick() {
-        val head = msgEvents.headUnsent(Targets.KAFKA_MSG)
-        if (head != null && (head.state == EventStatus.SENT || head.nextAttemptAt == null || head.nextAttemptAt <= scheduler.now())) {
-            deliver(Targets.KAFKA_MSG, head)
-        }
+        val sent = drainKafkaMsg()
         if (flushDue()) flushSchd()
-        if (running) sleepQuietly(props.pipeline.pollInterval)
+        // 有活就不睡：pollInterval 只用于"本轮无事可做或队头在退避"，不能当投递节流用。
+        if (sent == 0 && running) sleepQuietly(props.pipeline.pollInterval)
+    }
+
+    /**
+     * 按 `EVENT_ID` 顺序批量投递 `KAFKA:msg`。
+     *
+     * 保序规则不变：**同一目标内不越序**——队头失败或退避未到期时立即停止本轮，
+     * 不跳过它去投后面的。批量只用来省掉"每条一次 DB 往返 + 一轮一次 sleep"。
+     *
+     * @return 本轮成功发出的条数
+     */
+    private fun drainKafkaMsg(): Int {
+        val batchSize = props.pipeline.deliveryBatch.coerceAtLeast(1)
+        val maxRounds = props.pipeline.deliveryDrainRounds.coerceAtLeast(1)
+        var sent = 0
+        var round = 0
+        while (round < maxRounds) {
+            round++
+            val batch = try {
+                msgEvents.claimBatch(Targets.KAFKA_MSG, batchSize)
+            } catch (e: Exception) {
+                log.error("claimBatch(KAFKA_MSG) failed", e)
+                return sent
+            }
+            if (batch.isEmpty()) return sent
+            for (e in batch) {
+                val next = e.nextAttemptAt
+                if (next != null && next > scheduler.now()) return sent   // 队头退避未到期：停止推进
+                if (!deliver(Targets.KAFKA_MSG, e)) return sent            // 失败已记账：停止以保序
+                sent++
+            }
+            if (batch.size < batchSize) return sent
+        }
+        log.warn("kafka msg drain hit round cap ({} rounds, sent={}); yielding to schd flush", maxRounds, sent)
+        return sent
     }
 
     private fun flushDue(): Boolean =
         lastFlush?.let { Duration.between(it, scheduler.now()) >= props.schd.flushPeriod } ?: true
 
-    private fun deliver(target: String, e: MsgEvent) {
-        try {
+    /** @return 是否已确认发出（false 表示已按退避/死信记账，调用方应停止本轮以保序） */
+    private fun deliver(target: String, e: MsgEvent): Boolean {
+        val eventId = e.eventId ?: return false
+        return try {
             when (target) {
                 Targets.KAFKA_MSG -> port.sendKafka("msg", e.partitionKey, e.payloadJson)
+                // 未知目标不能在"未发送"的情况下被标记为已发；记账后停止本轮。
+                else -> error("unknown delivery target: $target")
             }
-            msgEvents.markSent(e.eventId ?: return)
+            msgEvents.markSent(eventId)
+            true
         } catch (ex: Exception) {
             retryOrDead(e, ex.message ?: ex.javaClass.simpleName)
+            false
         }
     }
 
@@ -117,14 +163,26 @@ class Dispatcher(
         }
         val sentIds = batch.mapNotNull { it.eventId }.toSet() - failures.mapNotNull { it.eventId }.toSet()
         if (sentIds.isNotEmpty()) msgEvents.markAllSent(sentIds.toList())
-        // 被更新版本压掉的旧事件也要标成已发，否则它们会一直留在队里：同一个 FLID 只按最新版本输出一次
         val sentVersions = batch.associate { it.partitionKey to it.stateVersion }
+        // 被更新版本压掉的旧事件也要标成已发，否则它们会一直留在队里：同一个 FLID 只按最新版本输出一次。
+        // 这里必须**有界**：若 markAllSent 因任何原因没有生效（例如行缺 eventId），
+        // 旧的无界 while(true) 会原地空转。
         runCatching {
-            while (true) {
+            var rounds = 0
+            while (rounds < SUPERSEDED_CLEANUP_MAX_ROUNDS) {
+                rounds++
                 val superseded = msgEvents.mergePendingSchd(scheduler.now(), props.schd.flushLimit)
                     .filter { sentVersions[it.partitionKey]?.let { v -> it.stateVersion < v } == true }
                 if (superseded.isEmpty()) break
-                msgEvents.markAllSent(superseded.mapNotNull { it.eventId })
+                val ids = superseded.mapNotNull { it.eventId }
+                if (ids.isEmpty()) {
+                    log.warn("superseded cleanup: {} rows without event_id, stop to avoid a spin", superseded.size)
+                    break
+                }
+                msgEvents.markAllSent(ids)
+                if (rounds == SUPERSEDED_CLEANUP_MAX_ROUNDS) {
+                    log.warn("superseded cleanup hit round cap ({}); remaining rows will be handled next flush", rounds)
+                }
             }
         }.onFailure { log.warn("superseded cleanup failed: {}", it.message) }
         failures.forEach { retryOrDead(it, it.lastError ?: "send-failed") }
@@ -145,4 +203,9 @@ class Dispatcher(
     private fun sleepQuietly(d: Duration) {
         if (!d.isNegative && !d.isZero) Thread.sleep(d.toMillis().coerceAtLeast(1))
     }
+
+    private companion object {
+        /** superseded 清理的轮数上限：只用于防止"标不掉又不报错"时的原地空转。 */
+        const val SUPERSEDED_CLEANUP_MAX_ROUNDS = 100
+    }
 }