From 592079055ffbb98f60fa341a735f2099e11c64c4 Mon Sep 17 00:00:00 2001
From: windyboy <windyboy@gmail.com>
Date: Sat, 12 Sep 2026 20:39:15 +0800
Subject: [PATCH] fix(delivery): drop redundant schd event index
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

V9 追加删除已无消费者的 idx_evt_flid（读端不再按 (PARTITION_KEY, STATE_VERSION) 合并），并在最终 schema 断言其不存在；新增用例覆盖 schd DEAD 行被更高合法代次替换为单行 PENDING、错误字段清空、旧代次不得复活。

验证：./gradlew test 130 tests / 0 fail。
---
 .../db/migration/V9__schd_single_row.sql      |  3 +++
 .../delivery/DispatcherTickTest.kt            | 22 +++++++++++++++++++
 .../persistence/jdbc/FlywayMigrationTest.kt   |  6 +++++
 3 files changed, 31 insertions(+)

diff --git a/src/main/resources/db/migration/V9__schd_single_row.sql b/src/main/resources/db/migration/V9__schd_single_row.sql
index dd1620d..4f9a23a 100644
--- a/src/main/resources/db/migration/V9__schd_single_row.sql
+++ b/src/main/resources/db/migration/V9__schd_single_row.sql
@@ -22,3 +22,6 @@ DELETE FROM msg_event a
 CREATE UNIQUE INDEX uq_schd_event
     ON msg_event (target, partition_key)
  WHERE target = 'KAFKA:schd';
+
+-- 读端不再按 (PARTITION_KEY, STATE_VERSION) 合并（单行化 + 条件确认），idx_evt_flid 已无消费者。
+DROP INDEX IF EXISTS idx_evt_flid;
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 cdd9799..2cedc60 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.ErrorClass
 import com.gzzn.omms.msgexchange.domain.EventStatus
 import com.gzzn.omms.msgexchange.domain.EventType
 import com.gzzn.omms.msgexchange.domain.MsgEvent
@@ -181,6 +182,27 @@ class DispatcherTickTest {
         assertEquals(EventStatus.PENDING, row.state)   // 旧代次的确认影响 0 行，不把新内容带走
     }
 
+    @Test
+    fun `a higher generation replaces a dead schd row and clears the old error`() {
+        val repo = StubMsgEvents()
+        repo.insertAll(listOf(ev(1, Targets.KAFKA_SCHD, "F1", """{"v":"v1"}""", 1)))
+        val deadId = repo.rows.values.single().eventId!!
+        repo.markDead(deadId, ErrorClass.EXHAUSTED, "broker-down", attempts = 5)
+
+        repo.insertAll(listOf(ev(2, Targets.KAFKA_SCHD, "F1", """{"v":"v2"}""", 2)))
+
+        val row = repo.rows.values.single()   // 仍是一行；DEAD 只保留到下一合法代次
+        assertEquals(EventStatus.PENDING, row.state)
+        assertEquals(2, row.stateVersion)
+        assertEquals(0, row.attempts)
+        assertNull(row.errorClass)
+        assertNull(row.lastError)
+        assertTrue(row.eventId != deadId, "接受新代次必须以新的 EVENT_ID 替换行主键")
+
+        repo.insertAll(listOf(ev(3, Targets.KAFKA_SCHD, "F1", """{"v":"v1-late"}""", 1)))
+        assertEquals(2, repo.rows.values.single().stateVersion)   // 不被接受的旧代次不得复活该行
+    }
+
     @Test
     fun `KAFKA msg delivery unaffected while schd batch is retrying`() {
         val repo = StubMsgEvents()
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 e68a66c..abfff44 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
@@ -177,6 +177,12 @@ class FlywayMigrationTest {
                     "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())",
                 )
+                stmt.executeQuery(
+                    "SELECT count(*) FROM pg_indexes WHERE tablename = 'msg_event' AND indexname = 'idx_evt_flid'",
+                ).use { rs ->
+                    assertTrue(rs.next())
+                    assertEquals(0, rs.getInt(1), "V9 必须删除已无消费者的 idx_evt_flid")
+                }
 
                 // V6：入队时间是超期判据 R 的比较对象，必须非空（received_at 则允许为 NULL）
                 stmt.executeQuery(