refactor(storage): ACM2-12 落地——自有 PostgreSQL 全内部状态,共享 MySQL 仅信箱

按 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 批次)。
This commit is contained in:
windyboy
2026-09-07 08:21:11 +08:00
parent a3f1da4bcc
commit e0ce863905
10 changed files with 280 additions and 235 deletions
+35 -31
View File
@@ -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/4I1/I2/I5 |
| `delivery/` | Delivery & Projection:每 target 严格 FIFO 投递、schd 聚合 | 流程 3 |
| `reference/` | Reference & Query21 类同步 + 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`:六表 DDLPROC_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_MASTERPG 方言),ACM2-12 定案;
共享 MySQLCMINMSGS/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`=自有 PGenabled=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_MASTERPG 方言)。全新库直接执行即可,
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 航班动态互不干扰。
**共享 MySQLcdairport,他人系统库)**:本系统**不建表/schema**,仅信箱 DML——收报写
`CMINMSGS`、出站写 `COUTMSGS`(他人读取发送);表结构与保留策略归库方管理。收报前需确认
共享库 CMINMSGS 已存在(他人系统提供);本仓库不声明其 schema。
- **事务模型**:与共享库交互均为外部副作用(ACM2-12)——收报=信箱落信成功(返回
CMINMSGS_ID)→ 自有 PG 建 PENDING 入队;处理成功回填 DATE_PROCESSED/STATUS 为最终一致
- **快照 gen**:随航班动态同在 Redisgen→Redis 协议属 U09,当前为过渡占位)。
- **阶段 BFLIGHT_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(评审与实施计划 U01U30);ACMA 系列仅作
归档历史/迁移来源(ACMA-8 v4 / ACMA-6 选型 / ACMA-9 JDK 口径在归档中可溯)。
架构权威,存储边界按 ACM2-11/12 修订)、ACM2-4(脚手架跟踪)、ACM2-10(评审与实施计划
U01U30)、ACM2-11/12(存储边界与共享信箱决策);ACMA 系列仅作归档历史/迁移来源
ACMA-8 v4 / ACMA-6 选型 / ACMA-9 JDK 口径在归档中可溯)。
+41 -25
View File
@@ -3,6 +3,9 @@
> 现行架构权威为 Plane `airport_chengdu_msgexchange_api` 工作区的 **ACM2-3(综合架构 v4**
> 脚手架跟踪 **ACM2-4**,评审与实施计划(U01U30**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 | 编译期 DIKSP`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 实装属 U05ACM2-12 |
| 权威存储 | Redis航班动态 flightInfo + 快照 gen | 仅主泵线程写(I5);Lua 原子覆盖/版本推进(gen 协议重设计属 U09) |
| 投递 | Kafkaacks=all + 幂等) | outbox 模式,经 MSG_EVENT 表中转 |
| 投影(阶段 B | Elasticsearch + Redis 投影 + FLIGHT_STATE | 仅阶段 B 启用(`msgx.phase` |
| 投影(阶段 B | Elasticsearch(历史) | **阶段 B 缓做(ACM2-12**FLIGHT_STATE 不落表,Redis 永续动态权威;历史投影链路不变 |
| 注册中心 | EurekaMicronaut 原生键) | 服务名契约 `msgexchangeapi`(影子 `msgexchangeapi-shadow`)——**U17 未落地**:当前注册名仍取 `micronaut.application.name`=msgexchange-nextgen),`msgx.service-name` 无运行时消费方(见 §8 与 design.md §9 |
| 可观测 | logstash TCPAsync 包装)+ 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 行;不解析报文 | 流程 1I3 | `InboxController` `InboxService` |
| `processing/` | 主泵:FIFO 领取、解码、identity 绑定、纯函数决策、事务2 | 流程 2/4I1/I2/I5 | `Pump` `MessageProcessor` `SnapshotFlow` `Identity` `Handler(Registry)` |
| `ingress/` | 收报:共享信箱写原文 + 自有 PG 建 PENDING(跨库外部副作用,ACM2-12;不解析报文 | 流程 1I3 | `InboxController` `InboxService` |
| `processing/` | 主泵:FIFO 领取、解码、identity 绑定、纯函数决策、自有 PG 事务2 | 流程 2/4I1/I2/I5 | `Pump` `MessageProcessor` `SnapshotFlow` `Identity` `Handler(Registry)` |
| `delivery/` | 投递:每 target 严格 FIFO、schd 聚合 | 流程 3 | `Dispatcher` `SchdAggregation` |
| `jobs/` | 泵作业:清场/归档/投影重建(PUMP_JOB 同队列 | 流程 4/5/7I4 | `JobExecutor` `HistorySweepJob` `ArchiveJob` `ProjectionRebuildJob` |
| `jobs/` | 泵作业:清场/归档/投影重建(PUMP_JOB 自有 PG,作业窗口执行 | 流程 4/5/7I4 | `JobExecutor` `HistorySweepJob` `ArchiveJob` `ProjectionRebuildJob` |
| `codec/` | XML 解码 + 失败分类(MALFORMED vs CODEC_ERROR | 决策 4 前置 | `XmlCodec` `DecodeResult` |
| `domain/` | 状态机枚举、事件/决策模型、Phase 开关 | I1I5 | `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-12ES 投递成功后同线程同步 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 审计)。
- **共享 MySQLcdairport,他人系统库)——本系统不建任何表/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)。
- **阶段 BFLIGHT_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+sensitivedev/影子经
+57 -51
View File
@@ -10,7 +10,7 @@
```
ProcStatusPROC_STATE.STATE,消息处理侧):
PENDING ──处理成功──▶ SUCCEEDED(终态回填 CMINMSGS
PENDING ──处理成功──▶ SUCCEEDED(终态回填共享库 CMINMSGS 为外部副作用,最终一致
│ ──同 identity 已绑定──▶ SKIPPED(终态,lastError=duplicate-of:<id>
└──失败──▶ 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 CASgen 行) |
| REQ_TRACK | 15 类请求状态机 | REGISTERED/SENT/WAITING/DONE/EXPIRED |
| PUMP_JOB | 泵作业队列 | kindARCHIVE/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 | 泵作业调度(作业不插队,队头空闲/退避窗口执行) | kindARCHIVE/HISTORY_SWEEP/PROJECTION_REBUILD |
| REQ_TRACK | 15 类请求状态机 | REGISTERED/SENT/WAITING/DONE/EXPIRED`COUTMSGS_ID BIGINT`U18 修正) |
| REF_MASTER | 21 类静态主数据 | `(RTYPE,RKEY)` PKSOURCE=ADMINAPI/AODB/PIPELINEREFRESHED_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-112026-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同事务;其余跨存储一律外部副作用。
- **共享 MySQLcdairport,他人系统)仅信箱 DML、不建表**:收报写 CMINMSGS、出站写
COUTMSGS(他人读取发送)。见 §3.1/§3.2 的事务模型。
- **Redis**:航班动态 flightInfo + 快照 **SCHD_GENgen**——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(两侧共用):
decodeMALFORMED→DEAD / CODEC_ERROR→FAILED)→ identity 首绑
`tryBindIdentity` 失败 → SKIPPEDI3)→ Schd DNLD → `SnapshotFlow`
其余 → `Handler.decide(redis.hgetAllFlightInfo(), msg)`
阶段 ARedis 先写(I2 happens-beforeTODO redisApply)→ 事务 2
MSG_EVENT 插入 + CMINMSGS 回填 + SUCCEEDED。
阶段 ARedis 先写(I2 happens-beforeTODO 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/D2ACM2-12 缓做):ES 投递成功 → 同线程同步 `insertSync` 删除事件
`deleteOf`Jackson 结构化序列化,refs 可空恒合法 JSON——U14)。当前不启用。
### 3.4 流程 4:日计划快照(`SnapshotFlow`
```
staging(流式解析+整包校验,TODO 阶段2;未实装→FAILED(UNSUPPORTED)
→ Redis Lua SNAPSHOT_REPLACE(同一 hash 原子「覆盖新代+按代差删」,删除集=旧代flids−新代)
putGenIfVersion CASversion 未变才写;CAS 后重放=version 已达标→no-op 成功)
→ SUCCEEDED
CAS 冲突 → FAILED(INFRA)+退避(串行泵下不应发生→告警语义)
gen 版本推进(ACM2-12gen 随 flightInfo 同在 RedisLua 内原子版本 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_SWEEP3:30 清场,I4 同步链):判史 → 同步写 ES → 仅删成功集。
**占位门禁(U10/T07 修订)**ES saveSync 接线前 `pickHistory` 恒空集、删除量恒 0
禁止「全量可删」fail-open 默认;现役五条判史规则 golden 通过后才允许接线。
- ARCHIVE3:00):1 天前且仅终态(SUCCEEDED/SKIPPED/DEAD迁 CMINMSGS_HSTTODO)。
- PROJECTION_REBUILD(阶段 B 切入时全量重建,TODO activeDays)。
- ARCHIVE3: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.1U03):`datasources.default.*`业务 MySQL)、
`datasources.reference.*`21 类静态 PGACM2-11enabled=false 待阶段 6)、
`flyway.datasources.default.*` / `flyway.datasources.reference.*``kafka.producers.default.*`
`eureka.client.*`logback 独立于本文件,环境变量前缀 `MSGX_LOGSTASH_*`
基础设施键位口径(Micronaut 5.1U03):`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 未原子 | 阶段 1Micronaut 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 | WP2legacy 逐字对拍固化) |
| 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 |
| U22U24 | 载荷类型收敛、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 |
@@ -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 批次)。
* 存储边界:自有 PostgreSQLdatasources.default= 本文件除 CminmsgInboxRepository 外
* 的全部接口(消息管道 PROC_STATE/MSG_EVENT、PUMP_JOB、REQ_TRACK、21 类 REF_MASTER);
* 共享 MySQL 信箱(CMINMSGS 收 / COUTMSGS 出)经信箱封装访问,仅 DML、不建表;
* Redis = 航班动态 + 快照 genRefDataRepository 目标实现);FLIGHT_STATE 缓做。
*/
interface ProcStateRepository {
fun insert(cminmsgsId: Long, state: ProcStatus = ProcStatus.PENDING)
@@ -60,9 +64,10 @@ interface MsgEventRepository {
}
/**
* 快照 generationSCHD_GEN)协议——只留 genACM2-11 拆分后)
* 位置:业务事务库(MySQL,与 PROC_STATE/SUCCEEDED 同事务,流程 4 重放幂等依据);
* 21 类静态主数据已迁独立 PostgreSQL 参考库(见 [StaticRefRepository],阶段 6 实装)。
* 快照 generationSCHD_GEN)协议——只留 gen。
* ACM2-12gen 迁 Redis(与 flightInfo 同源,Lua 内原子「覆盖+按代差删+版本推进」,
* DB 仅写 SUCCEEDED;重放幂等由 Lua 承接,协议重设计属 U09)。本接口为过渡占位,
* 目标实现为 Redis gen storescript 化),非关系表。
*/
interface RefDataRepository {
data class GenMeta(val flids: List<String>, val version: Long)
@@ -74,10 +79,10 @@ interface RefDataRepository {
}
/**
* 21 类静态主数据(航空公司/航线/机位/登机桥等)——独立 PostgreSQL 参考库
* `datasources.reference`ACM2-11 定案SOURCE=ADMINAPI/AODB/PIPELINEN19 对齐)。
* 与主处理链路弱事务耦合:写入者为 ReferenceService21 类同步)与请求应答路径
* 不参与主泵事务 2Redis 只作只读热点投影(legacy orms_stand 语义延续,阶段 6)。
* 21 类静态主数据(航空公司/航线/机位/登机桥等)——自有 PostgreSQL `REF_MASTER` 表
* ACM2-12:与消息管道同自有库SOURCE=ADMINAPI/AODB/PIPELINEN19 对齐)。
* 与主弱事务耦合:写入者为 ReferenceService21 类同步)与请求应答路径
* Redis 只作只读热点投影(legacy orms_stand 语义延续,阶段 6)。
*/
interface StaticRefRepository {
fun upsertAll(refs: List<RefUpsert>)
@@ -85,6 +90,10 @@ interface StaticRefRepository {
fun findByType(type: String): List<RefUpsert>
}
/**
* 15 类请求状态机——自有 PG `REQ_TRACK`ACM2-12Reference & 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<Pair<String, String>>)
@@ -129,15 +147,17 @@ interface FlightStateRepository {
fun findByDay(day: String): List<Pair<String, String>>
}
/**
* 收报信箱(共享 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_RECEIVEDI3:无接收唯一约束)。 */
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)
}
@@ -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())
}
@@ -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)) }
+25 -25
View File
@@ -1,12 +1,15 @@
# =====================================================================
# msgexchange-nextgen 配置(ACMA-8 v4
# 口令/端点全部环境变量外置——零入库(ACMA-8 契约冻结 + 安全基线)
# msgexchange-nextgen 配置(ACM2-12 口径
# 口令/端点全部环境变量外置——零入库
# 存储边界:自有 PostgreSQLdatasources.default,全部内部状态)+
# 共享 MySQL 信箱(mailbox.cminmsgs / mailbox.coutmsgs,仅 DML,不建表)+
# Redis(航班动态 + 快照 gen+ Kafka + Eureka/logstash(见 ACM2-12 与 docs)。
# 键位口径:Micronaut 5.1U03):datasources.* / flyway.datasources.* /
# kafka.producers.<name>.* / eureka.client.*Micronaut 原生键,勿改回 datasource 单数或 Spring 风格)。
# kafka.producers.<name>.* / 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/migrationACM2-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-1121 类静态主数据独立 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
# U04R01):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:
@@ -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)。本地事务只在此库。
-- · 共享 MySQLcdairport,他人系统库):仅信箱 DML —— 收报写 CMINMSGS、
-- 出站写 COUTMSGS(他人读取发送);本系统不在此建表/schema(迁移不覆盖)。
-- · Redis:航班动态 flightInfo + 快照 SCHD_GENLua 原子,另行脚本)。
-- · 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|SEQNdecode 后首次绑定(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:schdA起);ES:flight_hts / REDIS:flightInfoB起)
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)
);
@@ -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|SEQNdecode 后首次绑定(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:schdA起);ES:flight_hts / REDIS:flightInfoB起)
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)
);
+2
View File
@@ -3,6 +3,7 @@
# 以便以「注入 PipelineProps/DataSource 断言」离线验证绑定(N02/R09/R01a),不依赖 /env 端点。
# U07stubs=true 提供内存仓储使全链路(Controller→Inbox→Pump→Dispatcher)可装配;
# autostart=false:测试内不自动拉起后台循环(避免泄漏线程),按需手动 tick。
# ACM2-12main 的 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: ""