feat(processing): 领域/投影/终态三步提交与 FlightProjectionPort（ACM2-79）

PIPELINE_LOCK 跨 Redis 写；MSG_EVENT HELD→PENDING；Redis 失败不终态不投递；V3 迁移。

Co-authored-by: Cursor <cursoragent@cursor.com>

src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt

@@ -31,6 +31,14 @@ interface PipelineTransactionManager {
 /** 单行锁：进事务后先 `SELECT ... FOR UPDATE`，让并发的状态写事务排队执行。 */
 interface PipelineLockRepository {
     fun lock()
+
+    /**
+     * 跨事务持锁：块内的多次事务复用同一条连接，锁到块结束才释放。
+     *
+     * 「领域提交 → 写 Redis → 终态提交」整段都要互斥，而行锁在提交那一刻就没了，
+     * 盖不住中间的 Redis 写（docs/implementation.md「事务边界」）。
+     */
+    fun <T> holdAcrossTransactions(block: () -> T): T
 }
 
 /**
@@ -172,6 +180,20 @@ data class Backlog(
 interface MsgEventRepository {
     fun insertAll(events: List<MsgEvent>): List<Long>
 
+    /**
+     * 清掉这条消息上一轮残留的暂存事件（`HELD`）。
+     *
+     * 领域事务重跑前调用：投影写失败或进程崩溃时，上一轮登记的事件还停在 `HELD`，
+     * 不清掉就会在重处理后变成永远没人放行的悬挂行。
+     */
+    fun discardHeld(msgId: Long): Int
+
+    /**
+     * 放行这条消息的暂存事件：`HELD` → `PENDING`，投递这才领得到。
+     * 由终态事务调用，所以 Redis 投影没写成功之前不会发出去（`INV-3`、`INV-10`）。
+     */
+    fun releaseHeld(msgId: Long): Int
+
     /**
      * 领取待发事件；[excludePartitionKeys] 为本轮已暂停的 FLID，排除后避免单航班占满整批
      * 导致其他航班饿死（ACM2-95）。

src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcOps.kt

@@ -14,13 +14,36 @@ private val transactionConnections = ThreadLocal.withInitial {
     java.util.IdentityHashMap<DataSource, java.sql.Connection>()
 }
 
+/**
+ * 被钉住的连接：块内的事务与单语句都走它。会话级锁（`pg_advisory_lock`）只在自己的会话里成立，
+ * 所以要跨事务持锁，就得让这几次事务落在同一条连接上。
+ */
+private val pinnedConnections = ThreadLocal.withInitial {
+    java.util.IdentityHashMap<DataSource, java.sql.Connection>()
+}
+
+internal fun <T> DataSource.withPinnedSession(block: (java.sql.Connection) -> T): T {
+    val pinned = pinnedConnections.get()
+    check(pinned[this] == null) { "session already pinned for this data source" }
+    val conn = this.connection
+    pinned[this] = conn
+    try {
+        return block(conn)
+    } finally {
+        pinned.remove(this)
+        if (pinned.isEmpty()) pinnedConnections.remove()
+        conn.close()
+    }
+}
+
 internal fun <T> DataSource.withTransaction(block: () -> T): T {
     val connections = transactionConnections.get()
     val existing = connections[this]
     if (existing != null) {
         return block()
     }
-    val conn = this.connection
+    val pinned = pinnedConnections.get()[this]
+    val conn = pinned ?: this.connection
     val oldAutoCommit = conn.autoCommit
     conn.autoCommit = false
     connections[this] = conn
@@ -42,19 +65,20 @@ internal fun <T> DataSource.withTransaction(block: () -> T): T {
             conn.autoCommit = oldAutoCommit
         } catch (_: Throwable) {
         }
-        conn.close()
+        // 钉住的连接由 withPinnedSession 关闭：提前还池会连同会话锁一起丢掉。
+        if (pinned == null) conn.close()
     }
 }
 
 internal fun DataSource.obtainConnection(): java.sql.Connection =
-    transactionConnections.get()[this] ?: this.connection
+    transactionConnections.get()[this] ?: pinnedConnections.get()[this] ?: this.connection
 
 internal fun java.sql.Connection.releaseIfNotInTransaction(dataSource: DataSource) {
     val connections = transactionConnections.get()
-    if (connections[dataSource] !== this) {
-        this.close()
-        if (connections.isEmpty()) transactionConnections.remove()
-    }
+    if (connections[dataSource] === this) return
+    if (pinnedConnections.get()[dataSource] === this) return
+    this.close()
+    if (connections.isEmpty()) transactionConnections.remove()
 }
 
 internal fun <T> DataSource.query(sql: String, bind: (java.sql.PreparedStatement) -> Unit, map: (ResultSet) -> T): List<T> {

src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt

@@ -63,10 +63,40 @@ class JdbcPipelineTransactionManager(
 class JdbcPipelineLockRepository(
     private val ds: DataSource,
 ) : PipelineLockRepository {
-    /** 进事务后的第一步：对锁行 `FOR UPDATE`，让并发的状态写事务排队。 */
+    /**
+     * 进事务后的第一步：先在咨询锁上排队，再对锁行 `FOR UPDATE`。
+     *
+     * 行锁在提交时就释放，盖不住处理步骤中段的 Redis 写，所以两个写者都要过同一把咨询锁：
+     * 主泵整段持会话级（[holdAcrossTransactions]），历史清理在自己的事务里持事务级。
+     * 同一会话重复取同一把咨询锁不自阻塞，因此持锁期间内层事务照常通过。
+     */
     override fun lock() {
+        ds.queryOne("SELECT pg_advisory_xact_lock(?)", { ps -> ps.setLong(1, PIPELINE_ADVISORY_KEY) }) { 1 }
         ds.queryOne("SELECT lock_id FROM pipeline_lock WHERE lock_id = 1 FOR UPDATE", {}) { 1 } ?: error("PIPELINE_LOCK row missing")
     }
+
+    /** 钉住一条连接并在其上持会话级咨询锁：块内的两次事务复用它，出块才解锁还池。 */
+    override fun <T> holdAcrossTransactions(block: () -> T): T = ds.withPinnedSession {
+        advisory("SELECT pg_advisory_lock(?)")
+        try {
+            block()
+        } finally {
+            // 解锁失败只能记账：连接随后归还连接池，会话锁若残留会挡住后续处理步骤。
+            runCatching { advisory("SELECT pg_advisory_unlock(?)") }
+                .onFailure { log.error("pipeline advisory unlock failed", it) }
+        }
+    }
+
+    private fun advisory(sql: String) {
+        ds.queryOne(sql, { ps -> ps.setLong(1, PIPELINE_ADVISORY_KEY) }) { 1 }
+    }
+
+    private companion object {
+        private val log = org.slf4j.LoggerFactory.getLogger(JdbcPipelineLockRepository::class.java)
+
+        /** `PIPELINE_LOCK` 的咨询锁键（'msgx' 的 ASCII 加序号 1）：同库其他应用撞键的概率足够低。 */
+        private const val PIPELINE_ADVISORY_KEY = 0x6D73_6778_01L
+    }
 }
 
 @Singleton
@@ -379,8 +409,8 @@ class JdbcMsgEventRepository(
     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
+            INSERT INTO msg_event (target, partition_key, event_type, state_version, payload_json, state, attempts, next_attempt_at, created_at, msg_id)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING event_id
             """.trimIndent(),
             { ps -> bindEvent(ps, e) },
         )
@@ -393,15 +423,16 @@ class JdbcMsgEventRepository(
     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 (?, ?, ?, ?, ?, ?, ?, ?, ?)
+            INSERT INTO msg_event (target, partition_key, event_type, state_version, payload_json, state, attempts, next_attempt_at, created_at, msg_id)
+            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',
+                msg_id = EXCLUDED.msg_id,
+                state = EXCLUDED.state,
                 attempts = 0,
                 next_attempt_at = NULL,
                 error_class = NULL,
@@ -425,8 +456,23 @@ class JdbcMsgEventRepository(
         ps.setInt(7, e.attempts)
         ps.setTimestamp(8, e.nextAttemptAt?.toSqlTimestamp())
         ps.setTimestamp(9, e.createdAt.toSqlTimestamp())
+        e.msgId?.let { ps.setLong(10, it) } ?: ps.setNull(10, java.sql.Types.BIGINT)
     }
 
+    /** 重处理前清残：上一轮登记但没放行的事件行删掉，避免留下永远发不出去的悬挂行。 */
+    override fun discardHeld(msgId: Long): Int =
+        ds.update(
+            "DELETE FROM msg_event WHERE msg_id = ? AND state = 'HELD'",
+            { ps -> ps.setLong(1, msgId) },
+        )
+
+    /** 放行：投影已写成功，事件转 PENDING 交给投递。 */
+    override fun releaseHeld(msgId: Long): Int =
+        ds.update(
+            "UPDATE msg_event SET state = 'PENDING' WHERE msg_id = ? AND state = 'HELD'",
+            { ps -> ps.setLong(1, msgId) },
+        )
+
     override fun claimBatch(target: String, limit: Int, excludePartitionKeys: Set<String>): List<MsgEvent> {
         if (excludePartitionKeys.isEmpty()) {
             return ds.query(
@@ -536,6 +582,7 @@ class JdbcMsgEventRepository(
 
     private fun mapEvent(rs: ResultSet) = MsgEvent(
         eventId = rs.getLong("event_id"),
+        msgId = rs.getLong("msg_id").takeIf { !rs.wasNull() },
         target = rs.getString("target"),
         partitionKey = rs.getString("partition_key"),
         eventType = EventType.valueOf(rs.getString("event_type")),

src/main/kotlin/com/gzzn/omms/msgexchange/infra/projection/FlightProjectionAdapters.kt

@@ -0,0 +1,49 @@
+package com.gzzn.omms.msgexchange.infra.projection
+
+import com.gzzn.omms.msgexchange.config.RedisProps
+import com.gzzn.omms.msgexchange.processing.FlightProjectionPort
+import com.gzzn.omms.msgexchange.processing.FlightProjectionWrite
+import io.micronaut.context.annotation.Requires
+import jakarta.inject.Singleton
+
+/**
+ * Redis 投影适配器骨架：键位与写法在这里定死，**但客户端还没接进来**（`G-REDIS-PROJECTION`）。
+ *
+ * 写入形态沿用旧系统：一个哈希键（`PARAM:msgx.redis.flight-key`）装全部航班，
+ * field 是 `FLID`，value 是整态 JSON，不设过期；删除航班就删掉这个 field（`INV-8`）。
+ *
+ * 开了 `msgx.redis.enabled` 却没有客户端时，写入直接抛异常而不是假装成功：
+ * 按 `INV-10`，这条消息就停在未完成、下轮重试，不会被当成已处理写回信箱。
+ */
+@Requires(property = "msgx.stubs", notEquals = "true")
+@Requires(property = "msgx.redis.enabled", value = "true")
+@Singleton
+class RedisFlightProjectionPort(
+    private val props: RedisProps,
+) : FlightProjectionPort {
+
+    override fun write(writes: List<FlightProjectionWrite>) {
+        throw UnsupportedOperationException(
+            "redis client not wired yet (G-REDIS-PROJECTION); pending writes=${writes.size} key=${props.flightKey}",
+        )
+    }
+
+    override fun ping(): Boolean = false
+}
+
+/**
+ * 没接 Redis 时的投影出口：什么都不做。
+ *
+ * 这不是"投影写成功"的承诺——`INV-10` 在真实客户端接通之前只能空转（`G-REDIS-PROJECTION`）。
+ * 摆这个 bean 是为了让三步提交的时序在没有 Redis 的环境里也照常跑，而不是让每条航班报文都失败。
+ */
+@Requires(property = "msgx.stubs", notEquals = "true")
+@Requires(property = "msgx.redis.enabled", notEquals = "true")
+@Singleton
+class NoopFlightProjectionPort : FlightProjectionPort {
+    private val log = org.slf4j.LoggerFactory.getLogger(NoopFlightProjectionPort::class.java)
+
+    override fun write(writes: List<FlightProjectionWrite>) {
+        log.debug("redis projection disabled, skipping {} write(s) [G-REDIS-PROJECTION]", writes.size)
+    }
+}

src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubAdapters.kt

@@ -1,6 +1,8 @@
 package com.gzzn.omms.msgexchange.infra.stub
 
 import com.gzzn.omms.msgexchange.delivery.DeliveryPort
+import com.gzzn.omms.msgexchange.processing.FlightProjectionPort
+import com.gzzn.omms.msgexchange.processing.FlightProjectionWrite
 import io.micronaut.context.annotation.Requires
 import jakarta.inject.Singleton
 
@@ -32,3 +34,37 @@ class StubDeliveryPort : DeliveryPort {
         sent += Sent(topic, key, null)
     }
 }
+
+/**
+ * 航班查询投影的内存假实现，只有配置 msgx.stubs=true 时才装配。
+ *
+ * [snapshots] 是投影的当前内容（FLID → 整态 JSON），删除会把条目移走，行为对齐 Redis 哈希。
+ * [failWith] 用来做故障注入：置上以后每次写都抛这个异常，用于验证"投影写不成功就不算处理完成"
+ * （`INV-10`）；置回 null 即恢复。
+ */
+@Requires(property = "msgx.stubs", value = "true")
+@Singleton
+class StubFlightProjectionPort : FlightProjectionPort {
+    val snapshots = linkedMapOf<String, String>()
+    val writes = mutableListOf<FlightProjectionWrite>()
+
+    @Volatile
+    var failWith: RuntimeException? = null
+
+    fun clear() {
+        snapshots.clear(); writes.clear(); failWith = null
+    }
+
+    override fun write(writes: List<FlightProjectionWrite>) {
+        failWith?.let { throw it }
+        this.writes += writes
+        writes.forEach { w ->
+            when (w) {
+                is FlightProjectionWrite.Upsert -> snapshots[w.flid] = w.payloadJson
+                is FlightProjectionWrite.Delete -> snapshots.remove(w.flid)
+            }
+        }
+    }
+
+    override fun ping(): Boolean = failWith == null
+}

src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt

@@ -51,6 +51,8 @@ class StubPipelineTx : PipelineTransactionManager {
 @Requires(property = "msgx.stubs", value = "true")
 class StubPipelineLock : PipelineLockRepository {
     override fun lock() = Unit
+
+    override fun <T> holdAcrossTransactions(block: () -> T): T = block()
 }
 
 /** 内存版 PROC_STATE：入队幂等，写终态时一并写下回填待办。 */
@@ -261,7 +263,7 @@ class StubMsgEvents : MsgEventRepository {
 
     /**
      * 与 JDBC upsert 同口径的单行投影：只进不退；同版本只有 tombstone 能覆盖非 tombstone；
-     * 接受的写代次换新 `EVENT_ID` 并重置 `PENDING`（清空 attempts/next/error）。
+     * 接受的写代次换新 `EVENT_ID` 并沿用本次写入的状态（清空 attempts/next/error）。
      */
     private fun upsertSchd(e: MsgEvent): Long {
         val existing = rows.values.firstOrNull { it.target == Targets.KAFKA_SCHD && it.partitionKey == e.partitionKey }
@@ -273,12 +275,26 @@ class StubMsgEvents : MsgEventRepository {
         }
         val id = ids.incrementAndGet()
         rows[id] = e.copy(
-            eventId = id, state = EventStatus.PENDING, attempts = 0, nextAttemptAt = null,
+            eventId = id, attempts = 0, nextAttemptAt = null,
             errorClass = null, lastError = null, sentAt = null,
         )
         return id
     }
 
+    /** 清残：上一轮登记但没放行的事件行删掉（对齐 JDBC 的按 MSG_ID 删 HELD）。 */
+    override fun discardHeld(msgId: Long): Int {
+        val ids = rows.values.filter { it.msgId == msgId && it.state == EventStatus.HELD }.map { it.eventId!! }
+        ids.forEach { rows.remove(it) }
+        return ids.size
+    }
+
+    /** 放行：HELD → PENDING，投递这才领得到。 */
+    override fun releaseHeld(msgId: Long): Int {
+        val held = rows.values.filter { it.msgId == msgId && it.state == EventStatus.HELD }
+        held.forEach { rows[it.eventId!!] = it.copy(state = EventStatus.PENDING) }
+        return held.size
+    }
+
     override fun claimBatch(target: String, limit: Int, excludePartitionKeys: Set<String>): List<MsgEvent> =
         rows.values
             .filter { it.target == target && it.state == EventStatus.PENDING }