From d21555091fb6d8d4ca8b3981ef6d0335a06f81ec Mon Sep 17 00:00:00 2001 From: windyboy Date: Sat, 12 Sep 2026 20:38:18 +0800 Subject: [PATCH] fix(delivery): upsert latest schedule event MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit KAFKA:schd 收敛为每 FLID 单行：V9 先按 (STATE_VERSION DESC, EVENT_ID DESC) 收敛存量再建部分唯一索引 uq_schd_event；写入走带谓词的单行 upsert（只进不退、同版本 tombstone 优先、接受新代次即换 EVENT_ID 并重置 PENDING/清旧错误）；读取不再 DISTINCT ON；成功确认改按 (EVENT_ID, STATE_VERSION, PENDING) 条件更新，删除 Dispatcher superseded 清理循环。msg 路径不变。 验证：全量 ./gradlew test 129 tests / 0 fail（新增 3 个用例：单行只进不退、同版本 tombstone 不可复活、发送期间新代次保持 PENDING）。V9 的存量收敛/索引与 upsert SQL 需真实 PG，本机无 Docker 时 FlywayMigrationTest 跳过。 --- .../omms/msgexchange/delivery/Dispatcher.kt | 37 +------ .../infra/persistence/Repositories.kt | 10 +- .../persistence/jdbc/JdbcPgRepositories.kt | 96 ++++++++++++++----- .../infra/stub/StubRepositories.kt | 43 +++++++-- .../db/migration/V9__schd_single_row.sql | 24 +++++ .../delivery/DispatcherTickTest.kt | 63 +++++++++++- .../persistence/jdbc/FlywayMigrationTest.kt | 32 ++++++- 7 files changed, 238 insertions(+), 67 deletions(-) create mode 100644 src/main/resources/db/migration/V9__schd_single_row.sql 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 74acf3b..e634f76 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt @@ -31,8 +31,8 @@ interface DeliveryPort { /** * 投递调度：把 outbox（待发事件表）里的事件发给下游。 * - * KAFKA_MSG 一条一条按登记顺序发，不插队。KAFKA_SCHD 走 flushSchd 批量发：同一个 FLID 攒了 - * 多条未发事件时只发版本号最新的那条，旧的自然作废；删除通知发 value 为空的 tombstone。 + * KAFKA_MSG 一条一条按登记顺序发，不插队。KAFKA_SCHD 走 flushSchd：outbox 每个 FLID 只保留 + * 一行（写入侧单行 upsert），发出后按读取时刻的代次做条件确认；删除通知发 value 为空的 tombstone。 * 两个主题之间不保证先后顺序。 * * 失败处理：队首的重试时间没到就不取；一批里有发送失败，整批重试次数加一并推后退避， @@ -138,7 +138,7 @@ class Dispatcher( } } - /** 批量发 KAFKA_SCHD：每个 FLID 只发版本号最新的那条未发事件，删除通知发空 value。 */ + /** 发 KAFKA_SCHD：每个 FLID 只有一行待发事件，删除通知发空 value，成功后按代次条件确认。 */ internal fun flushSchd() { val batch = try { msgEvents.mergePendingSchd(scheduler.now(), props.schd.flushLimit) @@ -157,34 +157,12 @@ class Dispatcher( EventType.TOMBSTONE -> port.sendKafkaNull("schd", e.partitionKey) EventType.UPSERT -> port.sendKafkaSchd("schd", e.partitionKey, e.payloadJson) } + // 条件确认：读取时刻的代次（EVENT_ID + STATE_VERSION）被新写入覆盖时不标记，留待下一轮重发。 + e.eventId?.let { msgEvents.markSentIfVersion(it, e.stateVersion) } } catch (ex: Exception) { failures.add(e) } } - val sentIds = batch.mapNotNull { it.eventId }.toSet() - failures.mapNotNull { it.eventId }.toSet() - if (sentIds.isNotEmpty()) msgEvents.markAllSent(sentIds.toList()) - val sentVersions = batch.associate { it.partitionKey to it.stateVersion } - // 被更新版本压掉的旧事件也要标成已发，否则它们会一直留在队里：同一个 FLID 只按最新版本输出一次。 - // 这里必须**有界**：若 markAllSent 因任何原因没有生效（例如行缺 eventId）， - // 旧的无界 while(true) 会原地空转。 - runCatching { - 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 - 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") } lastFlush = scheduler.now() } @@ -203,9 +181,4 @@ 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 - } } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt index 1de4855..777436f 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt @@ -167,11 +167,19 @@ interface MsgEventRepository { fun claimBatch(target: String, limit: Int): List - /** 挑出待发的整态事件，同一条航班只取版本最高的那一条（中间的版本不用发）。 */ + /** 领取待发的整态事件：写入侧已保证每个 `FLID` 只有一行，读端不再去重合并。 */ fun mergePendingSchd(now: Instant, limit: Int): List fun markSent(eventId: Long) + /** + * 条件确认：只在该行仍是本批读取到的代次（`EVENT_ID` + `STATE_VERSION`）且尚未发出时标记 `SENT`。 + * 发送期间被新代次覆盖的行影响 0 行，留待下一轮重发。 + * + * @return 受影响行数（0 或 1） + */ + fun markSentIfVersion(eventId: Long, stateVersion: Long): Int + fun markAllSent(eventIds: List) fun scheduleRetry(eventId: Long, nextAttemptAt: Instant, attempts: Int) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt index 237507e..d69b3fc 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt @@ -7,6 +7,7 @@ import com.gzzn.omms.msgexchange.domain.EventType import com.gzzn.omms.msgexchange.domain.MsgEvent import com.gzzn.omms.msgexchange.domain.ProcState import com.gzzn.omms.msgexchange.domain.ProcStatus +import com.gzzn.omms.msgexchange.domain.Targets import com.gzzn.omms.msgexchange.domain.flight.FlightMainRow import com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot import com.gzzn.omms.msgexchange.domain.flight.FlightState @@ -433,27 +434,65 @@ class JdbcMsgEventRepository( private val ds: DataSource, private val clock: Clock, ) : MsgEventRepository { + /** + * 写待发事件。`KAFKA:schd` 走按 `FLID` 的单行 upsert（同一 `FLID` 只保留最新代次）， + * 其他目标仍是一行一条的普通插入；返回实际生效的事件 ID（被版本守卫拒绝的 schd 写不返回值）。 + */ override fun insertAll(events: List): List = - events.map { e -> - ds.updateReturningLong( - """ - INSERT INTO msg_event (target, partition_key, event_type, state_version, payload_json, state, attempts, next_attempt_at, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING event_id - """.trimIndent(), - { ps -> - ps.setString(1, e.target) - ps.setString(2, e.partitionKey) - ps.setString(3, e.eventType.name) - ps.setLong(4, e.stateVersion) - ps.setString(5, e.payloadJson) - ps.setString(6, e.state.name) - ps.setInt(7, e.attempts) - ps.setTimestamp(8, e.nextAttemptAt?.toSqlTimestamp()) - ps.setTimestamp(9, e.createdAt.toSqlTimestamp()) - }, - ) + events.mapNotNull { e -> + if (e.target == Targets.KAFKA_SCHD) upsertSchd(e) else insertOne(e) } + private fun insertOne(e: MsgEvent): Long = + ds.updateReturningLong( + """ + INSERT INTO msg_event (target, partition_key, event_type, state_version, payload_json, state, attempts, next_attempt_at, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING event_id + """.trimIndent(), + { ps -> bindEvent(ps, e) }, + ) + + /** + * `KAFKA:schd` 单行 upsert：只进不退（仅更高 `STATE_VERSION` 覆盖），同版本只有 tombstone + * 能覆盖非 tombstone；接受的写代次复用本次 `EXCLUDED.EVENT_ID` 作行主键，并重置为 `PENDING` + * 且清空旧错误，因此发送期间被覆盖的旧批次条件确认只会影响 0 行（`design` 事件投递）。 + */ + private fun upsertSchd(e: MsgEvent): Long? = + ds.queryOne( + """ + INSERT INTO msg_event (target, partition_key, event_type, state_version, payload_json, state, attempts, next_attempt_at, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (target, partition_key) WHERE target = 'KAFKA:schd' + DO UPDATE SET + event_id = EXCLUDED.event_id, + event_type = EXCLUDED.event_type, + state_version = EXCLUDED.state_version, + payload_json = EXCLUDED.payload_json, + state = 'PENDING', + attempts = 0, + next_attempt_at = NULL, + error_class = NULL, + last_error = NULL, + created_at = EXCLUDED.created_at + WHERE EXCLUDED.state_version > msg_event.state_version + OR (EXCLUDED.state_version = msg_event.state_version AND EXCLUDED.event_type = 'TOMBSTONE') + RETURNING event_id + """.trimIndent(), + { ps -> bindEvent(ps, e) }, + ) { rs -> rs.getLong("event_id") } + + private fun bindEvent(ps: java.sql.PreparedStatement, e: MsgEvent) { + ps.setString(1, e.target) + ps.setString(2, e.partitionKey) + ps.setString(3, e.eventType.name) + ps.setLong(4, e.stateVersion) + ps.setString(5, e.payloadJson) + ps.setString(6, e.state.name) + ps.setInt(7, e.attempts) + ps.setTimestamp(8, e.nextAttemptAt?.toSqlTimestamp()) + ps.setTimestamp(9, e.createdAt.toSqlTimestamp()) + } + override fun claimBatch(target: String, limit: Int): List = ds.query( "SELECT * FROM msg_event WHERE target = ? AND state = 'PENDING' ORDER BY event_id ASC LIMIT ?", @@ -461,16 +500,16 @@ class JdbcMsgEventRepository( ::mapEvent, ) - /** 同一条航班只取版本最高的待发整态事件；用了 PostgreSQL 的 DISTINCT ON 语法。 */ + /** + * 领取待发的整态事件。写入侧已保证每个 `FLID` 只有一行（部分唯一索引 `uq_schd_event`）， + * 读端不再做去重合并，只按退避与 `EVENT_ID` 顺序取。 + */ override fun mergePendingSchd(now: Instant, limit: Int): List = ds.query( """ - SELECT * FROM ( - SELECT DISTINCT ON (partition_key) * - FROM msg_event WHERE target = 'KAFKA:schd' AND state = 'PENDING' - ORDER BY partition_key, state_version DESC, event_id DESC - ) latest - WHERE next_attempt_at IS NULL OR next_attempt_at <= ? + SELECT * FROM msg_event + WHERE target = 'KAFKA:schd' AND state = 'PENDING' + AND (next_attempt_at IS NULL OR next_attempt_at <= ?) ORDER BY event_id ASC LIMIT ? """.trimIndent(), { ps -> @@ -484,6 +523,13 @@ class JdbcMsgEventRepository( ds.update("UPDATE msg_event SET state = 'SENT' WHERE event_id = ?", { ps -> ps.setLong(1, eventId) }) } + /** 条件确认：仅当行仍是本批读到的代次且尚未发出时才标记，被新代次覆盖则影响 0 行。 */ + override fun markSentIfVersion(eventId: Long, stateVersion: Long): Int = + ds.update( + "UPDATE msg_event SET state = 'SENT' WHERE event_id = ? AND state_version = ? AND state = 'PENDING'", + { ps -> ps.setLong(1, eventId); ps.setLong(2, stateVersion) }, + ) + override fun markAllSent(eventIds: List) { if (eventIds.isEmpty()) return val placeholders = eventIds.joinToString(",") { "?" } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt index f190613..276dfa7 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt @@ -6,6 +6,7 @@ import com.gzzn.omms.msgexchange.domain.EventType import com.gzzn.omms.msgexchange.domain.MsgEvent import com.gzzn.omms.msgexchange.domain.ProcState import com.gzzn.omms.msgexchange.domain.ProcStatus +import com.gzzn.omms.msgexchange.domain.Targets import com.gzzn.omms.msgexchange.domain.flight.FlightMainRow import com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot import com.gzzn.omms.msgexchange.domain.flight.FlightState @@ -233,20 +234,41 @@ class StubMsgEvents : MsgEventRepository { override fun insertAll(events: List): List = events.map { e -> - val id = ids.incrementAndGet() - rows[id] = e.copy(eventId = id) - id + if (e.target == Targets.KAFKA_SCHD) upsertSchd(e) else insertOne(e) } + private fun insertOne(e: MsgEvent): Long { + val id = ids.incrementAndGet() + rows[id] = e.copy(eventId = id) + return id + } + + /** + * 与 JDBC upsert 同口径的单行投影：只进不退；同版本只有 tombstone 能覆盖非 tombstone； + * 接受的写代次换新 `EVENT_ID` 并重置 `PENDING`（清空 attempts/next/error）。 + */ + private fun upsertSchd(e: MsgEvent): Long { + val existing = rows.values.firstOrNull { it.target == Targets.KAFKA_SCHD && it.partitionKey == e.partitionKey } + val accepted = existing == null || + e.stateVersion > existing.stateVersion || + (e.stateVersion == existing.stateVersion && e.eventType == EventType.TOMBSTONE && existing.eventType != EventType.TOMBSTONE) + if (!accepted) return existing!!.eventId!! + existing?.let { rows.remove(it.eventId!!) } + val id = ids.incrementAndGet() + rows[id] = e.copy( + eventId = id, state = EventStatus.PENDING, attempts = 0, nextAttemptAt = null, + errorClass = null, lastError = null, + ) + return id + } + override fun claimBatch(target: String, limit: Int): List = rows.values.filter { it.target == target && it.state == EventStatus.PENDING }.sortedBy { it.eventId!! }.take(limit) - /** 同一条航班只取版本最高的待发整态事件，按事件先后输出。 */ + /** 写入侧已保证每个 FLID 单行，按退避与事件先后输出。 */ override fun mergePendingSchd(now: Instant, limit: Int): List = rows.values - .filter { it.target == "KAFKA:schd" && it.state == EventStatus.PENDING } - .groupBy { it.partitionKey } - .map { (_, group) -> group.maxBy { it.stateVersion } } + .filter { it.target == Targets.KAFKA_SCHD && it.state == EventStatus.PENDING } .filter { it.nextAttemptAt == null || it.nextAttemptAt <= now } .sortedBy { it.eventId!! } .take(limit) @@ -255,6 +277,13 @@ class StubMsgEvents : MsgEventRepository { rows[eventId] = (rows[eventId] ?: return).copy(state = EventStatus.SENT) } + override fun markSentIfVersion(eventId: Long, stateVersion: Long): Int { + val row = rows[eventId] ?: return 0 + if (row.state != EventStatus.PENDING || row.stateVersion != stateVersion) return 0 + rows[eventId] = row.copy(state = EventStatus.SENT) + return 1 + } + override fun markAllSent(eventIds: List) { eventIds.forEach { markSent(it) } } diff --git a/src/main/resources/db/migration/V9__schd_single_row.sql b/src/main/resources/db/migration/V9__schd_single_row.sql new file mode 100644 index 0000000..dd1620d --- /dev/null +++ b/src/main/resources/db/migration/V9__schd_single_row.sql @@ -0,0 +1,24 @@ +-- ===================================================================== +-- V9：KAFKA:schd outbox 单行化（按 FLID） +-- --------------------------------------------------------------------- +-- 只动自有 PostgreSQL；共享 MySQL 不建表、不改结构。 +-- +-- design「事件投递」：`KAFKA:schd` 只提供最新状态，outbox 按 `FLID` 单行 upsert。 +-- 存量是按版本追加的多行，必须先收敛再建部分唯一约束，否则升级直接失败： +-- 每个 (target='KAFKA:schd', partition_key) 留下 `STATE_VERSION DESC, EVENT_ID DESC` +-- 的第一行，与运行时的「只进不退」规则一致；`KAFKA:msg` 仍是多行 append-log，不受约束。 +-- ===================================================================== + +DELETE FROM msg_event a + USING msg_event b + WHERE a.target = 'KAFKA:schd' + AND b.target = 'KAFKA:schd' + AND a.partition_key = b.partition_key + AND ( + a.state_version < b.state_version + OR (a.state_version = b.state_version AND a.event_id < b.event_id) + ); + +CREATE UNIQUE INDEX uq_schd_event + ON msg_event (target, partition_key) + WHERE target = 'KAFKA:schd'; 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 38f784e..cdd9799 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/delivery/DispatcherTickTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/delivery/DispatcherTickTest.kt @@ -2,6 +2,7 @@ package com.gzzn.omms.msgexchange.delivery import com.gzzn.omms.msgexchange.MutableClock import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.domain.EventStatus import com.gzzn.omms.msgexchange.domain.EventType import com.gzzn.omms.msgexchange.domain.MsgEvent import com.gzzn.omms.msgexchange.domain.Targets @@ -63,7 +64,7 @@ class DispatcherTickTest { } @Test - fun `flushSchd aggregates latest version per flight and marks sent`() { + fun `flushSchd sends one row per flight and marks it sent`() { val repo = StubMsgEvents() val port = StubDeliveryPort() repo.insertAll( @@ -134,6 +135,52 @@ class DispatcherTickTest { assertTrue(final.attempts >= p.pipeline.maxAttempts || final.state.name == "DEAD") } + @Test + fun `schd writes keep one row per flight and never regress to an older version`() { + val repo = StubMsgEvents() + + repo.insertAll(listOf(ev(1, Targets.KAFKA_SCHD, "F1", """{"v":"v1"}""", 1))) + repo.insertAll(listOf(ev(2, Targets.KAFKA_SCHD, "F1", """{"v":"v2"}""", 2))) + repo.insertAll(listOf(ev(3, Targets.KAFKA_SCHD, "F1", """{"v":"late-old"}""", 1))) + + val row = repo.rows.values.single() + assertEquals("F1", row.partitionKey) + assertEquals(2, row.stateVersion) + assertTrue(row.payloadJson.contains("v2")) + } + + @Test + fun `same version tombstone wins and a later upsert cannot revive it`() { + val repo = StubMsgEvents() + repo.insertAll(listOf(ev(1, Targets.KAFKA_SCHD, "F1", """{"v":"v3"}""", 3))) + + repo.insertAll( + listOf( + MsgEvent( + target = Targets.KAFKA_SCHD, partitionKey = "F1", + eventType = EventType.TOMBSTONE, stateVersion = 3, payloadJson = """{"flid":"F1","deleted":true}""", + ), + ), + ) + assertEquals(EventType.TOMBSTONE, repo.rows.values.single().eventType) + + repo.insertAll(listOf(ev(9, Targets.KAFKA_SCHD, "F1", """{"v":"v3-again"}""", 3))) + assertEquals(EventType.TOMBSTONE, repo.rows.values.single().eventType) + } + + @Test + fun `new generation written during send keeps the row pending for the next round`() { + val repo = StubMsgEvents() + repo.insertAll(listOf(ev(1, Targets.KAFKA_SCHD, "F1", """{"v":"old"}""", 1))) + + dispatcher(repo, ReentrantSchdPort(repo)).flushSchd() + + val row = repo.rows.values.single() + assertEquals(99, row.stateVersion) + assertTrue(row.payloadJson.contains("new")) + assertEquals(EventStatus.PENDING, row.state) // 旧代次的确认影响 0 行，不把新内容带走 + } + @Test fun `KAFKA msg delivery unaffected while schd batch is retrying`() { val repo = StubMsgEvents() @@ -158,3 +205,17 @@ private class FailingSchdPort : DeliveryPort { throw IllegalStateException("broker-down") } } + +/** 在发送过程中给同一 FLID 写入新代次，用来验证「条件确认」不会把新内容标记成已发。 */ +private class ReentrantSchdPort(private val repo: StubMsgEvents) : DeliveryPort { + override fun sendKafka(topic: String, key: String, payloadJson: String) = Unit + override fun sendKafkaSchd(topic: String, key: String, payloadJson: String) { + repo.insertAll( + listOf( + MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = key, stateVersion = 99, payloadJson = """{"v":"new"}"""), + ), + ) + } + + override fun sendKafkaNull(topic: String, key: String) = Unit +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt index 9d5c67d..e68a66c 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt @@ -49,7 +49,7 @@ class FlywayMigrationTest { while (rs.next()) { records.add(Triple(rs.getString("version"), rs.getString("script"), rs.getBoolean("success"))) } - assertTrue(records.size >= 8, "flyway_schema_history must record all migrations") + assertTrue(records.size >= 9, "flyway_schema_history must record all migrations") assertEquals("1", records[0].first) assertEquals("V1__flight_state_baseline.sql", records[0].second) assertEquals("2", records[1].first) @@ -66,6 +66,8 @@ class FlywayMigrationTest { assertEquals("V7__drop_processing_started_at.sql", records[6].second) assertEquals("8", records[7].first) assertEquals("V8__req_track_open_unique.sql", records[7].second) + assertEquals("9", records[8].first) + assertEquals("V9__schd_single_row.sql", records[8].second) assertTrue(records.all { it.third }) } @@ -148,6 +150,34 @@ class FlywayMigrationTest { "VALUES ('RQFD-NONE', DATE '2026-09-12', 'RMS', 'PENDING', now())", ) + // V9：KAFKA:schd 每个 FLID 单行（部分唯一索引），KAFKA:msg 仍可多行 + stmt.executeQuery( + "SELECT indexdef FROM pg_indexes WHERE tablename = 'msg_event' AND indexname = 'uq_schd_event'", + ).use { rs -> + assertTrue(rs.next(), "V9 必须建立 schd 单行唯一索引 uq_schd_event") + val indexDef = rs.getString(1) + assertTrue(indexDef.contains("UNIQUE"), "uq_schd_event 必须是唯一索引：$indexDef") + assertTrue(indexDef.contains("WHERE"), "uq_schd_event 必须是仅约束 schd 的部分索引：$indexDef") + } + stmt.executeUpdate( + "INSERT INTO msg_event (target, partition_key, event_type, state_version, payload_json, state, attempts, created_at) " + + "VALUES ('KAFKA:schd', 'F1', 'UPSERT', 1, '{}', 'PENDING', 0, now())", + ) + assertThrows(java.sql.SQLException::class.java) { + stmt.executeUpdate( + "INSERT INTO msg_event (target, partition_key, event_type, state_version, payload_json, state, attempts, created_at) " + + "VALUES ('KAFKA:schd', 'F1', 'UPSERT', 2, '{}', 'PENDING', 0, now())", + ) + } + stmt.executeUpdate( + "INSERT INTO msg_event (target, partition_key, event_type, state_version, payload_json, state, attempts, created_at) " + + "VALUES ('KAFKA:msg', 'F1', 'UPSERT', 1, '{}', 'PENDING', 0, now())", + ) + stmt.executeUpdate( + "INSERT INTO msg_event (target, partition_key, event_type, state_version, payload_json, state, attempts, created_at) " + + "VALUES ('KAFKA:msg', 'F1', 'UPSERT', 2, '{}', 'PENDING', 0, now())", + ) + // V6：入队时间是超期判据 R 的比较对象，必须非空（received_at 则允许为 NULL） stmt.executeQuery( "SELECT is_nullable FROM information_schema.columns " +