diff --git a/build.gradle.kts b/build.gradle.kts index 116c242..861c159 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -44,7 +44,6 @@ dependencies { implementation(libs.micronaut.data.jdbc) implementation(libs.micronaut.jdbc.hikari) implementation(libs.micronaut.flyway) - implementation(libs.micronaut.redis.lettuce) implementation(libs.micronaut.kafka) implementation(libs.micronaut.discovery.eureka) // 指标:把管道积压/回填/水位等暴露为 MeterRegistry 指标(版本由 platform BOM 管) diff --git a/docs/invariants.md b/docs/invariants.md index 6fa7187..916495c 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -113,7 +113,7 @@ | `G-PROC-HST` | `PROC_STATE_HST` 未建表,终态归档未落地 | US-11;归档能力 | | `G-FLOP-IDEMPOTENT` | 29 类 FLOP 幂等矩阵未补全 | `INV-20`、CLM-3 | | ~~`G-EVENT-RETENTION`~~ | ~~`MSG_EVENT` 已发送行的保留期与清理作业未实现~~ | 已关闭:`SENT_AT` 列 + 投递原子写 + `EventCleanupJob` 按 `eventRetention` 有界删除 | -| `G-BACKFILL-BACKOFF` | 回填独立退避键(`backfill-backoff-ms` / `-cap-ms`)未实现,当前为代码内硬编码(取值见 reference) | 回填重试节奏 | +| ~~`G-BACKFILL-BACKOFF`~~ | ~~回填独立退避键未实现,代码内硬编码~~ | 已关闭:`backfill-backoff-ms` / `backfill-backoff-cap-ms` 配置绑定落地,`BackfillService` 从 `PipelineProps` 读取 | | `G-KAFKA-D3` | ~~已闭合~~:`max-in-flight` 默认收敛到 1,启动自检钉住三项联合满足 D3 | ~~投递幂等前提~~ | | `G-REPLAY-CHANNEL` | 「打标即清除」语义下的独立原文保留通道未设计 | CLM-5 | | `G-MAFL` | 主航班 `MAFL` 派生投影及主/共享原子级联未实现(规则见 `INV-21`/`INV-22`);`MAFL` 不是 SIS/XML 入站字段 | 航班完整态;删除与重建 | diff --git a/docs/reference.md b/docs/reference.md index d429a98..eda010b 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -22,8 +22,8 @@ | `msgx.pipeline.backfill-batch` | `100` | 条 | 假定 | 回填扫描单批条数 | | `msgx.pipeline.backfill-scan-period` | `30s`(代码常量,无配置键) | Duration | 现役 | 回填扫描作业周期;批次积压与单行超时会延长实际标记延迟(`CLM-9`) | | `msgx.pipeline.backfill-max-attempts` | `100` | 次 | 假定 | 单行重试的**告警阈值**;放弃判据是 `R` 超期,不是次数(见 design「回填」) | -| `msgx.pipeline.backfill-backoff-ms` | 目标参数(未实现) | ms / 档 | 假定 | 回填独立退避表;**当前不存在**,`BackfillService` 内硬编码 30 秒起步、封顶 15 分钟 `[G-BACKFILL-BACKOFF]` | -| `msgx.pipeline.backfill-backoff-cap-ms` | 目标参数(未实现) | ms | 假定 | 回填退避封顶;对应实现是代码内常量,尚无配置键 `[G-BACKFILL-BACKOFF]` | +| `msgx.pipeline.backfill-backoff-ms` | `30000` | ms | 假定 | 回填独立退避起步间隔;`BackfillService` 指数退避的首档 `[G-BACKFILL-BACKOFF ✓]` | +| `msgx.pipeline.backfill-backoff-cap-ms` | `900000` | ms | 假定 | 回填退避封顶(15 分钟)`[G-BACKFILL-BACKOFF ✓]` | | `msgx.pipeline.cutover-watermark` | 不设置 | `min\|zero\|max\|` | 一次性运维决策 | 显式播种水位;非法值由启动自检挡下;升级实例拒绝重新播种 | | `msgx.pipeline.delivery-batch` | `200` | 条 | 假定 | `KAFKA:msg` 每轮每目标领取上限 | | `msgx.pipeline.delivery-drain-rounds` | `10` | 轮 | 假定 | 连取批数上限,让出循环跑 `schd` flush,防状态通知被积压饿死 | diff --git a/docs/user-stories.md b/docs/user-stories.md index b466039..137da14 100644 --- a/docs/user-stories.md +++ b/docs/user-stories.md @@ -190,7 +190,7 @@ 4. 重复补偿效果幂等,保留稳定的完成时间与审计;重放后的新处理结果不能被旧回填任务覆盖。非法报文缺 META 时也有明确回填方式。 5. 影子模式禁写,双跑仅一个系统持有标记写权;暴露 PG 终态、回填状态、积压、最老年龄与持续失败告警。 -**当前基础与落点**:回填意图与处理终态同体同行(`PROC_STATE.BACKFILL_*`),随业务事务提交,`BACKFILL_TODO` 已随 V2 迁移下线;终态落库后回填一律由定时扫描驱动(`BackfillService.sweep`,扫描周期与退避取值见 [reference.md](reference.md);独立退避键尚未落地 `[G-BACKFILL-BACKOFF]`),处理关键路径不做跨库写;本地入队时间(`ENQUEUED_AT`)超过超期期限 `R` 时强制补写(见 design「回填」)。死信同样可补写——回填只需消息 ID,不依赖 META。剩余:Q7 的标记值集与写权限书面确认;影子环境禁写尚未实装。 +**当前基础与落点**:回填意图与处理终态同体同行(`PROC_STATE.BACKFILL_*`),随业务事务提交,`BACKFILL_TODO` 已随 V2 迁移下线;终态落库后回填一律由定时扫描驱动(`BackfillService.sweep`,扫描周期与退避取值见 [reference.md](reference.md);独立退避键 `backfill-backoff-ms` / `backfill-backoff-cap-ms` 已落地),处理关键路径不做跨库写;本地入队时间(`ENQUEUED_AT`)超过超期期限 `R` 时强制补写(见 design「回填」)。死信同样可补写——回填只需消息 ID,不依赖 META。剩余:Q7 的标记值集与写权限书面确认;影子环境禁写尚未实装。 **前置**:US-03 终态接口;Q7、共享库更新权限。覆盖四类终态、事务回滚、重复补偿和重放竞争;生命周期与超期补写以 [design.md](design.md)「中断恢复」「回填」为准,清除口径以 [contracts.md](contracts.md)「保留与清除」为准。 diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index e55d991..c55d045 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -29,7 +29,6 @@ micronaut-jackson-databind = { module = "io.micronaut:micronaut-jackson-databind micronaut-data-jdbc = { module = "io.micronaut.data:micronaut-data-jdbc" } micronaut-jdbc-hikari = { module = "io.micronaut.sql:micronaut-jdbc-hikari" } micronaut-flyway = { module = "io.micronaut.flyway:micronaut-flyway" } -micronaut-redis-lettuce = { module = "io.micronaut.redis:micronaut-redis-lettuce" } micronaut-kafka = { module = "io.micronaut.kafka:micronaut-kafka" } micronaut-discovery-eureka = { module = "io.micronaut.discovery:micronaut-discovery-client" } micronaut-management = { module = "io.micronaut:micronaut-management" } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt index d0022a3..e969b79 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt +++ b/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 扫)、 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/kafka/KafkaDeliveryPort.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/kafka/KafkaDeliveryPort.kt new file mode 100644 index 0000000..e76bcd2 --- /dev/null +++ b/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 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(topic, key, null)).get() + } + + override fun ping(): Boolean = try { + producer.partitionsFor("msg") + true + } catch (_: Exception) { + false + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/BackfillService.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/BackfillService.kt index c0cf106..7af5a3f 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/BackfillService.kt +++ b/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) } /** diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/BackfillServiceTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/BackfillServiceTest.kt index 1243c71..a7f9144 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/BackfillServiceTest.kt +++ b/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