feat(delivery): 真实 Kafka 投递端口 + 闭合 G-BACKFILL-BACKOFF 配置缺口

- 新增 KafkaDeliveryPort：ProducerRegistry + 同步 Future.get() 至少一次投递，
  闭合 OPS-1 真实适配器要求与 US-07 INV-10/C-29
- 回填退避从硬编码 30s/15min 提升为 backfill-backoff-ms / backfill-backoff-cap-ms
  配置绑定，闭合 [G-BACKFILL-BACKOFF]
- 删除死依赖 micronaut-redis-lettuce（代码零引用，ACM2-28 Redis 已退出阶段 A）

src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt

@@ -54,6 +54,12 @@ class PipelineProps {
          */
         var backfillMaxAttempts: Int = 100
 
+        /** 回填独立退避的起步间隔 `[G-BACKFILL-BACKOFF]`。 */
+        var backfillBackoffMs: Long = 30_000
+
+        /** 回填独立退避的封顶间隔 `[G-BACKFILL-BACKOFF]`。 */
+        var backfillBackoffCapMs: Long = 900_000
+
         /**
          * 切流水位播种（一次性、显式）。取值：
          * `min` = `W:MIN(ID)−1`（读当前全部现存行）、`zero` = `W:0`（按空洞规则从 0 扫）、

src/main/kotlin/com/gzzn/omms/msgexchange/infra/kafka/KafkaDeliveryPort.kt

@@ -0,0 +1,46 @@
+package com.gzzn.omms.msgexchange.infra.kafka
+
+import com.gzzn.omms.msgexchange.delivery.DeliveryPort
+import io.micronaut.configuration.kafka.ProducerRegistry
+import io.micronaut.context.annotation.Requires
+import io.micronaut.core.type.Argument
+import jakarta.inject.Singleton
+import org.apache.kafka.clients.producer.Producer
+import org.apache.kafka.clients.producer.ProducerRecord
+
+/**
+ * Kafka 真实投递端口：通过 [ProducerRegistry] 拿 default 生产者，D3 参数
+ * （acks=all / enable-idempotence=true / max-in-flight=1）由 `KafkaD3Check` 启动自检。
+ *
+ * 发送同步等 broker 确认（`Future.get()`），语义是至少一次——满足 INV-10 / C-29。
+ * 仅在 `msgx.stubs != true` 时装配；stub 模式走 `StubDeliveryPort`。
+ */
+@Requires(property = "msgx.stubs", notEquals = "true")
+@Singleton
+class KafkaDeliveryPort(
+    private val registry: ProducerRegistry,
+) : DeliveryPort {
+
+    private val producer: Producer<String, String> by lazy {
+        registry.getProducer("default", Argument.of(String::class.java), Argument.of(String::class.java))
+    }
+
+    override fun sendKafka(topic: String, key: String, payloadJson: String) {
+        producer.send(ProducerRecord(topic, key, payloadJson)).get()
+    }
+
+    override fun sendKafkaSchd(topic: String, key: String, payloadJson: String) {
+        producer.send(ProducerRecord(topic, key, payloadJson)).get()
+    }
+
+    override fun sendKafkaNull(topic: String, key: String) {
+        producer.send(ProducerRecord<String, String>(topic, key, null)).get()
+    }
+
+    override fun ping(): Boolean = try {
+        producer.partitionsFor("msg")
+        true
+    } catch (_: Exception) {
+        false
+    }
+}

src/main/kotlin/com/gzzn/omms/msgexchange/processing/BackfillService.kt

@@ -42,8 +42,6 @@ class BackfillService(
     private val log = org.slf4j.LoggerFactory.getLogger(BackfillService::class.java)
 
     companion object {
-        private val INITIAL_BACKOFF: Duration = Duration.ofSeconds(30)
-        private val MAX_BACKOFF: Duration = Duration.ofMinutes(15)
         private val TERMINAL_STATES = setOf(ProcStatus.SUCCEEDED, ProcStatus.SKIPPED, ProcStatus.DEAD)
 
         /** 放弃原因：运行时查询确认信箱行不存在（确定性结论，重试不会改变结果）。 */
@@ -51,11 +49,13 @@ class BackfillService(
 
         /** 放弃原因：暂时性故障持续到 `R` 仍未打标；停止自动重试，但保留人工恢复能力。 */
         const val ABANDON_TRANSIENT_DEADLINE = "TRANSIENT_DEADLINE"
+    }
 
-        fun backoffDelayFor(attempts: Int): Duration {
-            val shift = (attempts - 1).coerceIn(0, 20)
-            return INITIAL_BACKOFF.multipliedBy(1L shl shift).coerceAtMost(MAX_BACKOFF)
-        }
+    internal fun backoffDelayFor(attempts: Int): Duration {
+        val initial = Duration.ofMillis(props.pipeline.backfillBackoffMs)
+        val cap = Duration.ofMillis(props.pipeline.backfillBackoffCapMs)
+        val shift = (attempts - 1).coerceIn(0, 20)
+        return initial.multipliedBy(1L shl shift).coerceAtMost(cap)
     }
 
     /**

src/test/kotlin/com/gzzn/omms/msgexchange/processing/BackfillServiceTest.kt

@@ -221,10 +221,11 @@ class BackfillServiceTest {
 
     @Test
     fun `backoff delay doubles per attempt and caps at fifteen minutes`() {
-        assertEquals(Duration.ofSeconds(30), BackfillService.backoffDelayFor(1))
-        assertEquals(Duration.ofMinutes(2), BackfillService.backoffDelayFor(3))
-        assertEquals(Duration.ofMinutes(15), BackfillService.backoffDelayFor(10))
-        assertEquals(Duration.ofMinutes(15), BackfillService.backoffDelayFor(50))
+        val svc = BackfillService(StubProcState(), StubInbox(), MailboxProps(), props, Clock.fixed(t0, ZoneOffset.UTC), MessageLifecycleGate())
+        assertEquals(Duration.ofSeconds(30), svc.backoffDelayFor(1))
+        assertEquals(Duration.ofMinutes(2), svc.backoffDelayFor(3))
+        assertEquals(Duration.ofMinutes(15), svc.backoffDelayFor(10))
+        assertEquals(Duration.ofMinutes(15), svc.backoffDelayFor(50))
     }
 
     @Test