From 2c3d6af30916ffafba54d60659221dfeb359db43 Mon Sep 17 00:00:00 2001 From: windyboy Date: Mon, 21 Sep 2026 15:24:34 +0800 Subject: [PATCH] =?UTF-8?q?feat(processing):=20=E9=A2=86=E5=9F=9F/?= =?UTF-8?q?=E6=8A=95=E5=BD=B1/=E7=BB=88=E6=80=81=E4=B8=89=E6=AD=A5?= =?UTF-8?q?=E6=8F=90=E4=BA=A4=E4=B8=8E=20FlightProjectionPort=EF=BC=88ACM2?= =?UTF-8?q?-79=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PIPELINE_LOCK 跨 Redis 写；MSG_EVENT HELD→PENDING；Redis 失败不终态不投递；V3 迁移。 Co-authored-by: Cursor --- README.md | 2 +- docs/implementation.md | 20 +- docs/reference.md | 5 +- docs/specification.md | 2 +- .../omms/msgexchange/config/RedisProps.kt | 13 + .../gzzn/omms/msgexchange/domain/MsgEvent.kt | 12 +- .../infra/persistence/Repositories.kt | 22 ++ .../infra/persistence/jdbc/JdbcOps.kt | 38 ++- .../persistence/jdbc/JdbcPgRepositories.kt | 59 +++- .../projection/FlightProjectionAdapters.kt | 49 ++++ .../msgexchange/infra/stub/StubAdapters.kt | 36 +++ .../infra/stub/StubRepositories.kt | 20 +- .../processing/DynamicProcessors.kt | 112 ++++---- .../msgexchange/processing/FlightCommit.kt | 67 +++++ .../processing/FlightProjection.kt | 47 ++++ .../gzzn/omms/msgexchange/processing/Pump.kt | 3 +- .../processing/ScheduleProcessor.kt | 66 +---- src/main/resources/application.yml | 3 + .../db/migration/V3__msg_event_hold.sql | 16 ++ .../db/migration/oracle11g/README.md | 6 +- .../persistence/jdbc/FlywayMigrationTest.kt | 27 +- .../jdbc/JdbcMsgEventUpsertPgTest.kt | 28 ++ .../jdbc/JdbcPipelineLockPgTest.kt | 110 ++++++++ .../processing/FdelAndAdftProcessorTest.kt | 24 +- .../processing/FlightCommitTest.kt | 263 ++++++++++++++++++ .../processing/IgnoreBranchTest.kt | 14 +- .../processing/ScheduleProcessorTest.kt | 6 +- .../processing/TestFlightCommit.kt | 25 ++ 28 files changed, 928 insertions(+), 167 deletions(-) create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/config/RedisProps.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/infra/projection/FlightProjectionAdapters.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommit.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/processing/FlightProjection.kt create mode 100644 src/main/resources/db/migration/V3__msg_event_hold.sql create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPipelineLockPgTest.kt create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommitTest.kt create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/processing/TestFlightCommit.kt diff --git a/README.md b/README.md index 25579c1..e53b81c 100644 --- a/README.md +++ b/README.md @@ -73,7 +73,7 @@ 仓库根目录提供兼容 Podman Compose 与 Docker Compose 的开发中间件栈 `compose.yaml`，包含： - **共享信箱 MySQL**（`mysql:8.4` LTS，端口 3306，库 `cdairport`）：容器启动时自动执行 `deploy/dev/mysql-init/01-mailbox.sql` 创建本地联调所需的 `CMINMSGS`、`CMINMSGS_HST`、`COUTMSGS` 模拟表。 - **自有 PostgreSQL**（`postgres:17-alpine`，端口 5432，库 `msgx`）：容器提供干净数据库，应用启动时由 Flyway（`src/main/resources/db/migration/V1__flight_state_baseline.sql`）自动建自有表。 -- **Valkey**（`valkey/valkey:8-alpine`，端口 6379）：本地兼容服务；应用不依赖它（`application.yml` 无对应配置键，航班状态权威在自有 PG）。Redis 航班快照与 `GET /all/flights` 是目标设计（`G-REDIS-PROJECTION`），当前实现不写入也不读取。 +- **Valkey**（`valkey/valkey:8-alpine`，端口 6379）：本地兼容服务；应用仍不连接它（航班状态权威在自有 PG）。处理侧的投影写入时序已就位，但 `msgx.redis.enabled` 默认关闭、写入是空操作，真实客户端与 `GET /all/flights` 仍未接通（`G-REDIS-PROJECTION`）。 - **Kafka**（`apache/kafka:3.8.0` KRaft 单节点，端口 9092）：listener `PLAINTEXT://localhost:9092`，`default.replication.factor=1`，已预配幂等生产者与 acks=all 所需的单节点参数。 ### 快速启动 diff --git a/docs/implementation.md b/docs/implementation.md index 056f238..d780d71 100644 --- a/docs/implementation.md +++ b/docs/implementation.md @@ -60,13 +60,15 @@ msgexchange-v2 怎么处理报文：记录模型、状态机、事务边界、 → FAILED（等待退避重试） → DEAD（MALFORMED / PROTOCOL / EXHAUSTED，均需人工处置） -投递：PENDING → SENT - → PENDING（退避后重试） - → DEAD（重试耗尽，记录保留作死信） +投递：HELD → PENDING → SENT + → PENDING（退避后重试） + → DEAD（重试耗尽，记录保留作死信） ``` `SUCCEEDED`/`SKIPPED`/`DEAD` 是终态，不挡后续；`FAILED` 仍占队头。用尽重试 → `DEAD(EXHAUSTED)`。错误分类与重放白名单见 [reference.md](reference.md)。 +`HELD` 是尚未放行的待发事件：领域事务登记，投影写成功后由终态事务转 `PENDING`（「事务边界」）。 + ## 4. 收报 ### 4.1 收报 @@ -116,9 +118,9 @@ processOne(head)： 载荷缺失 → DEAD(MALFORMED) 整包协议拒绝 → DEAD(PROTOCOL)，不落半包 5. 业务型成功，分三步（INV-3、INV-10）： - ① 事务提交：航班变更 + 待发事件 + ① 事务提交：航班变更 + 待发事件（登记为 HELD，投递领不到） ② 写 Redis 航班快照；失败 → 保持未完成，下轮重处理 - ③ 事务提交：SUCCEEDED + 回填意图 + ③ 事务提交：SUCCEEDED + 回填意图 + 待发事件放行（HELD → PENDING） 6. 结束：主泵不做回填；回填意图已随终态落库，由扫描补写信箱标记 ``` @@ -130,15 +132,17 @@ processOne(head)： |---|---|---|---|---| | 收报入队（`insertIfAbsent`） | 是 | 否 | 否 | 同库事务 | | 身份首次绑定 | 否 | 否 | 否 | 单语句 + 唯一约束 | -| 业务型领域变更（航班变更 + 待发事件） | 是 | 是 | 是 | 同库事务（`INV-3`） | +| 业务型领域变更（航班变更 + 待发事件） | 是 | 是 | 是 | 同库事务（`INV-3`）；事件落 `HELD` | | Redis 航班快照写 | 否 | 是（处理步骤锁跨越本步） | 否 | 外部副作用，不在 PG 事务内；写成功是终态事务的前置（`INV-10`） | -| 业务型终态（`SUCCEEDED` + 回填意图） | 是 | 是 | 否 | 同库事务（`INV-3`） | +| 业务型终态（`SUCCEEDED` + 回填意图 + 事件放行） | 是 | 是 | 否 | 同库事务（`INV-3`） | | 非业务型终态（`MALFORMED` / `PROTOCOL` / `SKIPPED` / `EXHAUSTED`） | 否 | 否 | 否 | 单语句（终态与回填意图同一条 UPDATE） | | 航班历史清理的物理删除 | 是 | 是（落实 `US-14` AC4） | 是 | 同库事务：复查判据 + 历史写入成功后删除 | | 回填（信箱标记 + `BACKFILL_AT`） | 否 | 否 | 否 | 跨库两次单写；幂等可重跑 | | 人工重放（批量改回 `PENDING`） | 否 | 否 | 否 | 单语句批量；`MessageLifecycleGate` 与回填互斥 | -航班变更与终态不在同一事务，中间夹 Redis 写（`INV-3`、`INV-10`）。`PIPELINE_LOCK` 跨 Redis 写，与历史清理互斥（`US-14` AC4）；无第二写者时不额外串行。 +航班变更与终态不在同一事务，中间夹 Redis 写（`INV-3`、`INV-10`）。三步跑在同一条数据库连接上：进处理步骤先在该连接取会话级咨询锁（`pg_advisory_lock`），两次事务复用这条连接，整步结束才解锁。行锁随提交释放，单靠它盖不住提交之后的 Redis 写，所以第二个写者（历史清理）在自己的事务里也要先取同一把咨询锁，再照旧锁 `PIPELINE_LOCK` 行，两边因此在整段内互斥（`US-14` AC4）；无第二写者时不额外串行。 + +投影失败时领域事务已提交而终态未写，`MSG_EVENT` 行停在 `HELD`：投递只领 `PENDING`，Kafka 不会先于处理完成发出（`INV-3`）。下一轮重处理按 `MSG_ID` 先清掉上一轮的 `HELD` 行再重新登记，崩溃重启后既不留悬挂行，也不把同一次变更投两遍。 ### 5.4 历史积压 diff --git a/docs/reference.md b/docs/reference.md index df5d48c..179b7f5 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -54,6 +54,8 @@ | `msgx.identity.include-day-boundary` | `false` | 去重时是否把日期算进消息身份；受 `C-3` 约束，不能随意改变 | 默认关闭 | | `msgx.health.backlog-cache-ttl-ms` | `30000` ms | `/health` 与 `/metrics` 共用的未处理消息统计最多缓存多久；`0` 表示每次重算 | 暂定 | | `msgx.history.history-store-enabled` | `false` | 是否启用历史航班写入；关闭时不删除实时航班 | 默认关闭 | +| `msgx.redis.enabled` | `false` | 是否把航班快照写进 Redis 查询投影；关闭时投影写入是空操作（`G-REDIS-PROJECTION`） | 默认关闭 | +| `msgx.redis.flight-key` | `flightInfo` | 航班快照所在的 Redis 哈希键，字段名是 `FLID` | 沿用旧系统 | | `msgx.history.planned-age-days` | `3` 天 | `US-14` AC2 条件 1：计划时间早于当前超过该天数视为已结束 | `US-14` AC2 | | `msgx.history.cancelled-hours` | `1` 小时 | `US-14` AC2 条件 2：取消时间早于当前超过该窗口视为已结束 | `US-14` AC2 | | `msgx.history.diverted-hours` | `1` 小时 | `US-14` AC2 条件 3：备降且计划时间早于当前超过该窗口；备降依据 `FDIV`（`DDES`/`DDIR`）落地前不参与判定（`Q3`、ACM2-100） | `US-14` AC2 | @@ -90,7 +92,7 @@ **缺值**：带环境变量占位的必需键（`MSGX_PG_URL`、`MSGX_MAILBOX_URL`、`MSGX_KAFKA_SERVERS` 等）缺失时由 DI 解析失败拒绝启动，无需自检代码。 -**显式开关（非配置错误）**：`msgx.stubs`、`datasources.default.enabled`、`mailbox.shared-mysql.enabled`、`msgx.history.history-store-enabled`、`msgx.pipeline.autostart`。默认关闭是"未接入"的显式声明，不判违规；只有 `msgx.pipeline.autostart=true` 才要求依赖齐备。Kafka 运行时不可达不是配置错误，归 `/health` 与投递指标告警（`OPS-2`）。 +**显式开关（非配置错误）**：`msgx.stubs`、`datasources.default.enabled`、`mailbox.shared-mysql.enabled`、`msgx.history.history-store-enabled`、`msgx.redis.enabled`、`msgx.pipeline.autostart`。默认关闭是"未接入"的显式声明，不判违规；只有 `msgx.pipeline.autostart=true` 才要求依赖齐备。Kafka 运行时不可达不是配置错误，归 `/health` 与投递指标告警（`OPS-2`）。 ## 指标与健康 @@ -121,6 +123,7 @@ | 读取信箱与 HTTP 入口 | `ingress/InboxPoller.kt`、`ingress/InboxController.kt` | | XML 解析 | `codec/JacksonXmlCodec.kt`、`codec/SisWireMapper.kt` | | 顺序处理与航班变更 | `processing/Pump.kt`、`processing/DynamicProcessors.kt`、`processing/ScheduleProcessor.kt` | +| 三步提交与 Redis 投影 | `processing/FlightCommit.kt`、`processing/FlightProjection.kt`、`infra/projection/` | | 写回信箱与发送 Kafka | `processing/BackfillService.kt`、`delivery/Dispatcher.kt`、`infra/kafka/KafkaDeliveryPort.kt` | | 定时任务 | `jobs/JobRunner.kt`、`jobs/HistorySweepJob.kt`、`jobs/EventCleanupJob.kt` | | 数据库读写与手动重试 | `infra/persistence/jdbc/`、`infra/retry/` | diff --git a/docs/specification.md b/docs/specification.md index 552dec6..907ba6f 100644 --- a/docs/specification.md +++ b/docs/specification.md @@ -153,7 +153,7 @@ | `G-FLOP-UNMAPPED` | [XSD](legacy/unisysaodbsis.xsd) FLOP 字段映射不全 | `US-05` | | `G-MAFL` | 主航班共享列表未做 | `US-06` AC2 | | `G-PROC-CLEANUP` | 处理记录到期清理未做 | `US-11` | -| `G-REDIS-PROJECTION` | Redis 快照写入与删除未做 | `US-05`、`US-06`、`US-07`、`US-12` | +| `G-REDIS-PROJECTION` | Redis 快照写入与删除未做：投影端口与三步提交时序已就位，真实 Redis 客户端未接通，日计划覆盖范围的缺席清扫也未做 | `US-05`、`US-06`、`US-07`、`US-12` | | `G-REF-DATA` | 静态参考数据处理与 admin-api 直读未做 | `US-13` | | `G-REQ-OPEN-UNIQUE` | 同类型未处理完时不允许再发，未做 | `US-09` | | `G-REQ-TRACK` | 出站请求跟踪未做 | `US-09` | diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/config/RedisProps.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/config/RedisProps.kt new file mode 100644 index 0000000..c808a79 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/config/RedisProps.kt @@ -0,0 +1,13 @@ +package com.gzzn.omms.msgexchange.config + +import io.micronaut.context.annotation.ConfigurationProperties + +/** 航班查询投影所在的 Redis，对应 `msgx.redis.*`；键位沿用旧系统（`C-11`、`Q6`）。 */ +@ConfigurationProperties("msgx.redis") +class RedisProps { + /** 有没有 Redis 可用。关着的时候投影写是空操作（`G-REDIS-PROJECTION`）。 */ + var enabled: Boolean = false + + /** 航班快照所在的哈希键，字段名是 `FLID`。 */ + var flightKey: String = "flightInfo" +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/MsgEvent.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/MsgEvent.kt index bc8fbb7..bce19a6 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/MsgEvent.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/MsgEvent.kt @@ -6,8 +6,14 @@ package com.gzzn.omms.msgexchange.domain */ enum class EventType { UPSERT, TOMBSTONE } -/** 投递状态：PENDING 待发 → SENT 已发出；发送失败按退避重试，重试次数用尽转 DEAD，留在表里当死信队列。 */ -enum class EventStatus { PENDING, SENT, DEAD } +/** + * 投递状态：HELD 暂存 → PENDING 待发 → SENT 已发出；发送失败按退避重试，重试次数用尽转 DEAD， + * 留在表里当死信队列。 + * + * HELD 是领域事务登记、还没放行的事件：Redis 投影写成功后由终态事务转成 PENDING， + * 投递只领 PENDING，因此处理完成前不会发 Kafka（`INV-3`、`INV-10`）。 + */ +enum class EventStatus { HELD, PENDING, SENT, DEAD } /** * MSG_EVENT：一张待发事件表（outbox），记录航班状态变更要对外发什么。 @@ -21,6 +27,8 @@ enum class EventStatus { PENDING, SENT, DEAD } */ data class MsgEvent( val eventId: Long? = null, + /** 登记这条事件的消息 ID；暂存事件按它清理与放行（`HELD`）。非管道写入方可留空。 */ + val msgId: Long? = null, val target: String, val partitionKey: String, // 分区键，恒为 FLID（航班实例 ID） val eventType: EventType = EventType.UPSERT, 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 45930a4..62246d4 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 @@ -31,6 +31,14 @@ interface PipelineTransactionManager { /** 单行锁：进事务后先 `SELECT ... FOR UPDATE`，让并发的状态写事务排队执行。 */ interface PipelineLockRepository { fun lock() + + /** + * 跨事务持锁：块内的多次事务复用同一条连接，锁到块结束才释放。 + * + * 「领域提交 → 写 Redis → 终态提交」整段都要互斥，而行锁在提交那一刻就没了， + * 盖不住中间的 Redis 写（docs/implementation.md「事务边界」）。 + */ + fun holdAcrossTransactions(block: () -> T): T } /** @@ -172,6 +180,20 @@ data class Backlog( interface MsgEventRepository { fun insertAll(events: List): List + /** + * 清掉这条消息上一轮残留的暂存事件（`HELD`）。 + * + * 领域事务重跑前调用：投影写失败或进程崩溃时，上一轮登记的事件还停在 `HELD`， + * 不清掉就会在重处理后变成永远没人放行的悬挂行。 + */ + fun discardHeld(msgId: Long): Int + + /** + * 放行这条消息的暂存事件：`HELD` → `PENDING`，投递这才领得到。 + * 由终态事务调用，所以 Redis 投影没写成功之前不会发出去（`INV-3`、`INV-10`）。 + */ + fun releaseHeld(msgId: Long): Int + /** * 领取待发事件；[excludePartitionKeys] 为本轮已暂停的 FLID，排除后避免单航班占满整批 * 导致其他航班饿死（ACM2-95）。 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcOps.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcOps.kt index b9dd2fe..85dcd39 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcOps.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcOps.kt @@ -14,13 +14,36 @@ private val transactionConnections = ThreadLocal.withInitial { java.util.IdentityHashMap() } +/** + * 被钉住的连接：块内的事务与单语句都走它。会话级锁（`pg_advisory_lock`）只在自己的会话里成立， + * 所以要跨事务持锁，就得让这几次事务落在同一条连接上。 + */ +private val pinnedConnections = ThreadLocal.withInitial { + java.util.IdentityHashMap() +} + +internal fun 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 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 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 DataSource.query(sql: String, bind: (java.sql.PreparedStatement) -> Unit, map: (ResultSet) -> T): List { 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 4e968e2..aac979f 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 @@ -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 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): List { 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")), diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/projection/FlightProjectionAdapters.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/projection/FlightProjectionAdapters.kt new file mode 100644 index 0000000..c71c5a9 --- /dev/null +++ b/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) { + 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) { + log.debug("redis projection disabled, skipping {} write(s) [G-REDIS-PROJECTION]", writes.size) + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubAdapters.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubAdapters.kt index bde7ca3..05f3508 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubAdapters.kt +++ b/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() + val writes = mutableListOf() + + @Volatile + var failWith: RuntimeException? = null + + fun clear() { + snapshots.clear(); writes.clear(); failWith = null + } + + override fun write(writes: List) { + 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 +} 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 2a897c7..0fc1795 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 @@ -51,6 +51,8 @@ class StubPipelineTx : PipelineTransactionManager { @Requires(property = "msgx.stubs", value = "true") class StubPipelineLock : PipelineLockRepository { override fun lock() = Unit + + override fun 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): List = rows.values .filter { it.target == target && it.state == EventStatus.PENDING } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/DynamicProcessors.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/DynamicProcessors.kt index abde0d3..a96931c 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/DynamicProcessors.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/DynamicProcessors.kt @@ -4,11 +4,10 @@ import com.fasterxml.jackson.databind.ObjectMapper import com.gzzn.omms.msgexchange.codec.FlopPayload import com.gzzn.omms.msgexchange.config.OperationDayProps import com.gzzn.omms.msgexchange.domain.DecodedMessage -import com.gzzn.omms.msgexchange.domain.EventType +import com.gzzn.omms.msgexchange.domain.EventStatus import com.gzzn.omms.msgexchange.domain.MsgEvent import com.gzzn.omms.msgexchange.domain.OperationDayCalculator 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.FlightSnapshot import com.gzzn.omms.msgexchange.domain.flight.FlightState @@ -18,9 +17,6 @@ import com.gzzn.omms.msgexchange.domain.flight.ScheduleRecord import com.gzzn.omms.msgexchange.infra.persistence.FlightStateRepository import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository import com.gzzn.omms.msgexchange.infra.persistence.PersistOutcome -import com.gzzn.omms.msgexchange.infra.persistence.PipelineLockRepository -import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager -import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository import jakarta.inject.Singleton import java.time.Clock import java.time.Instant @@ -35,29 +31,22 @@ import java.time.ZoneId */ @Singleton class FlopProcessor( - private val txManager: PipelineTransactionManager, - private val lock: PipelineLockRepository, + private val commit: FlightCommit, private val flightState: FlightStateRepository, private val msgEvents: MsgEventRepository, - private val procState: ProcStateRepository, private val mapper: ObjectMapper, private val clock: Clock, ) { - fun apply(head: ProcState, msg: DecodedMessage, payload: FlopPayload): ApplyResult = txManager.inTransaction { - lock.lock() + fun apply(head: ProcState, msg: DecodedMessage, payload: FlopPayload): ApplyResult = commit.commit(head) { + // 迟到/未知航班：幂等成功，不创建（创建入口只有 SCHD/ADFT），也没有投影要刷 val current = flightState.loadFullSnapshot(payload.flid) - if (current == null) { - // 迟到/未知航班：幂等成功，不创建（创建入口只有 SCHD/ADFT）；终态同事务落库 - procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED, now = clock.instant()) - return@inTransaction ApplyResult.Succeeded - } + ?: return@commit DomainOutcome(ApplyResult.Succeeded) val change = MergeChange(flid = payload.flid, scalars = payload.scalars, collections = payload.collections) val next = FlightStateEngine.mergedState(current, change) flightState.persistFullState(next, msgId = head.msgId, now = clock.instant()) - msgEvents.insertAll(eventsFor(next, mapper, clock.instant())) - procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED, now = clock.instant()) - ApplyResult.Succeeded + msgEvents.insertAll(eventsFor(next, head.msgId, mapper, clock.instant())) + DomainOutcome(ApplyResult.Succeeded, listOf(projectionOf(next, mapper))) } } @@ -65,43 +54,47 @@ class FlopProcessor( * 处理 FDEL 删除报文：把在用的航班标记为 DELETED，版本号加一，明细数据保留， * 并且只发布一次删除事件。 * - * 已经删除过、或航班本来就不存在时，算处理成功但不再动版本、不重复发事件。 + * 已经删除过、或航班本来就不存在时，算处理成功但不再动版本、不重复发事件； + * 但只要航班还在库里且是删除态，投影删除就照发一次——幂等命令换来"上一轮没删干净"能自愈（`INV-8`）。 */ @Singleton class FdelProcessor( - private val txManager: PipelineTransactionManager, - private val lock: PipelineLockRepository, + private val commit: FlightCommit, private val flightState: FlightStateRepository, private val msgEvents: MsgEventRepository, - private val procState: ProcStateRepository, private val mapper: ObjectMapper, private val clock: Clock, ) { - fun apply(head: ProcState, msg: DecodedMessage, payload: FlopPayload): ApplyResult = txManager.inTransaction { - lock.lock() + fun apply(head: ProcState, msg: DecodedMessage, payload: FlopPayload): ApplyResult = commit.commit(head) { val deleted = flightState.markDeleted(payload.flid, msgId = head.msgId, now = clock.instant()) - if (deleted) { - val current = flightState.loadFullSnapshot(payload.flid) - // 只有"在用 → 删除"这一步才发删除通知，而且和状态变更写在同一个事务里 - val createdAt = clock.instant() + // 航班根本不存在（迟到或多余的删除报文）：算成功，也没有投影要删 + val main = flightState.findMainRow(payload.flid) + ?: return@commit DomainOutcome(ApplyResult.Succeeded) + + // 这一次删掉的，或者上一轮就是本消息删掉的（投影写失败后重处理），都要登记删除通知： + // 重处理开始时已把上一轮没放行的事件清掉，这里不补登就会把这条通知丢了。 + if (deleted || (main.state == FlightState.DELETED && main.lastMsgId == head.msgId)) { msgEvents.insertAll( listOf( // C-9：删除通知只走 KAFKA:msg，schd 不再发 tombstone； // value 形态（"deleted":true 的 JSON）沿用现状，待 Q5 定稿 MsgEvent( + msgId = head.msgId, target = Targets.KAFKA_MSG, partitionKey = payload.flid, - stateVersion = current?.stateVersion ?: 0L, + stateVersion = main.stateVersion, payloadJson = mapper.writeValueAsString( - mapOf("flid" to payload.flid, "stateVersion" to (current?.stateVersion ?: 0L), "deleted" to true), + mapOf("flid" to payload.flid, "stateVersion" to main.stateVersion, "deleted" to true), ), - createdAt = createdAt, + state = EventStatus.HELD, + createdAt = clock.instant(), ), ), ) } - procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED, now = clock.instant()) // 没删到东西说明是迟到或重复报文，照样算成功 - ApplyResult.Succeeded + // INV-8：只要航班是删除态就得从投影里抹掉。重复报文也重删一次——代价是一条幂等命令， + // 换来"上一轮删投影没成功"能自愈。 + DomainOutcome(ApplyResult.Succeeded, listOf(FlightProjectionWrite.Delete(payload.flid, main.stateVersion))) } } @@ -114,11 +107,9 @@ class FdelProcessor( */ @Singleton class AdftProcessor( - private val txManager: PipelineTransactionManager, - private val lock: PipelineLockRepository, + private val commit: FlightCommit, private val flightState: FlightStateRepository, private val msgEvents: MsgEventRepository, - private val procState: ProcStateRepository, operationDayProps: OperationDayProps, private val mapper: ObjectMapper, private val clock: Clock, @@ -130,24 +121,27 @@ class AdftProcessor( fun apply(head: ProcState, msg: DecodedMessage, record: ScheduleRecord): ApplyResult = try { - txManager.inTransaction { - lock.lock() + commit.commit(head) { val main = flightState.findMainRow(record.flid) if (main != null && main.state == FlightState.DELETED) { // 已删除的航班重新激活：状态改回 ACTIVE、版本号加一，并登记状态事件 + var revived: FlightSnapshot? = null if (flightState.revive(record.flid, msgId = head.msgId, now = clock.instant())) { val current = flightState.loadFullSnapshot(record.flid) if (current != null) { val next = FlightStateEngine.mergedState(current, setOnly(record)) flightState.persistFullState(next, msgId = head.msgId, now = clock.instant()) - msgEvents.insertAll(eventsFor(next, mapper, clock.instant())) + msgEvents.insertAll(eventsFor(next, head.msgId, mapper, clock.instant())) + revived = next } } else { // 并发下可能已被别的消息处理掉：不改版本、不重复发事件，但留下痕迹便于对账。 log.warn("adft revive no-op (already active or vanished) msgId={} flid={}", head.msgId, record.flid) } - procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED, now = clock.instant()) - return@inTransaction ApplyResult.Succeeded + return@commit DomainOutcome( + ApplyResult.Succeeded, + listOfNotNull(revived?.let { projectionOf(it, mapper) }), + ) } val current = flightState.loadFullSnapshot(record.flid) @@ -173,9 +167,8 @@ class AdftProcessor( // 若按 INFRA 抛出去，会被当成暂时性故障白白重试到耗尽，并给出误导的错误类别。 throw ProtocolViolation("operation-day guard violated flid=${record.flid}") } - msgEvents.insertAll(eventsFor(next, mapper, clock.instant())) - procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED, now = clock.instant()) - ApplyResult.Succeeded + msgEvents.insertAll(eventsFor(next, head.msgId, mapper, clock.instant())) + DomainOutcome(ApplyResult.Succeeded, listOf(projectionOf(next, mapper))) } } catch (e: ProtocolViolation) { log.error("adft DEAD(PROTOCOL) msgId={} reason={}", head.msgId, e.message) @@ -201,28 +194,37 @@ class AdftProcessor( // 共享小工具（处理器层私有约定） // ===================================================================== -/** 航班状态变化后要发的两类事件：KAFKA_SCHD 发完整状态，KAFKA_MSG 发一条变更通知。 */ -internal fun eventsFor(next: FlightSnapshot, mapper: ObjectMapper, createdAt: Instant): List { - val payload = linkedMapOf( - "flid" to next.flid, - "stateVersion" to next.stateVersion, - "scalars" to next.scalars, - "collections" to next.collections, - ) - return listOf( +/** 航班整态载荷：`KAFKA:schd` 事件与 Redis 投影共用这一份形状。 */ +internal fun flightPayload(next: FlightSnapshot): Map = linkedMapOf( + "flid" to next.flid, + "stateVersion" to next.stateVersion, + "scalars" to next.scalars, + "collections" to next.collections, +) + +/** + * 航班状态变化后要发的两类事件：KAFKA_SCHD 发完整状态，KAFKA_MSG 发一条变更通知。 + * + * 两条都以 [EventStatus.HELD] 登记，投影写成功后由终态事务放行（`INV-3`、`INV-10`）。 + */ +internal fun eventsFor(next: FlightSnapshot, msgId: Long, mapper: ObjectMapper, createdAt: Instant): List = + listOf( MsgEvent( + msgId = msgId, target = Targets.KAFKA_SCHD, partitionKey = next.flid, stateVersion = next.stateVersion, - payloadJson = mapper.writeValueAsString(payload), + payloadJson = mapper.writeValueAsString(flightPayload(next)), + state = EventStatus.HELD, createdAt = createdAt, ), MsgEvent( + msgId = msgId, target = Targets.KAFKA_MSG, partitionKey = next.flid, stateVersion = next.stateVersion, payloadJson = mapper.writeValueAsString(mapOf("flid" to next.flid, "stateVersion" to next.stateVersion)), + state = EventStatus.HELD, createdAt = createdAt, ), ) -} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommit.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommit.kt new file mode 100644 index 0000000..da82b07 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommit.kt @@ -0,0 +1,67 @@ +package com.gzzn.omms.msgexchange.processing + +import com.gzzn.omms.msgexchange.domain.ProcState +import com.gzzn.omms.msgexchange.domain.ProcStatus +import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository +import com.gzzn.omms.msgexchange.infra.persistence.PipelineLockRepository +import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager +import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository +import jakarta.inject.Singleton +import java.time.Clock + +/** + * 领域事务的产物：交给调用方的结果，加上这次变更要刷进查询投影的航班。 + * 没有航班变更（迟到报文、空日计划）时投影为空，这一步直接跳过。 + */ +data class DomainOutcome(val value: T, val projections: List = emptyList()) + +/** + * 业务型成功的三步提交（`INV-3`、`INV-10`，见 docs/implementation.md「事务边界」）： + * + * 1. 领域事务：航班变更与待发事件一起提交，事件落 `HELD`（投递领不到）； + * 2. 写 Redis 投影：失败就抛出去，这条消息保持未完成，下轮重处理； + * 3. 终态事务：放行待发事件，写 `SUCCEEDED` 与回填意图。 + * + * 三步整段持 `PIPELINE_LOCK`：行锁提交即释放，盖不住中间的投影写，所以由 + * [PipelineLockRepository.holdAcrossTransactions] 在一条连接上从头持到尾。 + * + * 投影失败时第一步已经提交，**不回滚**（`US-03` AC3）：重处理按当前完整态重写投影， + * 上一轮的 `HELD` 事件先清后补，不会重复投递。 + */ +@Singleton +class FlightCommit( + private val txManager: PipelineTransactionManager, + private val lock: PipelineLockRepository, + private val procState: ProcStateRepository, + private val msgEvents: MsgEventRepository, + private val projection: FlightProjectionPort, + private val clock: Clock, +) { + fun commit(head: ProcState, domain: () -> DomainOutcome): T = lock.holdAcrossTransactions { + val outcome = txManager.inTransaction { + lock.lock() + msgEvents.discardHeld(head.msgId) // 清掉上一轮没放行的事件，再登记本轮的 + domain() + } + + writeProjection(head, outcome.projections) + + txManager.inTransaction { + lock.lock() + msgEvents.releaseHeld(head.msgId) + procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED, now = clock.instant()) + } + outcome.value + } + + private fun writeProjection(head: ProcState, writes: List) { + if (writes.isEmpty()) return + try { + projection.write(writes) + } catch (e: InterruptedException) { + throw e // 停机信号不归到这条消息头上 + } catch (e: Exception) { + throw FlightProjectionFailure(head.msgId, e) + } + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/FlightProjection.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/FlightProjection.kt new file mode 100644 index 0000000..f079653 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/FlightProjection.kt @@ -0,0 +1,47 @@ +package com.gzzn.omms.msgexchange.processing + +import com.fasterxml.jackson.databind.ObjectMapper +import com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot + +/** + * 一次投影写：把某个航班的整态写进查询投影，或者把它从投影里删掉。 + * 带上版本号，适配层据此做"只进不退"的覆盖判断。 + */ +sealed interface FlightProjectionWrite { + val flid: String + val stateVersion: Long + + data class Upsert( + override val flid: String, + override val stateVersion: Long, + val payloadJson: String, + ) : FlightProjectionWrite + + data class Delete(override val flid: String, override val stateVersion: Long) : FlightProjectionWrite +} + +/** + * 航班查询投影（Redis）的出口。投影是 PG 当前态的副本，只读不权威（`INV-5`、`INV-11`）。 + * + * 写成功这条消息才算处理完成（`INV-10`）：**写不成功必须抛异常**，不能吞掉—— + * 调用方据此保持未完成状态、下轮整条重处理（`US-05` AC4）。 + */ +interface FlightProjectionPort { + /** 一批一起写：同一条消息产生的写要么整批成功，要么抛异常整批重来。 */ + fun write(writes: List) + + /** 给健康检查用的连通性探测。 */ + fun ping(): Boolean = true +} + +/** 投影写失败：包一层好让日志和错误分类看得出是投影这一步挂的，处理侧按 `INFRA` 重试。 */ +class FlightProjectionFailure(msgId: Long, cause: Throwable) : + RuntimeException("flight projection write failed msgId=$msgId: ${cause.message ?: cause.javaClass.simpleName}", cause) + +/** 投影载荷与 `KAFKA:schd` 整态同一份 JSON：投影和通知同源，对账时不用比两种形状（`C-11`）。 */ +internal fun projectionOf(snapshot: FlightSnapshot, mapper: ObjectMapper): FlightProjectionWrite.Upsert = + FlightProjectionWrite.Upsert( + flid = snapshot.flid, + stateVersion = snapshot.stateVersion, + payloadJson = mapper.writeValueAsString(flightPayload(snapshot)), + ) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt index b797260..6de640b 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt @@ -106,7 +106,8 @@ class Pump( /** * 处理一条消息：读原文 → 解码 → 绑定业务身份 → 分派给对应处理器。 * - * 业务数据、终态与回填意图都由各处理器在自己的事务里写入（终态与回填意图是同一条 UPDATE）。 + * 业务数据、Redis 投影、终态与回填意图由各处理器按三步提交写入（[FlightCommit]）： + * 领域事务提交后才写投影，投影成功后才记终态与回填意图（同一条 UPDATE）。 * **这里不做信箱回填**：主泵是 FIFO 关键路径，跨库写会把它绑在共享 MySQL 的可用性上。 * 回填由 `JobRunner` 定时的 `BackfillService.sweep` 驱动（调度周期不等于完成时限）。 * diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessor.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessor.kt index 08acbc4..90530d5 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessor.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessor.kt @@ -6,33 +6,26 @@ import com.gzzn.omms.msgexchange.config.OperationDayProps import com.gzzn.omms.msgexchange.domain.DecodedMessage 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.SnapshotFlag import com.gzzn.omms.msgexchange.domain.SnapshotLogEntry import com.gzzn.omms.msgexchange.domain.SnapshotResult -import com.gzzn.omms.msgexchange.domain.Targets import com.gzzn.omms.msgexchange.domain.OperationDayCalculator -import com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot import com.gzzn.omms.msgexchange.domain.flight.FlightState import com.gzzn.omms.msgexchange.domain.flight.FlightStateEngine -import com.gzzn.omms.msgexchange.domain.flight.ScheduleRecord import com.gzzn.omms.msgexchange.domain.flight.SnapshotValidation import com.gzzn.omms.msgexchange.infra.persistence.FlightStateRepository import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository import com.gzzn.omms.msgexchange.infra.persistence.PersistOutcome -import com.gzzn.omms.msgexchange.infra.persistence.PipelineLockRepository -import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository import com.gzzn.omms.msgexchange.infra.persistence.SnapshotLogRepository import jakarta.inject.Singleton import java.time.Clock -import java.time.Instant import java.time.LocalDate import java.time.ZoneId /** * 处理器告诉调用方这一条消息处理成了什么。 - * 无论哪种结果，终态和回填待办都由处理器在自己的事务里写好。 + * 业务型成功的终态和回填待办已由三步提交写好（[FlightCommit]），调用方不必再补。 */ sealed interface ApplyResult { /** 业务处理成功（包含"重复写入但结果一致"这种幂等成功）。 */ @@ -52,15 +45,14 @@ class ProtocolViolation(message: String) : RuntimeException(message) * 处理 SCHD 日计划报文（DNLD 和 RESP 走同一条路）：把报文里的航班记录合并进航班当前态。 * * 顺序是：整包校验 → 加锁 → 核对每条航班的运营日 → 逐条合并写入 → 登记待发事件 → - * 写终态和回填待办。除了校验，后面所有步骤都在同一个事务里，任何一步失败整包回滚， - * 不会留下写了一半的数据。 + * 刷投影 → 写终态和回填待办。领域这一段在同一个事务里，任何一步失败整包回滚， + * 不会留下写了一半的数据；投影与终态按三步提交各走一步（[FlightCommit]）。 * * 报文里没提到的航班不会被删除——日计划只负责写它带来的那部分。 */ @Singleton class ScheduleProcessor( - private val txManager: PipelineTransactionManager, - private val lock: PipelineLockRepository, + private val commit: FlightCommit, private val procState: ProcStateRepository, private val flightState: FlightStateRepository, private val msgEvents: MsgEventRepository, @@ -98,21 +90,17 @@ class ScheduleProcessor( val ok = validation as SnapshotValidation.Ok if (ok.perRecordDay.isEmpty()) { - // 报文合法但没有记录：不写航班，但终态与回填意图仍在锁事务内一起提交 - txManager.inTransaction { - lock.lock() - procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED, now = clock.instant()) - } + // 报文合法但没有记录：不写航班、没有投影要刷，终态与回填意图仍走三步提交的最后一步 + commit.commit(head) { DomainOutcome(Unit) } logSnapshot(head, body, SnapshotResult.COMMITTED, upserted = 0, setOf(SnapshotFlag.EMPTY), started) return ApplyResult.Succeeded } val flags = linkedSetOf() return try { - // 以下都在同一个事务里：加锁 → 校验运营日 → 逐条合并写入 → 登记事件与回填待办 - val upserted = txManager.inTransaction { - lock.lock() - + // 领域这一段在同一个事务里：加锁 → 校验运营日 → 逐条合并写入 → 登记事件； + // 投影与终态由 FlightCommit 接着走后两步 + val upserted = commit.commit(head) { // 一次批量查出这些航班现有的运营日，逐个比对（避免逐条查询） val mains = flightState.findMainRows(ok.perRecordDay.keys) ok.perRecordDay.forEach { (flid, day) -> @@ -126,6 +114,7 @@ class ScheduleProcessor( var written = 0 val events = mutableListOf() + val projections = mutableListOf() // 一次建索引，避免在航班级循环里反复线性扫描（大日计划下是 O(N²)）。 val recordsByFlid = body.records.associateBy { it.flid } ok.perRecordDay.forEach { (flid, day) -> @@ -145,12 +134,11 @@ class ScheduleProcessor( throw ProtocolViolation("operation-day guard violated flid=$flid") else -> written++ } - events += snapshotEvents(next) + events += eventsFor(next, head.msgId, mapper, clock.instant()) + projections += projectionOf(next, mapper) } if (events.isNotEmpty()) msgEvents.insertAll(events) - // 终态与回填待办跟业务数据同事务提交：要么全成，要么全回滚 - procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED, now = clock.instant()) - written + DomainOutcome(written, projections) } logSnapshot(head, body, SnapshotResult.COMMITTED, upserted, flags, started) ApplyResult.Succeeded @@ -160,34 +148,6 @@ class ScheduleProcessor( } } - /** 每次航班状态变化登记两个事件：KAFKA_SCHD 发完整状态，KAFKA_MSG 发一条变更通知。 */ - private fun snapshotEvents(next: FlightSnapshot): List { - val createdAt = clock.instant() - val payload = linkedMapOf( - "flid" to next.flid, - "stateVersion" to next.stateVersion, - "scalars" to next.scalars, - "collections" to next.collections, - ) - val notify = mapper.writeValueAsString(mapOf("flid" to next.flid, "stateVersion" to next.stateVersion)) - return listOf( - MsgEvent( - target = Targets.KAFKA_SCHD, - partitionKey = next.flid, - stateVersion = next.stateVersion, - payloadJson = mapper.writeValueAsString(payload), - createdAt = createdAt, - ), - MsgEvent( - target = Targets.KAFKA_MSG, - partitionKey = next.flid, - stateVersion = next.stateVersion, - payloadJson = notify, - createdAt = createdAt, - ), - ) - } - /** * 写一条处理留痕（仅供排查和统计，不参与业务判断）。放在事务外做， * 写失败也只记日志，不会连累处理结果。scope 是这份报文覆盖的运营日范围。 diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 50beb20..7b5fdb0 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -35,6 +35,9 @@ msgx: arrived-hours: 1 # 条件 5：实际到港早于当前超过 1 小时 snap-log-retention-days: 90 history-store-enabled: false # 未接通历史存储时 HISTORY_SWEEP 删 0 条 + redis: # 航班查询投影（C-11）；真实客户端未接通（G-REDIS-PROJECTION） + enabled: false # 关闭时投影写是空操作；开启但无客户端时写入直接失败，不伪装成功 + flight-key: flightInfo # 航班快照哈希键，field = FLID（沿用旧系统） micronaut: application: diff --git a/src/main/resources/db/migration/V3__msg_event_hold.sql b/src/main/resources/db/migration/V3__msg_event_hold.sql new file mode 100644 index 0000000..3e1bb1a --- /dev/null +++ b/src/main/resources/db/migration/V3__msg_event_hold.sql @@ -0,0 +1,16 @@ +-- ===================================================================== +-- MSG_EVENT 暂存放行：新增 MSG_ID 归属列 +-- --------------------------------------------------------------------- +-- 领域变更与待发事件同一次提交（INV-3），但 Redis 投影写成功之前不能发 Kafka（INV-10）。 +-- 事件因此先以 STATE='HELD' 落库，终态事务按 MSG_ID 放行成 'PENDING'；投递只领 'PENDING'。 +-- 归属列也是重处理的清残依据：崩溃或投影失败后重跑，先删同一 MSG_ID 的 HELD 行再重新登记， +-- 既不留悬挂行，也不把同一次变更投两遍（docs/implementation.md「事务边界」）。 +-- 只作用于自有 PostgreSQL，共享 MySQL 信箱不受影响（C-2）。 +-- ===================================================================== + +ALTER TABLE msg_event ADD COLUMN msg_id BIGINT; + +COMMENT ON COLUMN msg_event.msg_id IS '登记该事件的信箱消息 ID；HELD 行按它清理与放行，非管道写入方可为空'; + +-- 暂存行按消息清理/放行，行数远小于全表，故只索引 HELD。 +CREATE INDEX idx_evt_held ON msg_event (msg_id) WHERE state = 'HELD'; diff --git a/src/main/resources/db/migration/oracle11g/README.md b/src/main/resources/db/migration/oracle11g/README.md index 6998b4d..e798c3d 100644 --- a/src/main/resources/db/migration/oracle11g/README.md +++ b/src/main/resources/db/migration/oracle11g/README.md @@ -21,7 +21,9 @@ Flyway 配置**；PG 路径使用 `classpath:db/migration`，两者互不混用 - `INSERT ... ON CONFLICT` 改 11g MERGE（OPERATION_DAY 不可变条件，implementation「航班身份与运营日」）。 - 空串按 NULL 的语义回归：显式清空的 presence 信息不得被 11g 空串语义吞掉 （增量报文未携带不清空见 `US-04` AC2，日计划清除语义见 `C-6`）。 -- 11g 无部分索引：`uq_req_open` / `uq_schd_event` / `idx_proc_backfill_due` 三个带 `WHERE` - 的索引必须换成等价的函数索引或冗余列方案，不能照搬 PG 定义。 +- 11g 无部分索引：`uq_req_open` / `uq_schd_event` / `idx_proc_backfill_due` / `idx_evt_held` + 四个带 `WHERE` 的索引必须换成等价的函数索引或冗余列方案，不能照搬 PG 定义。 +- 11g 无 `pg_advisory_lock`：跨事务持 `PIPELINE_LOCK` 的会话锁须另找等价物 + （`DBMS_LOCK` 或独占连接上的长事务），语义见 `docs/implementation.md`「事务边界」。 版本号与 PG location 各自独立推进，禁止复用版本号语义。 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 203d234..60d4bd0 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 @@ -11,10 +11,10 @@ import java.sql.DriverManager /** * 在真实 PostgreSQL 上跑一遍迁移链，确认结果符合预期：`V1__flight_state_baseline.sql` - * 基线加 `V2__flight_chute_class_type_rename.sql` 更名依序执行成功，该建的表和 PIPELINE_LOCK - * 单行种子都在，回填事实落在基线里，`INBOX_CURSOR`、`BACKFILL_TODO`、 - * `idx_evt_flid`、`PROC_STATE` 的处理开始时间列都不复存在；FLIGHT_CHUTE 的类字段列 - * 已由 V2 更名为 CCLS/CTYP（SIS 口径）。 + * 基线加 `V2__flight_chute_class_type_rename.sql` 更名、`V3__msg_event_hold.sql` 事件归属列 + * 依序执行成功，该建的表和 PIPELINE_LOCK 单行种子都在，回填事实落在基线里，`INBOX_CURSOR`、 + * `BACKFILL_TODO`、`idx_evt_flid`、`PROC_STATE` 的处理开始时间列都不复存在；FLIGHT_CHUTE + * 的类字段列已由 V2 更名为 CCLS/CTYP（SIS 口径）。 * * 没有可用的 PostgreSQL 时跳过（不假装通过）。 */ @@ -51,11 +51,13 @@ class FlywayMigrationTest { while (rs.next()) { records.add(Triple(rs.getString("version"), rs.getString("script"), rs.getBoolean("success"))) } - assertEquals(2, records.size, "迁移链应为 V1 基线 + V2 更名两条") + assertEquals(3, records.size, "迁移链应为 V1 基线 + V2 更名 + V3 事件归属三条") assertEquals("1", records[0].first) assertEquals("V1__flight_state_baseline.sql", records[0].second) assertEquals("2", records[1].first) assertEquals("V2__flight_chute_class_type_rename.sql", records[1].second) + assertEquals("3", records[2].first) + assertEquals("V3__msg_event_hold.sql", records[2].second) assertTrue(records.all { it.third }) } @@ -183,6 +185,21 @@ class FlywayMigrationTest { assertEquals(0, rs.getInt(1), "基线必须不含已无消费者的 idx_evt_flid") } + // V3：事件归属列与暂存行的部分索引（HELD 按 MSG_ID 清理与放行） + stmt.executeQuery( + "SELECT count(*) FROM information_schema.columns WHERE table_name = 'msg_event' " + + "AND column_name = 'msg_id'", + ).use { rs -> + assertTrue(rs.next()) + assertEquals(1, rs.getInt(1), "V3 必须加上 MSG_EVENT.MSG_ID") + } + stmt.executeQuery( + "SELECT indexdef FROM pg_indexes WHERE tablename = 'msg_event' AND indexname = 'idx_evt_held'", + ).use { rs -> + assertTrue(rs.next(), "V3 必须建立暂存行索引 idx_evt_held") + assertTrue(rs.getString(1).contains("WHERE"), "idx_evt_held 只索引 HELD 行") + } + // 基线：投递确认时间 SENT_AT 是保留期判定的唯一基准 stmt.executeQuery( "SELECT count(*) FROM information_schema.columns WHERE table_name = 'msg_event' " + diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcMsgEventUpsertPgTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcMsgEventUpsertPgTest.kt index bb7b6c3..7df5745 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcMsgEventUpsertPgTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcMsgEventUpsertPgTest.kt @@ -69,6 +69,34 @@ class JdbcMsgEventUpsertPgTest { assertNull(row.lastError) } + @Test + fun `held events are invisible to delivery until released and stale holds are discarded`() { + val repo = JdbcMsgEventRepository(dataSource(), Clock.systemUTC()) + val key = "PG-" + UUID.randomUUID().toString().take(8) + val msgId = System.nanoTime() + + repo.insertAll(listOf(held(msgId, key, Targets.KAFKA_MSG, 1), held(msgId, key, Targets.KAFKA_SCHD, 1))) + + assertTrue(repo.claimBatch(Targets.KAFKA_MSG, 1000).none { it.partitionKey == key }, "HELD 不进投递批") + assertTrue(pendingSchd(repo, key).isEmpty(), "HELD 的整态行也领不到") + + // 重处理：上一轮的暂存行先清掉，再重新登记 + assertEquals(2, repo.discardHeld(msgId)) + repo.insertAll(listOf(held(msgId, key, Targets.KAFKA_MSG, 2), held(msgId, key, Targets.KAFKA_SCHD, 2))) + assertEquals(2, repo.releaseHeld(msgId)) + + assertEquals(1, repo.claimBatch(Targets.KAFKA_MSG, 1000).count { it.partitionKey == key }) + assertEquals(2, pendingSchd(repo, key).single().stateVersion) + assertEquals(0, repo.releaseHeld(msgId), "已放行的行不会被重复放行") + } + + private fun held(msgId: Long, key: String, target: String, version: Long) = + MsgEvent( + msgId = msgId, target = target, partitionKey = key, stateVersion = version, + payloadJson = """{"v":$version}""", state = com.gzzn.omms.msgexchange.domain.EventStatus.HELD, + createdAt = now, + ) + private fun pendingSchd(repo: JdbcMsgEventRepository, key: String): List = repo.mergePendingSchd(now, 1000).filter { it.partitionKey == key } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPipelineLockPgTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPipelineLockPgTest.kt new file mode 100644 index 0000000..9c94ec7 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPipelineLockPgTest.kt @@ -0,0 +1,110 @@ +package com.gzzn.omms.msgexchange.infra.persistence.jdbc + +import com.gzzn.omms.msgexchange.support.PgTestSupport +import com.zaxxer.hikari.HikariConfig +import com.zaxxer.hikari.HikariDataSource +import org.flywaydb.core.Flyway +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertThrows +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Assumptions.assumeTrue +import org.junit.jupiter.api.Test +import java.util.UUID +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import javax.sql.DataSource + +/** + * 在真实 PostgreSQL 上验证跨事务持锁：`holdAcrossTransactions` 把连接钉住并持会话级咨询锁， + * 块内的多次事务各自独立提交，另一个写者要等整段结束才拿得到锁 + * （docs/implementation.md「事务边界」）。拿不到 PG 时跳过，不假装通过。 + */ +class JdbcPipelineLockPgTest { + + @Test + fun `each transaction inside one hold commits on its own`() { + val ds = dataSource() + val lock = JdbcPipelineLockRepository(ds) + val tx = JdbcPipelineTransactionManager(ds) + val key = "LOCK-" + UUID.randomUUID().toString().take(8) + + // 第一段提交后中途失败：模拟"领域已提交、投影写挂了" + assertThrows(IllegalStateException::class.java) { + lock.holdAcrossTransactions { + tx.inTransaction { + lock.lock() + insertHeldEvent(ds, key) + } + error("projection-down") + } + } + + assertEquals(1, countEvents(ds, key), "已提交的第一段不因后续失败而回滚") + } + + @Test + fun `a second writer waits until the whole hold is released`() { + val ds = dataSource() + val lock = JdbcPipelineLockRepository(ds) + val tx = JdbcPipelineTransactionManager(ds) + val holding = CountDownLatch(1) + val release = CountDownLatch(1) + val secondAcquired = CountDownLatch(1) + + val holder = Thread.ofPlatform().daemon(true).start { + lock.holdAcrossTransactions { + tx.inTransaction { lock.lock() } // 第一段事务提交，行锁随之释放 + holding.countDown() + release.await() + } + } + val contender = Thread.ofPlatform().daemon(true).unstarted { + tx.inTransaction { + lock.lock() + secondAcquired.countDown() + } + } + try { + assertTrue(holding.await(10, TimeUnit.SECONDS), "持锁方必须先进入") + contender.start() + assertFalse(secondAcquired.await(500, TimeUnit.MILLISECONDS), "行锁已释放，但整段仍由咨询锁挡着") + release.countDown() + assertTrue(secondAcquired.await(10, TimeUnit.SECONDS), "整段结束后第二个写者应立刻拿到锁") + } finally { + release.countDown() + holder.join(5_000) + contender.join(5_000) + } + } + + private fun insertHeldEvent(ds: DataSource, key: String) { + ds.update( + "INSERT INTO msg_event (target, partition_key, event_type, state_version, payload_json, state, attempts, created_at, msg_id) " + + "VALUES ('KAFKA:msg', ?, 'UPSERT', 1, '{}', 'HELD', 0, now(), 1)", + { ps -> ps.setString(1, key) }, + ) + } + + private fun countEvents(ds: DataSource, key: String): Int = + ds.queryOne("SELECT count(*) AS n FROM msg_event WHERE partition_key = ?", { ps -> ps.setString(1, key) }) { + it.getInt("n") + } ?: 0 + + private fun dataSource(): DataSource { + assumeTrue(PgTestSupport.canConnect(), PgTestSupport.skipMessage()) + Flyway.configure() + .dataSource(PgTestSupport.jdbcUrl, PgTestSupport.user, PgTestSupport.password) + .locations("classpath:db/migration") + .load() + .migrate() + return HikariDataSource( + HikariConfig().apply { + jdbcUrl = PgTestSupport.jdbcUrl + username = PgTestSupport.user + password = PgTestSupport.password + maximumPoolSize = 4 + }, + ) + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FdelAndAdftProcessorTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FdelAndAdftProcessorTest.kt index 423763b..642416e 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FdelAndAdftProcessorTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FdelAndAdftProcessorTest.kt @@ -63,8 +63,9 @@ class FdelAndAdftProcessorTest { val proc = StubProcState() proc.insertIfAbsent(msgId, null) - val result = FdelProcessor(StubPipelineTx(), StubPipelineLock(), f, events, proc, ObjectMapper(), java.time.Clock.systemUTC()) - .apply(head(), msg(), FlopPayload("121")) + val result = FdelProcessor( + testCommit(procState = proc, msgEvents = events), f, events, ObjectMapper(), java.time.Clock.systemUTC(), + ).apply(head(), msg(), FlopPayload("121")) assertEquals(ApplyResult.Succeeded, result) assertEquals(FlightState.DELETED, f.findMainRow("121")!!.state) @@ -84,11 +85,15 @@ class FdelAndAdftProcessorTest { fun `repeated FDEL is idempotent without version bump or duplicate event`() { val f = flights() val events = StubMsgEvents() - val proc = FdelProcessor(StubPipelineTx(), StubPipelineLock(), f, events, StubProcState(), ObjectMapper(), java.time.Clock.systemUTC()) + val proc = FdelProcessor( + testCommit(msgEvents = events), f, events, ObjectMapper(), java.time.Clock.systemUTC(), + ) proc.apply(head(), msg(), FlopPayload("121")) val version = f.findMainRow("121")!!.stateVersion - val result = proc.apply(head(), msg(), FlopPayload("121")) + // 第二份删除报文是另一条消息（同一条消息到终态后不会再进主泵） + val later = ProcState(msgId + 1, ProcStatus.PENDING, updatedAt = java.time.Instant.EPOCH) + val result = proc.apply(later, msg(), FlopPayload("121")) assertEquals(ApplyResult.Succeeded, result) // 重复的删除报文算成功，但不再动数据 assertEquals(version, f.findMainRow("121")!!.stateVersion) @@ -100,8 +105,9 @@ class FdelAndAdftProcessorTest { val f = StubFlightState() val events = StubMsgEvents() - val result = FdelProcessor(StubPipelineTx(), StubPipelineLock(), f, events, StubProcState(), ObjectMapper(), java.time.Clock.systemUTC()) - .apply(head(), msg(), FlopPayload("999")) + val result = FdelProcessor( + testCommit(msgEvents = events), f, events, ObjectMapper(), java.time.Clock.systemUTC(), + ).apply(head(), msg(), FlopPayload("999")) assertEquals(ApplyResult.Succeeded, result) // 航班不存在（迟到或多余）也算成功 assertEquals(0, events.rows.size) @@ -113,7 +119,7 @@ class FdelAndAdftProcessorTest { f.markDeleted("121", msgId = 2, now = java.time.Instant.now()) val events = StubMsgEvents() val adft = AdftProcessor( - StubPipelineTx(), StubPipelineLock(), f, events, StubProcState(), + testCommit(msgEvents = events), f, events, OperationDayProps().apply { zone = "Asia/Shanghai" }, ObjectMapper(), java.time.Clock.systemUTC(), ) val record = com.gzzn.omms.msgexchange.domain.flight.ScheduleRecord( @@ -134,7 +140,7 @@ class FdelAndAdftProcessorTest { val f = StubFlightState() val events = StubMsgEvents() val adft = AdftProcessor( - StubPipelineTx(), StubPipelineLock(), f, events, StubProcState(), + testCommit(msgEvents = events), f, events, OperationDayProps().apply { zone = "Asia/Shanghai" }, ObjectMapper(), java.time.Clock.systemUTC(), ) @@ -160,7 +166,7 @@ class FdelAndAdftProcessorTest { msgId = 1, now = java.time.Instant.now(), ) val adft = AdftProcessor( - StubPipelineTx(), StubPipelineLock(), f, StubMsgEvents(), StubProcState(), + testCommit(), f, StubMsgEvents(), OperationDayProps().apply { zone = "Asia/Shanghai" }, ObjectMapper(), java.time.Clock.systemUTC(), ) diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommitTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommitTest.kt new file mode 100644 index 0000000..1bc3cec --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommitTest.kt @@ -0,0 +1,263 @@ +package com.gzzn.omms.msgexchange.processing + +import com.fasterxml.jackson.databind.ObjectMapper +import com.gzzn.omms.msgexchange.MutableClock +import com.gzzn.omms.msgexchange.codec.DecodeResult +import com.gzzn.omms.msgexchange.codec.FlopPayload +import com.gzzn.omms.msgexchange.codec.XmlCodec +import com.gzzn.omms.msgexchange.config.OperationDayProps +import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.delivery.Dispatcher +import com.gzzn.omms.msgexchange.domain.DecodedMessage +import com.gzzn.omms.msgexchange.domain.ErrorClass +import com.gzzn.omms.msgexchange.domain.EventStatus +import com.gzzn.omms.msgexchange.domain.MetaFields +import com.gzzn.omms.msgexchange.domain.MsgKind +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.FlightSnapshot +import com.gzzn.omms.msgexchange.domain.flight.FlightState +import com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters +import com.gzzn.omms.msgexchange.infra.persistence.PipelineLockRepository +import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager +import com.gzzn.omms.msgexchange.infra.retry.FailureScheduler +import com.gzzn.omms.msgexchange.infra.retry.ProcFailure +import com.gzzn.omms.msgexchange.infra.stub.StubDeliveryPort +import com.gzzn.omms.msgexchange.infra.stub.StubFlightProjectionPort +import com.gzzn.omms.msgexchange.infra.stub.StubFlightState +import com.gzzn.omms.msgexchange.infra.stub.StubInbox +import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents +import com.gzzn.omms.msgexchange.infra.stub.StubProcState +import com.gzzn.omms.msgexchange.infra.stub.StubSnapshotLog +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Assertions.assertThrows +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import java.time.Duration +import java.time.LocalDate + +/** + * 三步提交（`INV-3`、`INV-10`）的故障注入： + * 投影写不成功时，已提交的航班数据不撤销，但这条消息不进终态、不记回填意图， + * 待发事件停在 `HELD` 发不出去；恢复后重处理只留一次效果。 + */ +class FlightCommitTest { + + private val clock = MutableClock(MutableClock.BASE) + private val mapper = ObjectMapper() + private val msgId = 21L + private val flid = "121" + + private fun head() = ProcState(msgId, ProcStatus.PENDING, updatedAt = MutableClock.BASE) + + /** 处理器不读报文体以外的内容，几个用例共用这一条 FDEL 形状的解码结果。 */ + private fun msg() = DecodedMessage( + meta = MetaFields("AODB", "FDEL", "", 9L, 1L), + kind = MsgKind.Fdel, + rawXml = "", + body = FlopPayload(flid), + ) + + /** 一套内存装置：航班 121 已在库、版本 1。 */ + private inner class Fixture { + val flights = StubFlightState() + val events = StubMsgEvents() + val procState = StubProcState() + val projection = StubFlightProjectionPort() + + init { + flights.persistFullState( + FlightSnapshot(flid, LocalDate.of(2026, 12, 15), FlightState.ACTIVE, 1, mapOf("FLNO" to "CA001"), emptyMap()), + msgId = 1, + now = MutableClock.BASE, + ) + procState.insertIfAbsent(msgId, MutableClock.BASE) + } + + /** 每次新建处理器：等价于重启后由新进程接着处理同一条消息。 */ + fun flop() = FlopProcessor(commit(), flights, events, mapper, clock) + + fun fdel() = FdelProcessor(commit(), flights, events, mapper, clock) + + private fun commit() = testCommit( + procState = procState, msgEvents = events, projection = projection, clock = clock, + ) + } + + @Test + fun `projection failure keeps committed flight data but leaves the message unfinished`() { + val f = Fixture() + f.projection.failWith = IllegalStateException("redis-down") + + assertThrows(FlightProjectionFailure::class.java) { f.flop().apply(head(), msg(), FlopPayload(flid)) } + + // 已提交的 PG 不因投影失败而撤销（US-03 AC3） + assertEquals(2L, f.flights.findMainRow(flid)!!.stateVersion) + // 不记终态、不记回填意图（INV-10） + val row = f.procState.find(msgId)!! + assertEquals(ProcStatus.PENDING, row.state) + assertNull(row.backfillNextAt) + // 事件登记了但发不出去：投递只领 PENDING + assertTrue(f.events.rows.values.all { it.state == EventStatus.HELD }) + assertTrue(f.events.claimBatch(Targets.KAFKA_MSG, 10).isEmpty()) + assertTrue(f.events.mergePendingSchd(clock.instant(), 10).isEmpty()) + } + + @Test + fun `reprocessing after recovery releases each event exactly once`() { + val f = Fixture() + f.projection.failWith = IllegalStateException("redis-down") + assertThrows(FlightProjectionFailure::class.java) { f.flop().apply(head(), msg(), FlopPayload(flid)) } + + f.projection.failWith = null + val result = f.flop().apply(head(), msg(), FlopPayload(flid)) // 新处理器 = 重启后重处理 + + assertEquals(ApplyResult.Succeeded, result) + val row = f.procState.find(msgId)!! + assertEquals(ProcStatus.SUCCEEDED, row.state) + assertNotNull(row.backfillNextAt) + // 上一轮的暂存行先清后补：通知不重复，整态仍是每个 FLID 一行 + assertEquals(1, f.events.rows.values.count { it.target == Targets.KAFKA_MSG }) + assertEquals(1, f.events.rows.values.count { it.target == Targets.KAFKA_SCHD }) + assertTrue(f.events.rows.values.all { it.state == EventStatus.PENDING }) + assertEquals(3L, f.flights.findMainRow(flid)!!.stateVersion) // 领域再跑一次，版本推进 + assertTrue(f.projection.snapshots.containsKey(flid)) + } + + @Test + fun `FDEL removes the flight from the projection`() { + val f = Fixture() + f.projection.snapshots[flid] = """{"flid":"121"}""" + + val result = f.fdel().apply(head(), msg(), FlopPayload(flid)) + + assertEquals(ApplyResult.Succeeded, result) + assertFalse(f.projection.snapshots.containsKey(flid)) // INV-8：标删即从投影移除 + assertTrue(f.projection.writes.single() is FlightProjectionWrite.Delete) + assertEquals(ProcStatus.SUCCEEDED, f.procState.find(msgId)!!.state) + } + + @Test + fun `projection failure on FDEL keeps the deletion committed and the projection entry untouched`() { + val f = Fixture() + f.projection.snapshots[flid] = """{"flid":"121"}""" + f.projection.failWith = IllegalStateException("redis-down") + + assertThrows(FlightProjectionFailure::class.java) { f.fdel().apply(head(), msg(), FlopPayload(flid)) } + + assertEquals(FlightState.DELETED, f.flights.findMainRow(flid)!!.state) + assertTrue(f.projection.snapshots.containsKey(flid)) // 投影还在，下轮重删 + assertEquals(ProcStatus.PENDING, f.procState.find(msgId)!!.state) + + f.projection.failWith = null + f.fdel().apply(head(), msg(), FlopPayload(flid)) + + assertFalse(f.projection.snapshots.containsKey(flid)) + assertEquals(ProcStatus.SUCCEEDED, f.procState.find(msgId)!!.state) + // 重复的 FDEL 不再推进版本，也不重复登记删除通知 + assertEquals(1, f.events.rows.size) + } + + @Test + fun `lock is held from domain commit through projection write to terminal transaction`() { + val trace = mutableListOf() + val flights = StubFlightState() + flights.persistFullState( + FlightSnapshot(flid, LocalDate.of(2026, 12, 15), FlightState.ACTIVE, 1, mapOf("FLNO" to "CA001"), emptyMap()), + msgId = 1, now = MutableClock.BASE, + ) + val events = StubMsgEvents() + val commit = FlightCommit( + txManager = TracingTx(trace), + lock = TracingLock(trace), + procState = StubProcState(), + msgEvents = events, + projection = TracingProjection(trace), + clock = clock, + ) + + FlopProcessor(commit, flights, events, mapper, clock).apply(head(), msg(), FlopPayload(flid)) + + assertEquals( + listOf("hold", "tx-begin", "lock", "tx-end", "projection", "tx-begin", "lock", "tx-end", "release"), + trace, + ) + } + + @Test + fun `main pump records a retryable failure and delivers nothing when the projection is down`() { + val f = Fixture() + f.projection.failWith = IllegalStateException("redis-down") + val inbox = StubInbox() + inbox.raws[msgId] = "" + val props = PipelineProps() + val opDay = OperationDayProps().apply { zone = "Asia/Shanghai"; cutoffHour = 0 } + val commit = testCommit(procState = f.procState, msgEvents = f.events, projection = f.projection, clock = clock) + val processor = MessageProcessor( + inbox = inbox, + procState = f.procState, + codec = codecReturning(msg()), + scheduleProcessor = ScheduleProcessor(commit, f.procState, f.flights, f.events, StubSnapshotLog(), opDay, mapper, clock), + flopProcessor = FlopProcessor(commit, f.flights, f.events, mapper, clock), + fdelProcessor = FdelProcessor(commit, f.flights, f.events, mapper, clock), + adftProcessor = AdftProcessor(commit, f.flights, f.events, opDay, mapper, clock), + procFailure = ProcFailure(f.procState, FailureScheduler(props, clock)), + props = props, + clock = clock, + operationDayProps = opDay, + counters = PipelineCounters(), + ) + + processor.processOne(head()) + + val row = f.procState.find(msgId)!! + assertEquals(ProcStatus.FAILED, row.state) // 可重试失败，不是终态 + assertEquals(ErrorClass.INFRA, row.errorClass) + assertNull(row.backfillNextAt) // 没有回填意图，信箱标记不会被写回 + + val port = StubDeliveryPort() + val deliveryProps = PipelineProps().apply { pipeline.pollInterval = Duration.ofMillis(1) } + Dispatcher(f.events, port, deliveryProps, FailureScheduler(deliveryProps, clock), PipelineCounters(), mapper).tick() + + assertTrue(port.sent.isEmpty()) // 处理完成前不发 Kafka（INV-3） + } + + private fun codecReturning(decoded: DecodedMessage) = object : XmlCodec { + override fun decode(rawXml: String): DecodeResult = DecodeResult.Ok(decoded) + override fun encodeRqrd(kind: String, rangeJson: String): String = "" + } + + private class TracingTx(private val trace: MutableList) : PipelineTransactionManager { + override fun inTransaction(block: () -> T): T { + trace += "tx-begin" + val result = block() + trace += "tx-end" + return result + } + } + + private class TracingLock(private val trace: MutableList) : PipelineLockRepository { + override fun lock() { + trace += "lock" + } + + override fun holdAcrossTransactions(block: () -> T): T { + trace += "hold" + try { + return block() + } finally { + trace += "release" + } + } + } + + private class TracingProjection(private val trace: MutableList) : FlightProjectionPort { + override fun write(writes: List) { + trace += "projection" + } + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt index b684f93..3d6d7f6 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt @@ -50,11 +50,6 @@ class IgnoreBranchTest { override fun encodeRqrd(kind: String, rangeJson: String): String = "" } - private val noopTx = object : PipelineTransactionManager { - override fun inTransaction(block: () -> T): T = block() - } - private val noopLock = object : PipelineLockRepository { override fun lock() = Unit } - private fun processor( proc: StubProcState = StubProcState(), inbox: StubInbox = StubInbox(), @@ -68,14 +63,15 @@ class IgnoreBranchTest { val events = StubMsgEvents() val log = StubSnapshotLog() val procFailure = ProcFailure(proc, FailureScheduler(props, clock)) + val commit = testCommit(procState = proc, msgEvents = events, clock = clock) return MessageProcessor( inbox = inbox, procState = proc, codec = codec, - scheduleProcessor = ScheduleProcessor(noopTx, noopLock, proc, flights, events, log, opDay, ObjectMapper(), clock), - flopProcessor = FlopProcessor(noopTx, noopLock, flights, events, proc, ObjectMapper(), clock), - fdelProcessor = FdelProcessor(noopTx, noopLock, flights, events, proc, ObjectMapper(), clock), - adftProcessor = AdftProcessor(noopTx, noopLock, flights, events, proc, opDay, ObjectMapper(), clock), + scheduleProcessor = ScheduleProcessor(commit, proc, flights, events, log, opDay, ObjectMapper(), clock), + flopProcessor = FlopProcessor(commit, flights, events, ObjectMapper(), clock), + fdelProcessor = FdelProcessor(commit, flights, events, ObjectMapper(), clock), + adftProcessor = AdftProcessor(commit, flights, events, opDay, ObjectMapper(), clock), procFailure = procFailure, props = props, clock = clock, diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt index 0a7ffc5..0304c18 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt @@ -15,10 +15,7 @@ import com.gzzn.omms.msgexchange.domain.flight.FlightState import com.gzzn.omms.msgexchange.infra.persistence.FlightStateRepository import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository import com.gzzn.omms.msgexchange.infra.persistence.PersistOutcome -import com.gzzn.omms.msgexchange.infra.persistence.PipelineLockRepository import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager -import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository -import com.gzzn.omms.msgexchange.infra.persistence.SnapshotLogRepository import com.gzzn.omms.msgexchange.infra.stub.StubFlightState import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents import com.gzzn.omms.msgexchange.infra.stub.StubProcState @@ -75,8 +72,7 @@ class ScheduleProcessorTest { override fun inTransaction(block: () -> T): T = block() } return ScheduleProcessor( - txManager = txRunner, - lock = object : PipelineLockRepository { override fun lock() = Unit }, + commit = testCommit(procState = proc, msgEvents = events, txManager = txRunner), procState = proc, flightState = flights, msgEvents = events, diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/TestFlightCommit.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/TestFlightCommit.kt new file mode 100644 index 0000000..7f641f4 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/TestFlightCommit.kt @@ -0,0 +1,25 @@ +package com.gzzn.omms.msgexchange.processing + +import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository +import com.gzzn.omms.msgexchange.infra.persistence.PipelineLockRepository +import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager +import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository +import com.gzzn.omms.msgexchange.infra.stub.StubFlightProjectionPort +import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents +import com.gzzn.omms.msgexchange.infra.stub.StubPipelineLock +import com.gzzn.omms.msgexchange.infra.stub.StubPipelineTx +import com.gzzn.omms.msgexchange.infra.stub.StubProcState +import java.time.Clock + +/** + * 处理器用例共用的三步提交装配：默认全内存、投影不失败。 + * 事件仓储要和处理器用同一个实例，否则放行（`HELD` → `PENDING`）落不到处理器写的那些行上。 + */ +internal fun testCommit( + procState: ProcStateRepository = StubProcState(), + msgEvents: MsgEventRepository = StubMsgEvents(), + projection: FlightProjectionPort = StubFlightProjectionPort(), + lock: PipelineLockRepository = StubPipelineLock(), + txManager: PipelineTransactionManager = StubPipelineTx(), + clock: Clock = Clock.systemUTC(), +) = FlightCommit(txManager, lock, procState, msgEvents, projection, clock)