diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt index a6820f6..93ae612 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt @@ -31,12 +31,13 @@ interface DeliveryPort { /** * 投递调度:把 outbox(待发事件表)里的事件发给下游。 * - * KAFKA_MSG 一条一条按登记顺序发,不插队。KAFKA_SCHD 走 flushSchd:outbox 每个 FLID 只保留 + * KAFKA_MSG 按 `EVENT_ID` 顺序发,保序以 FLID 为单位(D2):某航班的队头失败或退避未到期 + * 只暂停该 FLID,其他航班照常推进。KAFKA_SCHD 走 flushSchd:outbox 每个 FLID 只保留 * 一行(写入侧单行 upsert),发出后按读取时刻的代次做条件确认;删除通知发 value 为空的 tombstone。 * 两个主题之间不保证先后顺序。 * - * 失败处理:队首的重试时间没到就不取;一批里有发送失败,整批重试次数加一并推后退避, - * 次数用尽整批转 DEAD 当死信。见 docs/implementation.md「Kafka 与读取」。 + * 失败处理:某 FLID 队首的重试时间没到就跳过该 FLID;发送失败重试次数加一并推后退避, + * 次数用尽转 DEAD 当死信。见 docs/implementation.md「Kafka 与读取」。 */ @Singleton class Dispatcher( @@ -87,8 +88,10 @@ class Dispatcher( /** * 按 `EVENT_ID` 顺序批量投递 `KAFKA:msg`。 * - * 保序规则不变:**同一目标内不越序**——队头失败或退避未到期时立即停止本轮, - * 不跳过它去投后面的。批量只用来省掉"每条一次 DB 往返 + 一轮一次 sleep"。 + * 保序以 **FLID 为单位**(D2):某航班的队头发送失败或退避未到期时,只暂停该 FLID + * 的后续事件(保持 `PENDING`、不越队),其他航班照常推进——队头卡住不再拖住全部航班。 + * 同一 FLID 内部仍然严格按 `EVENT_ID` 先后发送。批量只用来省掉 + * "每条一次 DB 往返 + 一轮一次 sleep"。 * * @return 本轮成功发出的条数 */ @@ -97,6 +100,7 @@ class Dispatcher( val maxRounds = props.pipeline.deliveryDrainRounds.coerceAtLeast(1) var sent = 0 var round = 0 + val paused = HashSet() // 本轮投递中被暂停的 FLID:队头失败或退避未到期 while (round < maxRounds) { round++ val batch = try { @@ -107,9 +111,17 @@ class Dispatcher( } if (batch.isEmpty()) return sent for (e in batch) { + val flid = e.partitionKey + if (flid in paused) continue // 同 FLID 暂停中:不越队 val next = e.nextAttemptAt - if (next != null && next > scheduler.now()) return sent // 队头退避未到期:停止推进 - if (!deliver(Targets.KAFKA_MSG, e)) return sent // 失败已记账:停止以保序 + if (next != null && next > scheduler.now()) { // 队头退避未到期:只停该 FLID + paused += flid + continue + } + if (!deliver(Targets.KAFKA_MSG, e)) { // 失败已记账:只停该 FLID + paused += flid + continue + } sent++ } if (batch.size < batchSize) return sent diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/delivery/DispatcherTickTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/delivery/DispatcherTickTest.kt index 6bc3cb4..0d30b59 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/delivery/DispatcherTickTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/delivery/DispatcherTickTest.kt @@ -214,6 +214,82 @@ class DispatcherTickTest { d.tick() assertEquals(1, port.sent.count { it.topic == "msg" && it.key == "F9" }) } + + @Test + fun `msg head failure pauses only the same FLID while other flights deliver in the same tick`() { + val repo = StubMsgEvents() + val port = FailingMsgKeyPort("F1") + val d = dispatcher(repo, port) + repo.insertAll( + listOf( + ev(1, Targets.KAFKA_MSG, "F1", """{"flid":"F1","v":1}"""), + ev(2, Targets.KAFKA_MSG, "F2", """{"flid":"F2","v":1}"""), + ev(3, Targets.KAFKA_MSG, "F1", """{"flid":"F1","v":2}"""), + ), + ) + + d.tick() + + // D2:F1 队头失败只暂停 F1(含 id=3 的后续),F2 照常发出 + assertEquals(1, port.sent.count { it.topic == "msg" && it.key == "F2" }) + assertEquals(0, port.sent.count { it.topic == "msg" && it.key == "F1" }) + val f1Rows = repo.rows.values.filter { it.partitionKey == "F1" } + assertEquals(2, f1Rows.size) + assertTrue(f1Rows.all { it.state == EventStatus.PENDING }) + } + + @Test + fun `msg head backoff not yet due pauses only the same FLID`() { + val repo = StubMsgEvents() + val port = StubDeliveryPort() + val d = dispatcher(repo, port) + repo.insertAll( + listOf( + ev(1, Targets.KAFKA_MSG, "F1", """{"flid":"F1","v":1}"""), + ev(2, Targets.KAFKA_MSG, "F2", """{"flid":"F2","v":1}"""), + ), + ) + repo.scheduleRetry(1, clock.instant().plusSeconds(60), 1) // F1 队头退避未到期 + + d.tick() + + assertEquals(1, port.sent.count { it.topic == "msg" && it.key == "F2" }) + val f1 = repo.rows.values.single { it.partitionKey == "F1" } + assertEquals(EventStatus.PENDING, f1.state) + assertEquals(1, f1.attempts) + } + + @Test + fun `same FLID msg events keep strict order across ticks`() { + val repo = StubMsgEvents() + val port = StubDeliveryPort() + val d = dispatcher(repo, port) + repo.insertAll( + listOf( + ev(1, Targets.KAFKA_MSG, "F1", """{"v":1}"""), + ev(2, Targets.KAFKA_MSG, "F1", """{"v":2}"""), + ev(3, Targets.KAFKA_MSG, "F1", """{"v":3}"""), + ), + ) + + d.tick() + + val f1Keys = port.sent.filter { it.topic == "msg" }.map { it.payload } + assertEquals(listOf("""{"v":1}""", """{"v":2}""", """{"v":3}"""), f1Keys) // 同 FLID 不越队 + } +} + +/** KAFKA:msg 对指定 FLID 的发送永远失败,其他 FLID 照常记录,用来验证按 FLID 暂停。 */ +private class FailingMsgKeyPort(private val failingKey: String) : DeliveryPort { + val sent = mutableListOf() + + override fun sendKafka(topic: String, key: String, payloadJson: String) { + if (key == failingKey) throw IllegalStateException("broker-down:$key") + sent.add(StubDeliveryPort.Sent(topic, key, payloadJson)) + } + + override fun sendKafkaSchd(topic: String, key: String, payloadJson: String) = Unit + override fun sendKafkaNull(topic: String, key: String) = Unit } /** KAFKA_SCHD 发送永远失败的投递端口,用来验证退避重试和最终转死信的闭环。 */