From e0ce863905d871b90f777913cabeb7bc9c193604 Mon Sep 17 00:00:00 2001 From: windyboy Date: Mon, 7 Sep 2026 08:21:11 +0800 Subject: [PATCH] =?UTF-8?q?refactor(storage):=20ACM2-12=20=E8=90=BD?= =?UTF-8?q?=E5=9C=B0=E2=80=94=E2=80=94=E8=87=AA=E6=9C=89=20PostgreSQL=20?= =?UTF-8?q?=E5=85=A8=E5=86=85=E9=83=A8=E7=8A=B6=E6=80=81=EF=BC=8C=E5=85=B1?= =?UTF-8?q?=E4=BA=AB=20MySQL=20=E4=BB=85=E4=BF=A1=E7=AE=B1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 按 ACM2-12 定案把仓库存储口径从"MySQL 六辅助表"推进到最终形态： - 迁移：删除 MySQL V2.0.0 六表脚本，新增自有 PG V1.0.0 （PROC_STATE/MSG_EVENT/PUMP_JOB/REQ_TRACK/REF_MASTER，PG 方言，REQ_TRACK.COUTMSGS_ID 按 U18 修为 BIGINT；FLIGHT_STATE 阶段 B 缓做不建表）。 - 配置：datasources.default = 自有 PostgreSQL（enabled=false 待 U05）； 移除 datasources.reference/flyway.reference；新增 mailbox.shared-mysql（共享信箱， 仅 DML，不建表）；test profile 显式启用 H2 内存 datasource。 - 接口/注释：Repositories KDoc 按 ACM2-12 归属（自有 PG / 信箱封装 / gen→Redis 占位 / FLIGHT_STATE 缓做）；InboxService、Pump 事务模型注释对正 （信箱外部副作用 + PG 本地事务 + 回填最终一致）。 - 文档：architecture §1/§2/§3/§4/§5(D1/D2)/§6/§7/§8、design §1/§2/§3.1/§3.4/§3.5/§6/§9、 README 全部按 ACM2-12 收口（自有 PG + 共享信箱 + Redis 动态/gen + 阶段 B 缓做）。 验证：37 测试全绿；dev stub 冒烟仍可启动、/health UP、收报 200。 信箱适配层（CminmsgMailbox/OutboxMailbox）、gen→Redis Lua、作业窗口语义、影子重设计 属 ACM2-12 Checks ②③④⑤（U05/U09 批次）。 --- README.md | 66 ++++++----- docs/architecture.md | 66 +++++++---- docs/design.md | 108 +++++++++--------- .../nextgen/infra/persistence/Repositories.kt | 46 +++++--- .../nextgen/ingress/InboxService.kt | 13 ++- .../msgexchange/nextgen/processing/Pump.kt | 5 +- src/main/resources/application.yml | 50 ++++---- .../db/migration/V1.0.0__own_pg_pipeline.sql | 77 +++++++++++++ .../db/migration/V2.0.0__aux_tables.sql | 82 ------------- src/test/resources/application-test.yml | 2 + 10 files changed, 280 insertions(+), 235 deletions(-) create mode 100644 src/main/resources/db/migration/V1.0.0__own_pg_pipeline.sql delete mode 100644 src/main/resources/db/migration/V2.0.0__aux_tables.sql diff --git a/README.md b/README.md index 7ff1c37..efafb13 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,10 @@ # msgexchange-v2（新一代消息交换服务） -依据 **ACMA-8 v4 综合架构**（单写者严格 FIFO 管道 + 两阶段权威）与 **ACMA-6 技术选型** -（Micronaut 5.1 + Kotlin 2.3）搭建的新一代消息交换服务工程。本仓库独立于 legacy -`msgexchange-api`（Java 8 / Spring Boot 1.5 / Maven）——过渡期两套系统并存（影子对拍→ -切流→旧仓库冻结），legacy 维护不受本仓库影响。 +依据 **ACMA-8 v4 综合架构**（单写者严格 FIFO 管道）与 **ACMA-6 技术选型** +（Micronaut 5.1 + Kotlin 2.3）搭建的新一代消息交换服务工程；**存储边界与阶段 B 按 +ACM2-11/ACM2-12 定案**（自有 PostgreSQL 全内部状态 + 共享 MySQL 仅信箱 + Redis 动态/gen， +阶段 B 缓做）。本仓库独立于 legacy `msgexchange-api`（Java 8 / Spring Boot 1.5 / Maven）—— +过渡期两套系统并存（影子对拍→切流→旧仓库冻结），legacy 维护不受本仓库影响。 > **JDK 口径实测修正**：Micronaut 5.1 系构件（如 micronaut-http-server-netty:5.1.10） > 要求 JVM 25+，计划原定 JDK 21 不可行；工程已按 **JDK 25** 配置（ACMA-9 记录）。 @@ -12,7 +13,7 @@ | 包 | 职责 | 对应 ACMA-8 | |---|---|---| -| `ingress/` | Ingress & Inbox：接收事务（事务1），不解析报文 | 流程 1，I3 | +| `ingress/` | Ingress & Inbox：收报（共享信箱落信 → 自有 PG 入队，ACM2-12），不解析报文 | 流程 1，I3 | | `processing/` | Processing 主泵：严格 FIFO 领取、identity 绑定、纯函数决策、事务2 | 流程 2/4，I1/I2/I5 | | `delivery/` | Delivery & Projection：每 target 严格 FIFO 投递、schd 聚合 | 流程 3 | | `reference/` | Reference & Query：21 类同步 + 15 类请求状态机 | 流程 6 | @@ -25,13 +26,14 @@ - `SIS_AODB_RMS-V0.1.md` + `doc/unisysaodbsis.xsd`：**消息结构唯一事实源**（wire 契约冻结， 自 legacy 仓库复制以自包含；codec 实装依据，ACM2-2/ACM2-3）。 -- `db/migration/V2.0.0__aux_tables.sql`：六表 DDL（PROC_STATE / MSG_EVENT / REF_DATA / - REQ_TRACK / PUMP_JOB / FLIGHT_STATE），与 ACMA-8 v4 数据模型节逐字一致；影子实例在 - 独立 schema 执行同一脚本。**ACM2-11 拆分**：21 类静态主数据迁独立 PostgreSQL 参考库 - （`datasources.reference`，见 design.md §2）；`REF_DATA` 留守 SCHD_GEN 行（流程 4 gen CAS）。 -- `lua/snapshot_replace.lua`：同一 hash 原子“覆盖新代 + 按代差删”（流程 4，I4/I5）。 +- `db/migration/V1.0.0__own_pg_pipeline.sql`：**自有 PostgreSQL**（唯一自有库）表结构 + （PROC_STATE / MSG_EVENT / PUMP_JOB / REQ_TRACK / REF_MASTER，PG 方言），ACM2-12 定案； + 共享 MySQL（CMINMSGS/COUTMSGS）仅信箱 DML 不建表；快照 gen 与航班动态在 Redis。 +- `lua/snapshot_replace.lua`：同一 hash 原子“覆盖新代 + 按代差删”（流程 4，I4/I5； + 随 gen→Redis 协议（U09）扩展版本推进）。 - `lua/batch_delete.lua`：3:30 清场批量删除（仅 ES 写成功集，I4）。 -- `application.yml`：口令全部环境变量外置（零入库）；`msgx.phase` 为阶段 A/B 总开关； +- `application.yml`：口令全部环境变量外置（零入库）；`datasources.default`=自有 PG（enabled=false + 待 U05）、`mailbox.shared-mysql`=共享信箱；`msgx.phase` 权威口径（A 现役；B 缓做）； pipeline 参数 = ACMA-8 参数表初值。 ## 未完成（按计划属于后续阶段，不是本脚手架遗漏） @@ -41,27 +43,28 @@ 阶段 1 后续项。 3. **依赖版本锁定**：`gradle/libs.versions.toml` 中版本为计划口径，需阶段 0 「Micronaut×现网 Eureka 互操作冒烟 + logstash + ES REST」通过后固化。 -4. **仓储实装**：`infra/persistence/Repositories.kt` 目前是接口（Micronaut Data JDBC - 实装属阶段 1 后续），主泵/调度循环以接口驱动，纯逻辑已抽离可单测。 -5. **21 类静态主数据参考库**（ACM2-11）：独立 PostgreSQL（`datasources.reference` / - `StaticRefRepository`）已占位（enabled=false），表结构/迁移/同步实装属阶段 6。 +4. **自有 PG 数据层实装**（ACM2-12）：`infra/persistence/Repositories.kt` 目前是接口 + （Micronaut Data JDBC on PG + 信箱适配层 CminmsgMailbox/OutboxMailbox 属 U05 批次）， + 主泵/调度循环以接口驱动，纯逻辑已抽离可单测。 +5. **信箱适配层与共享库边界**（ACM2-12）：CMINMSGS 收 / COUTMSGS 出为共享 MySQL 仅 DML + （mailbox.shared-mysql 配置段已占位）；收报「信箱落信 → PG 入队」与处理回填的 + 外部副作用/补偿模型属 U05 批次。 -## 数据库初始化（U04/R01，务必先读） +## 数据库初始化（ACM2-12 口径） -`db/migration/V2.0.0__aux_tables.sql` **只建六张辅助表**（PROC_STATE / MSG_EVENT / REF_DATA / -REQ_TRACK / PUMP_JOB / FLIGHT_STATE），并假定 `CMINMSGS`（及其历史表）等 legacy 旧表已存在—— -迁移集是「现网 legacy 库的演进」而非「全新库初始化」。因此： +**自有 PostgreSQL**（唯一自有库）：`db/migration/V1.0.0__own_pg_pipeline.sql` 建 +PROC_STATE / MSG_EVENT / PUMP_JOB / REQ_TRACK / REF_MASTER（PG 方言）。全新库直接执行即可， +无 legacy 前置。 -- **现网/影子演进**：在既有 `cdairport` 库（含 CMINMSGS）上执行即可，Flyway 会补跑 V2.0.0。 -- **全新空库（本机/CI/演练/灾备重建）**：需先按 legacy 仓库（`airport_chengdu_msgexchange_api`） - 建好 CMINMSGS/CMINMSGS_HST/COUTMSGS 等旧表（或从现网导出 schema），再启动本服务；否则 - 收报第一句 SQL 即报「表不存在」。影子实例在独立 schema 执行同一脚本时同样先建旧表。 -- 为什么没有 CMINMSGS 的 V1 迁移：六表之外的旧 schema 归 legacy 仓库维护（冻结期）， - 本仓库不重复声明；若未来要求空库一键初始化，再补 V1 基线快照（见 ACM2-10 U04）。 -- 21 类静态主数据（航空公司/航线/机位等）走**独立 PostgreSQL 参考库**（ACM2-11）： - `application.yml` 的 `datasources.reference` 默认 `enabled=false`，阶段 6 实装时置 true - 并设 `MSGX_REF_DB_URL`（参考库迁移目录 `db/ref-migration`）；与业务 MySQL 库（六表+legacy） - 和 Redis 航班动态互不干扰。 +**共享 MySQL（cdairport，他人系统库）**：本系统**不建表/schema**，仅信箱 DML——收报写 +`CMINMSGS`、出站写 `COUTMSGS`（他人读取发送）；表结构与保留策略归库方管理。收报前需确认 +共享库 CMINMSGS 已存在（他人系统提供）；本仓库不声明其 schema。 + +- **事务模型**：与共享库交互均为外部副作用（ACM2-12）——收报=信箱落信成功（返回 + CMINMSGS_ID）→ 自有 PG 建 PENDING 入队；处理成功回填 DATE_PROCESSED/STATUS 为最终一致。 +- **快照 gen**：随航班动态同在 Redis（gen→Redis 协议属 U09，当前为过渡占位）。 +- **阶段 B（FLIGHT_STATE）**：缓做不落表。 +- 影子对拍：自有 PG 开独立 schema；共享信箱为单信箱无法双写，影子输入=只读水位/回放口径。 ## 构建 @@ -97,5 +100,6 @@ MICRONAUT_ENVIRONMENTS=dev ./gradlew run # dev stub 冒烟：内存 stub，无 ## 关联 Plane `airport_chengdu_msgexchange_api`（**ACM2 为现行入口**）：ACM2-3（综合架构 v4， -架构权威）、ACM2-4（脚手架跟踪）、ACM2-10（评审与实施计划 U01–U30）；ACMA 系列仅作 -归档历史/迁移来源（ACMA-8 v4 / ACMA-6 选型 / ACMA-9 JDK 口径在归档中可溯）。 +架构权威，存储边界按 ACM2-11/12 修订）、ACM2-4（脚手架跟踪）、ACM2-10（评审与实施计划 +U01–U30）、ACM2-11/12（存储边界与共享信箱决策）；ACMA 系列仅作归档历史/迁移来源 +（ACMA-8 v4 / ACMA-6 选型 / ACMA-9 JDK 口径在归档中可溯）。 diff --git a/docs/architecture.md b/docs/architecture.md index 0fab526..d904cfc 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -3,6 +3,9 @@ > 现行架构权威为 Plane `airport_chengdu_msgexchange_api` 工作区的 **ACM2-3（综合架构 v4）**； > 脚手架跟踪 **ACM2-4**，评审与实施计划（U01–U30）**ACM2-10**。本文是仓库内的架构速览， > 与代码同步维护；两者冲突时以 ACM2-3 为准并回改本文。 +> **存储边界与阶段 B 以 ACM2-11/ACM2-12 决策为准**（自有 PostgreSQL + 共享 MySQL 信箱 + +> Redis 动态/gen；阶段 B 缓做）——ACM2-3 中"同库事务锚 / MySQL 六辅助表 / 阶段 B"表述 +> 已被上述决策修订。 > 配套设计细节见 [design.md](design.md)。 ## 1. 系统定位 @@ -14,9 +17,12 @@ 影子对拍（同入口双收，比对输出）→ 切流（nextgen 权威）→ 旧仓库冻结 ``` -- legacy 维护不受本仓库影响；本仓库不声明 legacy 旧表 schema（见 §6 数据边界）。 +- legacy 维护不受本仓库影响；本仓库不声明 legacy 旧表 schema，**且不在共享 MySQL 建任何表** + （CMINMSGS/COUTMSGS 所在库属他人系统，本系统仅信箱 DML——ACM2-12，见 §6）。 - wire 契约冻结：消息结构唯一事实源为 `SIS_AODB_RMS-V0.1.md` + `doc/unisysaodbsis.xsd`； HTTP 端点路径与响应语义沿用现役（如 `POST /cminmsgs/send` 返回记录 ID）。 +- **影子对拍口径（ACM2-12）**：共享库是单信箱无法"同入口双收"，影子输入改以共享库只读 + 水位/回放 + 自有 PG 独立 schema 比对（详见 §8 与 ACM2-12 Checks ⑤）。 ## 2. 技术栈 @@ -24,10 +30,10 @@ |---|---|---| | 语言/运行时 | Kotlin 2.3 + JDK 25 | JDK 21 不可行（Micronaut 5.1 系要求 JVM 25+，ACMA-9 实测） | | 框架 | Micronaut 5.1.3 | 编译期 DI：KSP（`kotlin-ksp` + `micronaut-inject-kotlin`）生成 `*$Definition` | -| 持久化 | 业务库：MySQL + Flyway；静态参考库：**PostgreSQL**（ACM2-11） | 仓储现为接口（Micronaut Data JDBC 实装属 U05，阶段 1 后续；`datasources.reference` enabled=false 待阶段 6） | -| 权威存储 | Redis（阶段 A） | flightInfo hash；仅主泵线程写（I5）；Lua 脚本原子覆盖 | +| 持久化 | **自有 PostgreSQL**（全部内部状态）+ 共享 MySQL 信箱 | 自有库：消息管道 PROC_STATE/MSG_EVENT + PUMP_JOB/REQ_TRACK + 21 类 REF_MASTER（迁移 `db/migration`）；共享库仅 CMINMSGS/COUTMSGS DML。Micronaut Data JDBC 实装属 U05（ACM2-12） | +| 权威存储 | Redis（航班动态 flightInfo + 快照 gen） | 仅主泵线程写（I5）；Lua 原子覆盖/版本推进（gen 协议重设计属 U09） | | 投递 | Kafka（acks=all + 幂等） | outbox 模式，经 MSG_EVENT 表中转 | -| 投影（阶段 B） | Elasticsearch + Redis 投影 + FLIGHT_STATE | 仅阶段 B 启用（`msgx.phase`） | +| 投影（阶段 B） | Elasticsearch（历史） | **阶段 B 缓做（ACM2-12）**：FLIGHT_STATE 不落表，Redis 永续动态权威；历史投影链路不变 | | 注册中心 | Eureka（Micronaut 原生键） | 服务名契约 `msgexchangeapi`（影子 `msgexchangeapi-shadow`）——**U17 未落地**：当前注册名仍取 `micronaut.application.name`（=msgexchange-nextgen），`msgx.service-name` 无运行时消费方（见 §8 与 design.md §9） | | 可观测 | logstash TCP（Async 包装）+ MDC traceId + 自定义健康指示器 | 见 [design.md §8](design.md) | @@ -73,15 +79,20 @@ - **统一 FIFO（决策 1）**：定时作业（cron → PUMP_JOB 入队）与消息同队列， 作业产物不绕过队头顺序；job 与队头的先后目前为入队时间近似， 统一序号列属 U15（未实装，见 design.md §9 缺口清单）。 + **ACM2-12 修订**：PUMP_JOB 在自有 PG，作业不插队、仅在消息队头空闲/退避窗口由主泵执行。 +- **存储边界（ACM2-12）**：本框图内事务库为**自有 PostgreSQL**（PROC_STATE/MSG_EVENT/ + PUMP_JOB/REQ_TRACK/21 类）；CMINMSGS/COUTMSGS 在**共享 MySQL 信箱**（外部副作用，非本系统 + 表）；Redis 除 flightInfo 还承载快照 gen。图中"事务 1/2"与 ACM2-3 的"同库事务"语义不同， + 见 §6：本地事务只在自有 PG，信箱交互为外部写+最终一致。 ## 4. 模块职责 | 包 | 职责 | 对应 ACMA-8 | 主要类 | |---|---|---|---| -| `ingress/` | 收报事务1：原文落库 + 伴生 PENDING 行；不解析报文 | 流程 1，I3 | `InboxController` `InboxService` | -| `processing/` | 主泵：FIFO 领取、解码、identity 绑定、纯函数决策、事务2 | 流程 2/4，I1/I2/I5 | `Pump` `MessageProcessor` `SnapshotFlow` `Identity` `Handler(Registry)` | +| `ingress/` | 收报：共享信箱写原文 + 自有 PG 建 PENDING（跨库外部副作用，ACM2-12）；不解析报文 | 流程 1，I3 | `InboxController` `InboxService` | +| `processing/` | 主泵：FIFO 领取、解码、identity 绑定、纯函数决策、自有 PG 事务2 | 流程 2/4，I1/I2/I5 | `Pump` `MessageProcessor` `SnapshotFlow` `Identity` `Handler(Registry)` | | `delivery/` | 投递：每 target 严格 FIFO、schd 聚合 | 流程 3 | `Dispatcher` `SchdAggregation` | -| `jobs/` | 泵作业：清场/归档/投影重建（经 PUMP_JOB 同队列） | 流程 4/5/7，I4 | `JobExecutor` `HistorySweepJob` `ArchiveJob` `ProjectionRebuildJob` | +| `jobs/` | 泵作业：清场/归档/投影重建（PUMP_JOB 自有 PG，作业窗口执行） | 流程 4/5/7，I4 | `JobExecutor` `HistorySweepJob` `ArchiveJob` `ProjectionRebuildJob` | | `codec/` | XML 解码 + 失败分类（MALFORMED vs CODEC_ERROR） | 决策 4 前置 | `XmlCodec` `DecodeResult` | | `domain/` | 状态机枚举、事件/决策模型、Phase 开关 | I1–I5 | `ProcState` `MsgEvent` `Decision` `MsgKind` | | `infra/` | 仓储接口、重试策略、Redis Lua、stub、健康、日志 | 数据模型节 | 见 design.md | @@ -91,8 +102,8 @@ | # | 决策 | 落点 | |---|---|---| -| D1 | 作业与消息同队列（cron 只经 PUMP_JOB 入队，产物不绕过队头） | `Pump.tick` / `JobExecutor` | -| D2 | 阶段 B：ES 投递成功后同线程**同步** enqueue 删除事件（不轮询 ack） | `Dispatcher.tick`（定案 2） | +| D1 | 作业与消息同队列（cron 只经 PUMP_JOB 入队，产物不绕过队头）——**ACM2-12 修订**：PUMP_JOB 迁自有 PG 后不插队，仅队头空闲/退避窗口执行 | `Pump.tick` / `JobExecutor` | +| D2 | 阶段 B（缓做，ACM2-12）：ES 投递成功后同线程同步 enqueue 删除事件（不轮询 ack） | `Dispatcher.tick`（定案 2；阶段 B 评估后启用） | | D3 | schd 唯一出口是 flushSchd 批量聚合（逐条循环显式排除 KAFKA_SCHD） | `Dispatcher.tick`（U06/N03） | | D4 | 未实装 ≠ 非法：无 handler / staging 未实装 → FAILED(UNSUPPORTED) 可重放，绝不写终态 | `MessageProcessor` `SnapshotFlow`（U10/N21） | | D5 | 失败迁移在持有具体 head/batch 的边界完成；loop 只作最后防线，不吞 InterruptedException/Error | `MessageProcessor` `Dispatcher`（U08） | @@ -100,26 +111,30 @@ | D7 | stub 装配门禁：`msgx.stubs=true` 才装配内存实装；与 `autostart` 组合支撑 dev 冒烟 | `infra/stub`（U07/U01） | | D8 | 编译期 DI（KSP）+ 启动期冒烟测试锁定 BeanDefinition 生成 | `build.gradle.kts`（U01） | -## 6. 数据边界 +## 6. 数据边界（ACM2-12 最终口径） -- **本仓库 Flyway 只建六张辅助表**：`PROC_STATE` / `MSG_EVENT` / `REF_DATA` / `REQ_TRACK` / - `PUMP_JOB` / `FLIGHT_STATE`（`V2.0.0__aux_tables.sql`）。 -- **21 类静态主数据 → 独立 PostgreSQL 参考库**（ACM2-11 定案；`datasources.reference` / - `StaticRefRepository`）：航空公司/航线/机位/登机桥等静态数据迁出 `REF_DATA`； - **`REF_DATA` 仅留守 SCHD_GEN 行**（流程 4 gen CAS 与 SUCCEEDED 同事务，见 design.md §2 注）； - 航班动态权威 Redis（阶段 A flightInfo）**不变**。参考库表结构/迁移/21 类同步实装属阶段 6。 -- `CMINMSGS` / `CMINMSGS_HST` / `COUTMSGS` 等 legacy 旧表归 legacy 仓库维护（冻结期）， - 本仓库不重复声明；**全新空库需先建 legacy schema**，否则收报首句 SQL 报表不存在 - （README「数据库初始化」节）。 -- 回滚兼容关键：SUCCEEDED 时回填 `CMINMSGS.SUBSYSTEM_*` + `DATE_PROCESSED/STATUS`， - 旧系统可按自身语义无缝接管（Runbook 第 7 步）。 +- **自有 PostgreSQL（本系统唯一自有数据库）**：全部内部状态，本地事务只在此库成立。 + `db/migration/V1.0.0__own_pg_pipeline.sql`（PG 方言）建：`PROC_STATE`（取消息侧：处理 + 状态/重试/毒丸）+ `MSG_EVENT`（发消息侧：outbox）+ `PUMP_JOB`（作业调度）+ `REQ_TRACK` + （15 类请求）+ `REF_MASTER`（21 类静态，SOURCE 审计）。 +- **共享 MySQL（cdairport，他人系统库）——本系统不建任何表/schema，仅信箱 DML**： + 收报写 `CMINMSGS`（含处理完成回填 DATE_PROCESSED/STATUS）、出站写 `COUTMSGS` + （他人系统读取发送）。与信箱的交互是**外部副作用**，非本系统事务的一部分： + 收报 = 信箱落信成功（返回 CMINMSGS_ID）→ 自有 PG 建 PENDING 入队；PG 建行失败以共享库 + `DATE_PROCESSED IS NULL` 重扫补建；回填 = 处理成功后异步/补偿，最终一致（ACM2-12 影响面）。 +- **Redis**：航班动态 flightInfo（阶段 A 权威，I5）+ 快照 **SCHD_GEN**（Lua 内原子 + 「覆盖+按代差删+版本推进」，协议重设计属 U09）。 +- **阶段 B（FLIGHT_STATE）：缓做，不落表**；ES 仍承载历史航班（判史/查询）， + 相关投影/删除事件链待阶段 B 重评估后启用。 +- legacy 旧表 schema 归 legacy 仓库维护；CMINMSGS/CMINMSGS_HST/COUTMSGS 结构与保留策略 + 由共享库方管理，本系统只按契约读写（权限边界需与库方确认）。 -## 7. 两阶段权威与当前就绪度 +## 7. 权威（阶段 A Redis；阶段 B 缓做）与当前就绪度 | 阶段 | 权威 | 投递目标 | 状态 | |---|---|---|---| -| A（`msgx.phase=A`） | Redis flightInfo | KAFKA:msg、KAFKA:schd | 管道骨架+重试闭环已实装；Redis Lua/实仓储属 U05/U09 | -| B（`msgx.phase=B`） | FLIGHT_STATE + 投影 | + ES:flight_hts、REDIS:flightInfo | 未实施（阶段 2 后） | +| A（`msgx.phase=A`） | Redis flightInfo（+ gen） | KAFKA:msg、KAFKA:schd | 管道骨架+重试闭环已实装；自有 PG 数据层/信箱适配层属 U05；gen→Redis 协议属 U09 | +| B（`msgx.phase=B`） | —（缓做，ACM2-12） | ES 历史投影 | 不落表、不启用；Redis 永续动态权威，阶段 B 价值重评估后再定 | **就绪度（2026-09-07 复核口径）**：可编译、37 测试全绿、**dev stub 进程级冒烟实测可端到端** （`./gradlew run` 无外部依赖启动 → 收报 200 → `/health` UP，修复记录见 README「进程级 dev 冒烟」）； @@ -133,7 +148,8 @@ U13（投递毒丸补全）、U15（统一序号）。逐项状态见 ACM2-10「 - **影子隔离（目标态；U17/U26 未落地，勿按现状引用）**：服务名（`msgexchangeapi-shadow`）+ 独立 schema + Redis key 前缀 + 独立 topic 三层隔离；当前代码仅 `msgx.register-eureka=false` 生效—— Kafka topic 写死字面量 `"msg"`/`"schd"`（Dispatcher）、`FlightRedisClient.eval` 无 key 前缀参数、 - 服务名未接 `msgx.service-name`（§2）。 + 服务名未接 `msgx.service-name`（§2）。影子数据库：自有 PG 开独立 schema/实例；共享信箱无法 + 双写，影子输入=只读水位/回放口径（ACM2-12 Checks ⑤）。 - **网络信任模型**：`/cminmsgs/send` 无鉴权（沿用现役内网信任姿态）；eureka default-zone 回退 `127.0.0.1:8761`；口令/端点全部环境变量外置（零入库）。安全节细化属 U28。 - **管理端点**：Micronaut 5.1 下 `/env` 默认**禁用**、`/beans` 默认 enabled+sensitive；dev/影子经 diff --git a/docs/design.md b/docs/design.md index 53c47f1..d52b79d 100644 --- a/docs/design.md +++ b/docs/design.md @@ -10,7 +10,7 @@ ``` ProcStatus（PROC_STATE.STATE，消息处理侧）： - PENDING ──处理成功──▶ SUCCEEDED（终态，回填 CMINMSGS） + PENDING ──处理成功──▶ SUCCEEDED（终态；回填共享库 CMINMSGS 为外部副作用，最终一致） │ ──同 identity 已绑定──▶ SKIPPED（终态，lastError=duplicate-of:） └──失败──▶ FAILED（非终态，attempts+1 + nextAttemptAt 退避） │ attempts ≥ maxAttempts 或 队头滞留超 head-deadline @@ -46,42 +46,45 @@ ErrorClass（两侧共用）： schdPush / outboundIntents / refUpserts），不触碰 Redis/Kafka——副作用全部由泵边界执行。 - Handler 实装：0/32（骨架），翻译属阶段 2/3，逐条对照 ACM2-4 行为基线与 KEEP/FIX 矩阵。 -## 2. 数据模型（六辅助表 + 静态参考库） +## 2. 数据模型（自有 PostgreSQL · ACM2-12） -`db/migration/V2.0.0__aux_tables.sql`（legacy 旧表不在本仓库声明，见 architecture.md §6）： +`db/migration/V1.0.0__own_pg_pipeline.sql`（PG 方言，自有库；legacy 旧表与共享库表不在 +本仓库声明，见 architecture.md §6）： | 表 | 角色 | 关键列/约束 | |---|---|---| -| PROC_STATE | 消息处理伴生状态（不动 CMINMSGS 旧列） | `UK_PROC_IDENTITY(IDENTITY_KEY)` 唯一约束=I3 依据；`IDX_PROC_HEAD(STATE, CMINMSGS_ID)`=队头查询 | -| MSG_EVENT | 统一投递 outbox | `EVENT_ID` 自增=全序；`IDX_EVT_HEAD(TARGET, STATE, EVENT_ID)`=每 target 队头 | -| REF_DATA | **SCHD_GEN（流程 4 gen 协议，留守业务库）**；原 21 类静态数据已拆分迁独立 PG 参考库（ACM2-11） | `VERSION` 列支撑流程 4 CAS（gen 行） | -| REQ_TRACK | 15 类请求状态机 | REGISTERED/SENT/WAITING/DONE/EXPIRED | -| PUMP_JOB | 泵作业队列 | kind：ARCHIVE/HISTORY_SWEEP/PROJECTION_REBUILD | -| FLIGHT_STATE | 阶段 B 权威 | `replaceDay` 单事务删差集+写新代+版本提升 | +| PROC_STATE | 取消息侧：处理伴生状态/重试/毒丸（与共享库 CMINMSGS_ID 对应） | `uk_proc_identity(IDENTITY_KEY)` 唯一约束=I3 依据；`idx_proc_head(STATE, CMINMSGS_ID)`=队头 | +| MSG_EVENT | 发消息侧：统一投递 outbox | `EVENT_ID` 自增=全序；`idx_evt_head(TARGET, STATE, EVENT_ID)`=每 target 队头 | +| PUMP_JOB | 泵作业调度（作业不插队，队头空闲/退避窗口执行） | kind：ARCHIVE/HISTORY_SWEEP/PROJECTION_REBUILD | +| REQ_TRACK | 15 类请求状态机 | REGISTERED/SENT/WAITING/DONE/EXPIRED；`COUTMSGS_ID BIGINT`（U18 修正） | +| REF_MASTER | 21 类静态主数据 | `(RTYPE,RKEY)` PK；SOURCE=ADMINAPI/AODB/PIPELINE；REFRESHED_AT | -已知 DDL 缺口（U18，未修）：`REQ_TRACK.COUTMSGS_ID` 应 INT→BIGINT；`FLIGHT_STATE` -主键应含 FDAY；时间列 TIMESTAMP（秒级+会话时区）应 DATETIME(6)/显式 UTC，否则退避/毒丸 -判定存在系统性偏移风险。 +已知 DDL 缺口（U18，随本库 PG 化修正/收窄）：REQ_TRACK.COUTMSGS_ID 已按 BIGINT； +时间列须 DATETIME(6)/显式 UTC 口径在 U05 数据层实现时定；FLIGHT_STATE 因缓做不在本库。 -**库边界（ACM2-11，2026-09-07 定案）**： -- 业务事务库（MySQL `cdairport`）：PROC_STATE / MSG_EVENT / REQ_TRACK / PUMP_JOB / - FLIGHT_STATE + `REF_DATA` 的 **SCHD_GEN 行**（gen CAS 与 SUCCEEDED 同事务，快照协议依赖，不能随迁）； -- **21 类静态主数据 → 独立 PostgreSQL 参考库**（`datasources.reference`，`StaticRefRepository`， - SOURCE=ADMINAPI/AODB/PIPELINE）：写少读多、外部来源、弱事务；Redis 只作只读热点投影 - （legacy orms_stand 语义延续，阶段 6）； -- 航班动态权威 Redis（阶段 A flightInfo）**不变**。 -- 实现状态：Repository 已拆（`RefDataRepository`=gen-only / `StaticRefRepository`=21 类）； - reference datasource 已占位（enabled=false）；表结构草案、迁移 SQL、ReferenceService 21 类 - 同步与 codec 应答接线属阶段 6（U05 数据层批次后）。 +**存储边界（ACM2-12 定案）**： +- **自有 PostgreSQL** = 上表全部（消息管道 + 调度 + 请求 + 21 类）。本地事务只在此库： + 处理侧「MSG_EVENT 插入 + PROC_STATE→SUCCEEDED」同事务；其余跨存储一律外部副作用。 +- **共享 MySQL（cdairport，他人系统）仅信箱 DML、不建表**：收报写 CMINMSGS、出站写 + COUTMSGS（他人读取发送）。见 §3.1/§3.2 的事务模型。 +- **Redis**：航班动态 flightInfo + 快照 **SCHD_GEN（gen）**——Lua 内原子「覆盖+按代差删+ + 版本推进」；重放幂等由 Lua 承接（协议重设计属 U09），`RefDataRepository` 为目标实现的 + 过渡占位接口。 +- **FLIGHT_STATE（阶段 B 权威）：缓做不落表**（Redis 永续动态权威）。 +- 实现状态：迁移 SQL 已按 PG 落地（V1.0.0）；Repository 接口归属注释已对正（自有 PG / + 信箱封装 / gen→Redis 占位 / FlightState 缓做）；Micronaut Data 实装与信箱适配层 + （CminmsgMailbox/OutboxMailbox）属 U05 批次。 ## 3. 核心流程设计 ### 3.1 流程 1：收报（`InboxService.accept`） -事务 1 = `insertRaw`（CMINMSGS 原文）+ `procState.insert`（伴生 PENDING 行）； -不解析报文、接收层无唯一约束（I3）。响应 = 记录 ID（「已持久化」语义，与现役逐字对拍 -后固化，U16）。`wakePump()` 目前为 TODO 空操作——泵 1s 轮询兜底，唤醒仅为加速。 -事务边界随 U05（@Transactional + allopen）补齐。 +收报 = 共享信箱外部写 + 自有 PG 入队（ACM2-12，跨库，非同一事务）： +1. `insertRaw`：共享 MySQL CMINMSGS 落信成功（返回其主键 CMINMSGS_ID）——「已持久化」 + 语义即刻成立，响应可返回（与现役逐字对拍后固化，U16）； +2. `procState.insert(id)`：自有 PG 建 PENDING 行入队；本步失败 → 补偿（以共享库 + `DATE_PROCESSED IS NULL` 重扫补建，U05 批次实现）； +3. 不解析报文、接收层无唯一约束（I3）。`wakePump()` 为 TODO 空操作——泵 1s 轮询兜底。 ### 3.2 流程 2：主泵 tick（`Pump.tick`） @@ -102,8 +105,10 @@ ErrorClass（两侧共用）： decode（MALFORMED→DEAD / CODEC_ERROR→FAILED）→ identity 首绑 （`tryBindIdentity` 失败 → SKIPPED，I3）→ Schd DNLD → `SnapshotFlow` → 其余 → `Handler.decide(redis.hgetAllFlightInfo(), msg)` → - 阶段 A：Redis 先写（I2 happens-before，TODO redisApply）→ 事务 2： - MSG_EVENT 插入 + CMINMSGS 回填 + SUCCEEDED。 + 阶段 A：Redis 先写（I2 happens-before，TODO redisApply）→ 自有 PG 事务 2： + MSG_EVENT 插入 + PROC_STATE→SUCCEEDED（同库原子，@Transactional）； + CMINMSGS 回填（DATE_PROCESSED/STATUS）为共享信箱**外部回填**：PG 提交后异步/补偿执行， + 失败重试+告警（最终一致，ACM2-12；回滚接管语义以回填成功为界，需 Runbook 演练确认）。 5. 异常边界（U08）：`processOne` 内 try/catch → `ProcFailure.fail(INFRA)`（attempts+1、 退避、达上限 DEAD）；`InterruptedException` 恢复中断位后**上抛**；loop 仅 catch `Exception` 作最后防线，`Error` 任其终止进程（异常必可见）。 @@ -120,32 +125,32 @@ ErrorClass（两侧共用）： claim、`lastFlush` 仅成功后推进；达上限整批 DEAD(DLQ)。周期/批上限取参数表 （3s / 500）。 - 轮询间隔取参数表（下限 50ms，N18）；无 200ms 硬编码。 -- 阶段 B（定案 2/D2）：ES 投递成功 → 同线程同步 `insertSync` 删除事件 - （`deleteOf`：Jackson 结构化序列化，refs 可空恒合法 JSON——U14）。 +- 阶段 B（定案 2/D2，ACM2-12 缓做）：ES 投递成功 → 同线程同步 `insertSync` 删除事件 + （`deleteOf`：Jackson 结构化序列化，refs 可空恒合法 JSON——U14）。当前不启用。 ### 3.4 流程 4：日计划快照（`SnapshotFlow`） ``` staging（流式解析+整包校验，TODO 阶段2；未实装→FAILED(UNSUPPORTED)） → Redis Lua SNAPSHOT_REPLACE（同一 hash 原子「覆盖新代+按代差删」，删除集=旧代flids−新代） - → putGenIfVersion CAS（version 未变才写；CAS 后重放=version 已达标→no-op 成功） - → SUCCEEDED -CAS 冲突 → FAILED(INFRA)+退避（串行泵下不应发生→告警语义） + → gen 版本推进（ACM2-12：gen 随 flightInfo 同在 Redis，Lua 内原子版本 CAS—— + 目标实现；现 RefDataRepository/putGenIfVersion 为过渡占位） + → 自有 PG SUCCEEDED ``` -**已知缺口（U09，未定案）**：Lua 与 CAS 分属两存储，非同一事务；崩溃窗口 -（Lua 后/CAS 前、CAS 后/SUCCEEDED 前）与幂等重放判据（版本不二次自增）的显式恢复协议 -待定案并补测试。**实现注**：代码亦未实现「重放 no-op」——CAS 已提交而 SUCCEEDED 未写时， -重放会重新 `getGen` 到已达标版本并再次 CAS 成功（版本 1→2 二次自增），与「版本不二次自增」 -注释不符（今日因 staging 未实装不可达，阶段 2 接 staging 后即成真缺陷）。 +**已知缺口（U09，未定案，ACM2-12 后重设计为 Redis 内协议）**：gen 与 Lua/SUCCEEDED 不再 +分属两存储即可同原子（全部在 Redis Lua）；真正跨存储的窗口收窄为「Lua 已完成、PG SUCCEEDED +未写」——重放判据（版本不二次自增）与按代差删在 Lua 内以版本 CAS 承接，恢复协议待定案并补 +测试。现有代码的「CAS 重放二次自增」缺陷（版本 1→2）与实现注随协议重设计一并消除。 -### 3.5 泵作业（`JobExecutor`，经 PUMP_JOB 同队列） +### 3.5 泵作业（`JobExecutor`；PUMP_JOB 自有 PG，作业窗口执行） - HISTORY_SWEEP（3:30 清场，I4 同步链）：判史 → 同步写 ES → 仅删成功集。 **占位门禁（U10/T07 修订）**：ES saveSync 接线前 `pickHistory` 恒空集、删除量恒 0， 禁止「全量可删」fail-open 默认；现役五条判史规则 golden 通过后才允许接线。 -- ARCHIVE（3:00）：1 天前且仅终态（SUCCEEDED/SKIPPED/DEAD）可迁 CMINMSGS_HST（TODO）。 -- PROJECTION_REBUILD（阶段 B 切入时全量重建，TODO activeDays）。 +- ARCHIVE（3:00）：1 天前且仅终态（SUCCEEDED/SKIPPED/DEAD）迁 CMINMSGS_HST（共享库， + 外部副作用；TODO U05 批次）。 +- PROJECTION_REBUILD（阶段 B 缓做，ACM2-12；重新评估后再启用）。 ## 4. 失败与重试统一设计（U08） @@ -184,10 +189,10 @@ CAS 冲突 → FAILED(INFRA)+退避（串行泵下不应发生→告警语义） | `identity.include-day-boundary` | false | 幂等键日边界（CONFIRM 前禁开） | | `consistency-check.on-startup` / `daily-sample-ratio` | true / 0.01 | 一致性哨兵（实装属 U25） | -基础设施键位口径（Micronaut 5.1，U03）：`datasources.default.*`（业务 MySQL）、 -`datasources.reference.*`（21 类静态 PG，ACM2-11，enabled=false 待阶段 6）、 -`flyway.datasources.default.*` / `flyway.datasources.reference.*`、`kafka.producers.default.*`、 -`eureka.client.*`；logback 独立于本文件，环境变量前缀 `MSGX_LOGSTASH_*`。 +基础设施键位口径（Micronaut 5.1，U03）：`datasources.default.*`（**自有 PostgreSQL**， +ACM2-12）、`flyway.datasources.default.*`、`mailbox.shared-mysql.*`（共享信箱，仅 DML）、 +`kafka.producers.default.*`、`eureka.client.*`；logback 独立于本文件， +环境变量前缀 `MSGX_LOGSTASH_*`。 ## 7. 测试策略 @@ -215,16 +220,17 @@ CAS 冲突 → FAILED(INFRA)+退避（串行泵下不应发生→告警语义） | 项 | 缺口 | 计划 | |---|---|---| -| U05 | 仓储接口无实装；`@Transactional`/allopen 未引入 → 生产 DI 装配失败、事务1/2 未原子 | 阶段 1（Micronaut Data JDBC + allopen + 事务生效回归） | +| U05 | 自有 PG 数据层无实装（PROC_STATE/MSG_EVENT/PUMP_JOB/REQ_TRACK/REF_MASTER + 信箱适配层 CminmsgMailbox/OutboxMailbox）；`@Transactional`/allopen 未引入 → 生产 DI 装配失败 | U05 批次（Micronaut Data JDBC on PG + allopen + 信箱外部副作用与补偿回归） | | U07/U26 | `autostart` 默认关=有意门禁，但生产无 fail-fast；双实例无运行期防护 | fail-fast 定案 + 租约/DB 锁拒启 | -| U09 | 快照跨存储恢复协议未定案（含崩溃重放二次自增版本未实现 no-op，见 §3.4 注） | 崩溃窗口清单 + 幂等判据 + 测试 | +| U09 | gen→Redis 协议未重设计（Lua 内原子版本推进；崩溃窗口=「Lua 完成/PG SUCCEEDED 未写」） | Redis 内版本 CAS + 恢复协议 + 测试（ACM2-12） | | U13 | 投递侧无 createdAt/headDeadline 超时升级、无 DEAD 告警出口；处理侧 head-deadline 判据亦不可达（见 §3.2 注） | WP2 | -| U15 | job 与队头消息无统一全序（入队时间近似） | 统一序号列定案 | +| U15 | job 与队头消息无统一全序（入队时间近似）——ACM2-12 后改口径：作业窗口执行，不追求与消息全序 | 作业窗口语义定稿（ACM2-12 Checks ④） | | U16 | `/cminmsgs/send` 无 @Consumes/字符集（实测 text/plain 415）、无错误路径契约（@ControllerAdvice） | WP2（legacy 逐字对拍固化） | -| U17 | Eureka 注册名仍取 `micronaut.application.name`（=msgexchange-nextgen）；`msgx.service-name` 无运行时消费方 → 影子/切流前注册名与文档契约脱节 | `micronaut.application.name=${msgx.service-name}`（application.yml:9/30） | -| U18 | §2 所列 DDL 缺口 | WP2 | -| U19–U21 | 请求状态机量纲/死分支、identity 绑定静默跳过、REF_DATA 接口/DDL 对齐 | WP2 | +| U17 | Eureka 注册名仍取 `micronaut.application.name`（=msgexchange-nextgen）；`msgx.service-name` 无运行时消费方 → 影子/切流前注册名与文档契约脱节 | `micronaut.application.name=${msgx.service-name}`（application.yml） | +| U18 | DDL 缺口（随 PG 化收窄：REQ_TRACK.COUTMSGS_ID 已 BIGINT；时间列口径 U05 定） | U05 批次 | +| U19–U21 | 请求状态机量纲/死分支、identity 绑定静默跳过、21 类静态接口/表对齐（REF_MASTER/SOURCE） | WP2 | | U22–U24 | 载荷类型收敛、eventSeq 未接线、每报文全量读语义定案 | WP3 | | U25/U28/U30 | 一致性哨兵实装、README 安全节/入口、索引与杂项 | WP3/4 | | U27 | 专有材料（SIS md 703KB / XSD 版权头）治理决策 | WP4（ACL 核验先行） | +| ACM2-11/12 | 存储边界（自有 PG + 共享信箱 + Redis 动态/gen + 阶段 B 缓做）：迁移 SQL/配置/接口注释已按定案调整（V1.0.0 PG）；信箱适配层、gen Lua、作业窗口语义、影子重设计未实装 | ACM2-12 Checks ①–⑥ | | ACM2-11 | 21 类静态主数据独立 PG 参考库（datasources.reference）：Repository 接口已拆（RefDataRepository=gen-only / StaticRefRepository），表结构/迁移/SOURCE 审计/ReferenceService 同步与应答接线未实装 | 阶段 6（U05 批次后）；生产 Oracle 仅可能性，触发条件见 ACM2-11 | diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt index 21cc2a3..cdb32fd 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt @@ -9,7 +9,11 @@ import com.gzzn.omms.msgexchange.nextgen.domain.RefUpsert import java.time.Instant /** - * ACMA-8 v4 仓储接口（接口驱动，主泵/调度循环可单测；Micronaut Data JDBC 实装属阶段 1 后续）。 + * ACM2-12 仓储接口（接口驱动，主泵/调度循环可单测；Micronaut Data JDBC 实装属 U05 批次）。 + * 存储边界：自有 PostgreSQL（datasources.default）= 本文件除 CminmsgInboxRepository 外 + * 的全部接口（消息管道 PROC_STATE/MSG_EVENT、PUMP_JOB、REQ_TRACK、21 类 REF_MASTER）； + * 共享 MySQL 信箱（CMINMSGS 收 / COUTMSGS 出）经信箱封装访问，仅 DML、不建表； + * Redis = 航班动态 + 快照 gen（RefDataRepository 目标实现）；FLIGHT_STATE 缓做。 */ interface ProcStateRepository { fun insert(cminmsgsId: Long, state: ProcStatus = ProcStatus.PENDING) @@ -60,9 +64,10 @@ interface MsgEventRepository { } /** - * 快照 generation（SCHD_GEN）协议——只留 gen（ACM2-11 拆分后）。 - * 位置：业务事务库（MySQL，与 PROC_STATE/SUCCEEDED 同事务，流程 4 重放幂等依据）； - * 21 类静态主数据已迁独立 PostgreSQL 参考库（见 [StaticRefRepository]，阶段 6 实装）。 + * 快照 generation（SCHD_GEN）协议——只留 gen。 + * ACM2-12：gen 迁 Redis（与 flightInfo 同源，Lua 内原子「覆盖+按代差删+版本推进」， + * DB 仅写 SUCCEEDED；重放幂等由 Lua 承接，协议重设计属 U09）。本接口为过渡占位， + * 目标实现为 Redis gen store（script 化），非关系表。 */ interface RefDataRepository { data class GenMeta(val flids: List, val version: Long) @@ -74,10 +79,10 @@ interface RefDataRepository { } /** - * 21 类静态主数据（航空公司/航线/机位/登机桥等）——独立 PostgreSQL 参考库 - * （`datasources.reference`，ACM2-11 定案；SOURCE=ADMINAPI/AODB/PIPELINE，N19 对齐）。 - * 与主处理链路弱事务耦合：写入者为 ReferenceService（21 类同步）与请求应答路径， - * 不参与主泵事务 2；Redis 只作只读热点投影（legacy orms_stand 语义延续，阶段 6）。 + * 21 类静态主数据（航空公司/航线/机位/登机桥等）——自有 PostgreSQL `REF_MASTER` 表 + * （ACM2-12：与消息管道同自有库；SOURCE=ADMINAPI/AODB/PIPELINE，N19 对齐）。 + * 与主链弱事务耦合：写入者为 ReferenceService（21 类同步）与请求应答路径； + * Redis 只作只读热点投影（legacy orms_stand 语义延续，阶段 6）。 */ interface StaticRefRepository { fun upsertAll(refs: List) @@ -85,6 +90,10 @@ interface StaticRefRepository { fun findByType(type: String): List } +/** + * 15 类请求状态机——自有 PG `REQ_TRACK`（ACM2-12；Reference & Query 域）。 + * 与共享库 COUTMSGS（出站信箱）跨库：先 COUTMSGS 落库成功 → 再 markSent，补偿重扫，最终一致。 + */ interface ReqTrackRepository { data class Req( val reqId: Long, @@ -108,6 +117,11 @@ interface ReqTrackRepository { fun markDone(reqId: Long) } +/** + * 泵作业调度记录——自有 PG `PUMP_JOB`（ACM2-12）。 + * 决策 1 修订：作业不插队，仅在消息队头空闲/退避窗口由主泵执行（跨库/异队列无全序）； + * 作业动作本身（归档写共享库 CMINMSGS_HST、清场删 Redis/ES 等）仍在各自目标存储。 + */ interface PumpJobRepository { data class Job(val jobId: Long, val kind: String) // ARCHIVE/HISTORY_SWEEP/PROJECTION_REBUILD @@ -122,6 +136,10 @@ interface PumpJobRepository { fun markFailed(jobId: Long, lastError: String) } +/** + * 阶段 B 权威（ACM2-12：缓做，不落表——航班动态权威保持 Redis；阶段 B 重新 + * 评估后再定是否引入事务化权威）。本接口仅供占位与测试，勿据此建表。 + */ interface FlightStateRepository { /** 阶段 B 权威；replaceDay = 单事务删差集+写新代+版本提升。 */ fun replaceDay(day: String, flights: List>) @@ -129,15 +147,17 @@ interface FlightStateRepository { fun findByDay(day: String): List> } +/** + * 收报信箱（共享 MySQL CMINMSGS，仅 DML——ACM2-12：库属他人系统，本系统不建表）。 + * 事务模型：insertRaw = 外部副作用「信箱落信」（成功即返回其主键 CMINMSGS_ID）→ + * 随后自有 PG 建 PROC_STATE(PENDING) 入队；PG 建行失败以共享库 + * DATE_PROCESSED IS NULL 重扫补建。backfillOnSuccess 为处理成功后的外部回填 + * （DATE_PROCESSED/STATUS，最终一致、失败重试+告警）。 + */ interface CminmsgInboxRepository { - /** 事务 1：原文落库（沿用 SUBSYSTEM_* 列；仅 CLOB/DATE_RECEIVED，I3：无接收唯一约束）。 */ fun insertRaw(rawXml: String): Long fun rawOf(cminmsgsId: Long): String? - /** - * v4 回滚兼容关键：SUCCEEDED 时回填 SUBSYSTEM_*（FIX）+ DATE_PROCESSED/STATUS， - * 旧系统按自身 processed 语义可无缝接管（Runbook 第 7 步）。 - */ fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt index a641bce..0289a35 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt @@ -5,23 +5,24 @@ import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ProcStateRepository import jakarta.inject.Singleton import java.time.Instant -/** ACMA-8 流程 1：入站接收（事务 1）。响应语义 = “已持久化”（与现役一致）；不解析报文（I3）。 */ +/** ACMA-8 流程 1：入站接收。响应语义 = “已持久化”（与现役一致）；不解析报文（I3）。 */ @Singleton class InboxService( private val inbox: CminmsgInboxRepository, private val procState: ProcStateRepository, - // TODO(阶段1后续): 事务边界（@Transactional）随 Micronaut Data 实装补齐； - // pump 唤醒仅加速，崩溃后主泵 1s 轮询兜底。 + // TODO(U05 批次): 事务模型（ACM2-12）——insertRaw 为共享信箱外部写（成功即返回 ID）， + // 随后自有 PG 建 PROC_STATE(PENDING)；两写跨库，PG 建行失败以共享库 + // DATE_PROCESSED IS NULL 重扫补建；pump 唤醒仅加速，崩溃后主泵 1s 轮询兜底。 ) { private val log = org.slf4j.LoggerFactory.getLogger(InboxService::class.java) data class Receipt(val cminmsgsId: Long, val receivedAt: Instant) fun accept(rawXml: String): Receipt { - val id = inbox.insertRaw(rawXml) - procState.insert(id) // 同事务（实装后）；接收层无唯一约束（I3） + val id = inbox.insertRaw(rawXml) // 信箱外部写（ACM2-12，与 PG 跨库） + procState.insert(id) // 自有 PG 入队；接收层无唯一约束（I3） wakePump() - log.info("accepted cminmsgsId={} (tx1)", id) + log.info("accepted cminmsgsId={}", id) return Receipt(id, Instant.now()) } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt index b38e044..ff01f52 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt @@ -198,8 +198,9 @@ class MessageProcessor( if (props.phase == PipelineProps.Phase.A) { // 阶段 A：主泵线程先写 Redis（幂等），先于事件创建（I2）；TODO: redisApply(flightChanges) } - // 事务 2：事件 + CMINMSGS 回填 + SUCCEEDED（实装后 @Transactional）； - // 静态主数据（refUpserts）落独立 PG reference 库，弱事务不入本事务（ACM2-11） + // 事务 2（自有 PG 内原子）：事件 + SUCCEEDED（实装后 @Transactional）； + // CMINMSGS 回填 = 共享信箱外部副作用（最终一致，ACM2-12）； + // 静态主数据（refUpserts）同自有 PG 但弱事务独立提交（21 类 REF_MASTER）。 val events = buildList { decision.msgNotifies.forEach { add(MsgEvent(target = Targets.KAFKA_MSG, payloadJson = it.payloadJson)) } decision.schdPush.forEach { add(MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = it.flid, payloadJson = it.fltrJson)) } diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index b3b4989..0417f73 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -1,12 +1,15 @@ # ===================================================================== -# msgexchange-nextgen 配置（ACMA-8 v4） -# 口令/端点全部环境变量外置——零入库（ACMA-8 契约冻结 + 安全基线） +# msgexchange-nextgen 配置（ACM2-12 口径） +# 口令/端点全部环境变量外置——零入库。 +# 存储边界：自有 PostgreSQL（datasources.default，全部内部状态）+ +# 共享 MySQL 信箱（mailbox.cminmsgs / mailbox.coutmsgs，仅 DML，不建表）+ +# Redis（航班动态 + 快照 gen）+ Kafka + Eureka/logstash（见 ACM2-12 与 docs）。 # 键位口径：Micronaut 5.1（U03）：datasources.* / flyway.datasources.* / -# kafka.producers..* / eureka.client.*（Micronaut 原生键，勿改回 datasource 单数或 Spring 风格）。 +# kafka.producers..* / eureka.client.*（Micronaut 原生键）。 # ===================================================================== msgx: - phase: A # A=Redis 权威（主泵单线程写，I5）；B=FLIGHT_STATE 权威+投影 - service-name: msgexchangeapi # 契约冻结；影子实例用 msgexchangeapi-shadow + 独立表/key前缀/topic + phase: A # 航班动态权威 Redis（主泵单线程写，I5）；阶段 B 缓做（ACM2-12） + service-name: msgexchangeapi # 契约冻结；影子实例用 msgexchangeapi-shadow register-eureka: true # 影子对拍期置 false 或独立服务名 pipeline: # —— ACMA-8 参数表初值（阶段 0 基线校准）—— poll-interval: 1s # KEEP 现役 1s 轮询节奏 @@ -21,7 +24,7 @@ msgx: flush-limit: 500 # 批上限，防积压尖峰 identity: include-day-boundary: false # CONFIRM（矩阵 #11）：SEQN 重置作用域确认前保持关闭 - consistency-check: # 阶段 A 唯一主动一致性哨兵（必选） + consistency-check: # 一致性哨兵（实装属 U25） on-startup: true daily-sample-ratio: 0.01 @@ -32,36 +35,33 @@ micronaut: port: 8080 # 管理端点（U03/N32）：/env、/beans 默认 sensitive；仅开发/影子环境放开——见 application-dev.yml +# 自有 PostgreSQL：全部内部状态（消息管道 PROC_STATE/MSG_EVENT、PUMP_JOB、REQ_TRACK、 +# 21 类 REF_MASTER；迁移 db/migration，ACM2-12）。数据层实装（U05/阶段 6）前 enabled=false。 datasources: default: - url: ${MSGX_DB_URL} - username: ${MSGX_DB_USER} - password: ${MSGX_DB_PASSWORD} - driver-class-name: com.mysql.cj.jdbc.Driver - # ACM2-11：21 类静态主数据独立 PostgreSQL 参考库（reference datasource）。 - # 数据层实装（阶段 6 / U05 批次）前默认 disabled——不建连不解析占位符； - # 接入时置 enabled=true 并设 MSGX_REF_DB_URL（或 host/port/name）等环境变量。 - reference: enabled: false - url: ${MSGX_REF_DB_URL} - username: ${MSGX_REF_DB_USER:reference} - password: ${MSGX_REF_DB_PASSWORD:} + url: ${MSGX_PG_URL} + username: ${MSGX_PG_USER} + password: ${MSGX_PG_PASSWORD} driver-class-name: org.postgresql.Driver flyway: datasources: default: - enabled: true - locations: classpath:db/migration - # U04（R01）：V2.0.0 只建六张辅助表，假定 CMINMSGS/CMINMSGS_HST 等 legacy 旧表已存在； - # 全新库请按 README「数据库初始化」先落 legacy schema（或 baseline），否则收报首句 SQL 会报表不存在。 - reference: - # ACM2-11：21 类静态主数据 PG 库迁移（类表 DDL 草案见 ACM2-11 Checks，阶段 6 落库） enabled: false - locations: classpath:db/ref-migration + locations: classpath:db/migration + +# 共享 MySQL 信箱（他人系统库，仅 DML：收报写 CMINMSGS、出站写 COUTMSGS；不建表/schema， +# ACM2-12）。驱动/依赖与信箱适配层（CminmsgMailbox/OutboxMailbox）随 U05 批次引入。 +mailbox: + shared-mysql: + url: ${MSGX_MAILBOX_URL} + username: ${MSGX_MAILBOX_USER} + password: ${MSGX_MAILBOX_PASSWORD} + driver-class-name: com.mysql.cj.jdbc.Driver redis: - uri: ${MSGX_REDIS_URI} # 阶段 A 权威存储；仅主泵线程写（I5） + uri: ${MSGX_REDIS_URI} # 航班动态权威 + 快照 gen；仅主泵线程写（I5） kafka: bootstrap: diff --git a/src/main/resources/db/migration/V1.0.0__own_pg_pipeline.sql b/src/main/resources/db/migration/V1.0.0__own_pg_pipeline.sql new file mode 100644 index 0000000..ac36411 --- /dev/null +++ b/src/main/resources/db/migration/V1.0.0__own_pg_pipeline.sql @@ -0,0 +1,77 @@ +-- ===================================================================== +-- 自有 PostgreSQL 库（唯一自有数据库）· ACM2-12 定稿口径 +-- --------------------------------------------------------------------- +-- 存储边界（ACM2-12，取代 ACMA-8 v4「MySQL 六辅助表」口径，原 MySQL V2.0.0 +-- 六表脚本已删除，定义见 git 历史）： +-- · 自有 PG：全部内部状态 —— 消息管道（PROC_STATE / MSG_EVENT）+ 调度与请求 +-- （PUMP_JOB / REQ_TRACK）+ 21 类静态主数据（REF_MASTER）。本地事务只在此库。 +-- · 共享 MySQL（cdairport，他人系统库）：仅信箱 DML —— 收报写 CMINMSGS、 +-- 出站写 COUTMSGS（他人读取发送）；本系统不在此建表/schema（迁移不覆盖）。 +-- · Redis：航班动态 flightInfo + 快照 SCHD_GEN（Lua 原子，另行脚本）。 +-- · FLIGHT_STATE（阶段 B 权威）：缓做，不落表（Redis 永续动态权威）。 +-- 21 类静态表 REF_MASTER 为通用 rtype/rkey 形态（SOURCE 审计；gen/SCHD_GEN +-- 已迁 Redis，不再需要 VERSION/CAS 列）。 +-- 影子对拍：本库开独立 schema/实例；共享信箱只读水位/回放（见 ACM2-12 影响面）。 +-- ===================================================================== + +-- ① 收报处理伴生状态（取消息侧；与共享库 CMINMSGS 一一对应，CMINMSGS_ID 沿用其主键） +CREATE TABLE PROC_STATE ( + CMINMSGS_ID BIGINT PRIMARY KEY, + STATE VARCHAR(16) NOT NULL, -- PENDING/FAILED/SUCCEEDED/SKIPPED/DEAD + IDENTITY_KEY VARCHAR(200), -- SNDR|TYPE|STYP|SEQN，decode 后首次绑定（I3） + ATTEMPTS INT NOT NULL DEFAULT 0, + NEXT_ATTEMPT_AT TIMESTAMP, + ERROR_CLASS VARCHAR(20), -- MALFORMED/CODEC_ERROR/EXHAUSTED/INFRA/UNSUPPORTED + LAST_ERROR VARCHAR(1000), + UPDATED_AT TIMESTAMP NOT NULL, + CONSTRAINT uk_proc_identity UNIQUE (IDENTITY_KEY) +); +CREATE INDEX idx_proc_head ON PROC_STATE (STATE, CMINMSGS_ID); -- 主泵队头查询（I1） + +-- ② 统一投递事件 outbox（发消息侧） +CREATE TABLE MSG_EVENT ( + EVENT_ID BIGSERIAL PRIMARY KEY, + TARGET VARCHAR(30) NOT NULL, -- KAFKA:msg / KAFKA:schd（A起）；ES:flight_hts / REDIS:flightInfo（B起） + PARTITION_KEY VARCHAR(64), -- schd 事件恒为 FLID + PAYLOAD_JSON TEXT NOT NULL, + STATE VARCHAR(16) NOT NULL, -- PENDING/SENT/DEAD + ATTEMPTS INT NOT NULL DEFAULT 0, + NEXT_ATTEMPT_AT TIMESTAMP, + ERROR_CLASS VARCHAR(20), + LAST_ERROR VARCHAR(1000), + CREATED_AT TIMESTAMP NOT NULL +); +CREATE INDEX idx_evt_head ON MSG_EVENT (TARGET, STATE, EVENT_ID); -- 每 target 队头（I1 双层） + +-- ③ 泵作业调度记录（作业不插队：仅在消息队头空闲/退避窗口由主泵执行，决策 1 修订） +CREATE TABLE PUMP_JOB ( + JOB_ID BIGSERIAL PRIMARY KEY, + KIND VARCHAR(20) NOT NULL, -- ARCHIVE / HISTORY_SWEEP / PROJECTION_REBUILD + STATE VARCHAR(16) NOT NULL, -- QUEUED/RUNNING/DONE/FAILED + CREATED_AT TIMESTAMP NOT NULL, + UPDATED_AT TIMESTAMP NOT NULL, + LAST_ERROR VARCHAR(1000) +); + +-- ④ 15 类请求状态机（Reference & Query 域；与共享库 COUTMSGS 跨库： +-- 先落 COUTMSGS 成功 → 再标 SENT，补偿重扫，最终一致） +CREATE TABLE REQ_TRACK ( + REQ_ID BIGSERIAL PRIMARY KEY, + REQ_TYPE VARCHAR(20) NOT NULL, + PARAMS_JSON TEXT NOT NULL, + STATE VARCHAR(16) NOT NULL, -- REGISTERED/SENT/WAITING/DONE/EXPIRED + COUTMSGS_ID BIGINT, -- U18 修正：BIGINT（原 MySQL 版 INT） + SENT_AT TIMESTAMP, + COMPLETED_AT TIMESTAMP +); +CREATE INDEX idx_req_open ON REQ_TRACK (REQ_TYPE, STATE); + +-- ⑤ 21 类静态主数据（SOURCE=ADMINAPI/AODB/PIPELINE 审计；弱事务 upsert） +CREATE TABLE REF_MASTER ( + RTYPE VARCHAR(20) NOT NULL, -- AIRL/ARPT/... + RKEY VARCHAR(64) NOT NULL, + PAYLOAD_JSON TEXT NOT NULL, + SOURCE VARCHAR(20) NOT NULL, + REFRESHED_AT TIMESTAMP NOT NULL, + PRIMARY KEY (RTYPE, RKEY) +); diff --git a/src/main/resources/db/migration/V2.0.0__aux_tables.sql b/src/main/resources/db/migration/V2.0.0__aux_tables.sql deleted file mode 100644 index 5bb45bd..0000000 --- a/src/main/resources/db/migration/V2.0.0__aux_tables.sql +++ /dev/null @@ -1,82 +0,0 @@ --- ===================================================================== --- ACMA-8 v4 数据模型（六表）· Flyway 基线迁移 --- 与 issue「综合架构设计」数据模型节逐字一致；影子实例在独立 schema 执行同一脚本。 --- --- U04/R01 前置声明：本脚本只建六张辅助表，假定 CMINMSGS/CMINMSGS_HST/COUTMSGS 等 legacy --- 旧表已存在（本仓库不声明旧表 V1，归 legacy 仓库维护）。全新空库请先按 legacy schema --- 建旧表再执行本脚本，否则收报首句 SQL 会报表不存在（详见 README「数据库初始化」）。 --- ===================================================================== - --- ① 处理状态伴生表（不动 CMINMSGS 旧列； --- 回滚兼容：SUCCEEDED 时回填 CMINMSGS.DATE_PROCESSED/STATUS，旧系统可接管续跑） -CREATE TABLE PROC_STATE ( - CMINMSGS_ID BIGINT PRIMARY KEY, - STATE VARCHAR(16) NOT NULL, -- PENDING/FAILED/SUCCEEDED/SKIPPED/DEAD - IDENTITY_KEY VARCHAR(200) NULL, -- SNDR|TYPE|STYP|SEQN，decode 后首次绑定（I3） - ATTEMPTS INT NOT NULL DEFAULT 0, - NEXT_ATTEMPT_AT TIMESTAMP NULL, - ERROR_CLASS VARCHAR(20) NULL, -- MALFORMED/CODEC_ERROR/EXHAUSTED/INFRA/UNSUPPORTED - LAST_ERROR VARCHAR(1000) NULL, - UPDATED_AT TIMESTAMP NOT NULL, - UNIQUE KEY UK_PROC_IDENTITY (IDENTITY_KEY), - INDEX IDX_PROC_HEAD (STATE, CMINMSGS_ID) -- 主泵队头查询（I1） -); - --- ② 统一投递事件（outbox） -CREATE TABLE MSG_EVENT ( - EVENT_ID BIGINT AUTO_INCREMENT PRIMARY KEY, - TARGET VARCHAR(30) NOT NULL, -- KAFKA:msg / KAFKA:schd（A起）；ES:flight_hts / REDIS:flightInfo（B起） - PARTITION_KEY VARCHAR(64) NULL, -- schd 事件恒为 FLID - PAYLOAD_JSON MEDIUMTEXT NOT NULL, - STATE VARCHAR(16) NOT NULL, -- PENDING/SENT/DEAD - ATTEMPTS INT NOT NULL DEFAULT 0, - NEXT_ATTEMPT_AT TIMESTAMP NULL, - ERROR_CLASS VARCHAR(20) NULL, - LAST_ERROR VARCHAR(1000) NULL, - CREATED_AT TIMESTAMP NOT NULL, - INDEX IDX_EVT_HEAD (TARGET, STATE, EVENT_ID) -- 每 target 队头查询（I1 双层同策略） -); - --- ③ 参考数据 + SCHD generation 元数据（21 类；SCHD_GEN 行 RKEY=日期） -CREATE TABLE REF_DATA ( - RTYPE VARCHAR(20) NOT NULL, -- AIRL/ARPT/... | SCHD_GEN - RKEY VARCHAR(64) NOT NULL, - PAYLOAD_JSON MEDIUMTEXT NOT NULL, -- SCHD_GEN: {"flids":[...],"version":N} - VERSION BIGINT NOT NULL DEFAULT 0, -- putGenIfVersion CAS 依据（流程 4） - SOURCE VARCHAR(20) NOT NULL, -- ADMINAPI / AODB / PIPELINE - REFRESHED_AT TIMESTAMP NOT NULL, - PRIMARY KEY (RTYPE, RKEY) -); - --- ④ 请求状态机（15 类，流程 6） -CREATE TABLE REQ_TRACK ( - REQ_ID BIGINT AUTO_INCREMENT PRIMARY KEY, - REQ_TYPE VARCHAR(20) NOT NULL, - PARAMS_JSON VARCHAR(500) NOT NULL, - STATE VARCHAR(16) NOT NULL, -- REGISTERED/SENT/WAITING/DONE/EXPIRED - COUTMSGS_ID INT NULL, - SENT_AT TIMESTAMP NULL, - COMPLETED_AT TIMESTAMP NULL, - INDEX IDX_REQ_OPEN (REQ_TYPE, STATE) -); - --- ⑤ 泵作业队列（定时任务统一入口：cron 触发入队，主泵 FIFO 执行） -CREATE TABLE PUMP_JOB ( - JOB_ID BIGINT AUTO_INCREMENT PRIMARY KEY, - KIND VARCHAR(20) NOT NULL, -- ARCHIVE / HISTORY_SWEEP / PROJECTION_REBUILD - STATE VARCHAR(16) NOT NULL, -- QUEUED/RUNNING/DONE/FAILED - CREATED_AT TIMESTAMP NOT NULL, - UPDATED_AT TIMESTAMP NOT NULL, - LAST_ERROR VARCHAR(1000) NULL -); - --- ⑥ 航班状态权威（阶段 B 启用；此前 Redis 为权威，I5） -CREATE TABLE FLIGHT_STATE ( - FLID VARCHAR(32) PRIMARY KEY, - FDAY DATE NOT NULL, - VERSION BIGINT NOT NULL, - MAID VARCHAR(32) NULL, - PAYLOAD_JSON MEDIUMTEXT NOT NULL, -- 投影兼容格式的 FLTR JSON - UPDATED_AT TIMESTAMP NOT NULL, - INDEX IDX_FS_DAY (FDAY, VERSION) -); diff --git a/src/test/resources/application-test.yml b/src/test/resources/application-test.yml index 0812b01..c60ee55 100644 --- a/src/test/resources/application-test.yml +++ b/src/test/resources/application-test.yml @@ -3,6 +3,7 @@ # 以便以「注入 PipelineProps/DataSource 断言」离线验证绑定（N02/R09/R01a），不依赖 /env 端点。 # U07：stubs=true 提供内存仓储使全链路（Controller→Inbox→Pump→Dispatcher）可装配； # autostart=false：测试内不自动拉起后台循环（避免泄漏线程），按需手动 tick。 +# ACM2-12：main 的 datasources.default（自有 PG）默认 enabled=false，测试以 H2 内存替代并显式启用。 msgx: register-eureka: false stubs: true @@ -11,6 +12,7 @@ msgx: datasources: default: + enabled: true url: jdbc:h2:mem:cfgtest;DB_CLOSE_DELAY=-1 username: sa password: ""