refactor(flight-state): SCHD 日计划收敛为缺失保留合并语义并对齐现行设计文档
按现行 docs/flight-state.md（111 行版，§1-§7）全面对齐 domain 包及其消费方， 删除整套指向已退役长版文档（§5.x-§10）的引用与死代码： - 语义：FlightStateEngine.snapshotState 由"完整快照整体替换"改为 §3.1 合并语义 ——出现 Set/Replace、缺失保留、标量空串显式清空；DELETED 不被日计划恢复（§3.3）。 - 校验：validateMessage 移除从未接线的报文覆盖范围（scope）参数，只保留 §4 步骤 2 的声明数量/航班标识/运营日推导校验；ScheduleBody 删除 scopeStart/scopeEnd。 - 删除死代码：domain/Decision.kt、FlightModel 的 SnapshotPatch/UpsertOutcome、 ProcState.isTerminal/archivable、SnapshotFlag.SEQN_REGRESSION、 ScheduleRecord.seqn（及 wire FlightRecordXml.SEQN）；codec 移除未消费 FFID。 - 删除未接线且引用已退役列（fday/last_message_id）的 SqlDialect 方言脚手架， oracle11g README 改为按 V1 现列重建的口径。 - 注释/测试：domain、processing、infra 仓储与 jobs/delivery、配置类及对应测试的 KDoc 章节引用全部对齐现行 flight-state.md/design.md；FlightStateEngineTest 重写为合并语义（62/62 通过）。 V1__flight_state_baseline.sql 保留原样（内容注释仍带旧章节号，改动会破坏 已应用迁移的 Flyway checksum，待重建基线或 V2 净迁移时收敛）。
This commit is contained in:
windyboy
+1
-1
@@ -110,7 +110,7 @@ SNDR | TYPE | STYP | SEQN
|
||||
`SCHD-DNLD` 与 `SCHD-RESP` 共用 `ScheduleProcessor.applyScheduleRecords`：
|
||||
|
||||
1. **重放判定**：`PROC_STATE` 已存在成功终态 → 幂等成功，仅追加留痕，不重复写入。
|
||||
2. **整包校验**：声明记录数、记录范围与运营日归属等校验失败 → 整包 `DEAD(PROTOCOL)`，不写半包，既有状态保持不变。
|
||||
2. **整包校验**：声明记录数、航班标识与运营日推导等校验失败 → 整包 `DEAD(PROTOCOL)`，不写半包，既有状态保持不变。
|
||||
3. **事务写入**：锁内按 `FLID` 点查归属日，发现同一航班跨运营日即整包回滚并 `DEAD(PROTOCOL)`；通过后合并写主表与资源明细。报文未携带的航班不因本次日计划报文被删除。
|
||||
4. **提交结果**：同一事务保存 `KAFKA:schd` / `KAFKA:msg` 事件、置消息 `SUCCEEDED` 并预登记回填待办；事务提交后执行信箱回填，留痕在事务外追加。
|
||||
|
||||
|
||||
@@ -6,7 +6,6 @@ import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlProperty
|
||||
import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlRootElement
|
||||
import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlText
|
||||
import com.gzzn.omms.msgexchange.domain.flight.ScheduleRecord
|
||||
import java.time.LocalDate
|
||||
|
||||
/** Stable payload boundary used by processing. It is deliberately not an XML model. */
|
||||
data class FlopPayload(
|
||||
@@ -18,8 +17,6 @@ data class FlopPayload(
|
||||
data class ScheduleBody(
|
||||
val recsDeclared: Int,
|
||||
val records: List<ScheduleRecord>,
|
||||
val scopeStart: LocalDate? = null,
|
||||
val scopeEnd: LocalDate? = null,
|
||||
)
|
||||
|
||||
/**
|
||||
@@ -54,9 +51,7 @@ data class SchdXml(
|
||||
@JsonIgnoreProperties(ignoreUnknown = true)
|
||||
data class FlightRecordXml(
|
||||
@param:JacksonXmlProperty(localName = "FLID") val flid: String? = null,
|
||||
@param:JacksonXmlProperty(localName = "FFID") val ffid: String? = null,
|
||||
@param:JacksonXmlProperty(localName = "FDEL") val fdel: FdelXml? = null,
|
||||
@param:JacksonXmlProperty(localName = "SEQN") val seqn: Long? = null,
|
||||
@param:JacksonXmlProperty(localName = "ALCD") val alcd: String? = null,
|
||||
@param:JacksonXmlProperty(localName = "ALSC") val alsc: String? = null,
|
||||
@param:JacksonXmlProperty(localName = "FLNO") val flno: String? = null,
|
||||
|
||||
@@ -16,12 +16,12 @@ internal object SisWireMapper {
|
||||
|
||||
private fun scheduleRecord(xml: FlightRecordXml): ScheduleRecord? =
|
||||
xml.flid?.trim()?.takeIf(String::isNotEmpty)?.let {
|
||||
ScheduleRecord(it, xml.scalars(), xml.collections(), xml.seqn ?: 0L)
|
||||
ScheduleRecord(it, xml.scalars(), xml.collections())
|
||||
}
|
||||
|
||||
private fun FlightRecordXml.scalars(): Map<String, String> = linkedMapOf<String, String>().apply {
|
||||
listOf(
|
||||
"FFID" to ffid, "ALCD" to alcd, "ALSC" to alsc, "FLNO" to flno, "MVIN" to mvin, "SODT" to sodt,
|
||||
"ALCD" to alcd, "ALSC" to alsc, "FLNO" to flno, "MVIN" to mvin, "SODT" to sodt,
|
||||
"FLTY" to flty, "FLIN" to flin, "ACFT" to acft, "RENO" to reno, "TAOP" to taop,
|
||||
"TAFL" to tafl, "TAID" to taid, "TRML" to trml, "MAXP" to maxp, "CSOP" to csop,
|
||||
"CSFT" to csft, "MAID" to maid, "ESTT" to estt, "ACTT" to actt, "STND" to stnd,
|
||||
|
||||
@@ -2,13 +2,13 @@ package com.gzzn.omms.msgexchange.config
|
||||
|
||||
import io.micronaut.context.annotation.ConfigurationProperties
|
||||
|
||||
/** 生命周期与保留期（docs/flight-state.md §8）。窗口按机场时区计算。 */
|
||||
/** 生命周期与保留期（docs/flight-state.md §6；design.md §6.2）。窗口按机场时区计算。 */
|
||||
@ConfigurationProperties("msgx.history")
|
||||
class HistoryProps {
|
||||
/** 已取消（CNCL 非空）超过 N 小时。 */
|
||||
var cancelledHours: Long = 48
|
||||
|
||||
/** 到港/离港终态（NAAT/NEAT，含义待术语表确认 §10）超过 N 小时。 */
|
||||
/** 到港/离港终态（NAAT/NEAT，含义待术语表确认 §6 开放项）超过 N 小时。 */
|
||||
var terminalHours: Long = 48
|
||||
|
||||
/** STATE = DELETED 超过 N 小时。 */
|
||||
@@ -17,9 +17,9 @@ class HistoryProps {
|
||||
/** 无终态字段：最后有效更新超过兜底期限（默认 7 天）。 */
|
||||
var idleHours: Long = 24 * 7
|
||||
|
||||
/** 留痕 SCHD_SNAP_LOG 保留天数（§8.3，按 (SCOPE_END, RECV_AT) 清理）。 */
|
||||
/** 留痕 SCHD_SNAP_LOG 保留天数（design.md §6.2，按 (SCOPE_END, RECV_AT) 清理）。 */
|
||||
var snapLogRetentionDays: Long = 90
|
||||
|
||||
/** 是否接通历史存储——未接通时 HISTORY_SWEEP 必须删 0 条（§8.2）。 */
|
||||
/** 是否接通历史存储——未接通时清理必须删 0 条（flight-state.md §6 红线）。 */
|
||||
var historyStoreEnabled: Boolean = false
|
||||
}
|
||||
|
||||
@@ -3,8 +3,9 @@ package com.gzzn.omms.msgexchange.config
|
||||
import io.micronaut.context.annotation.ConfigurationProperties
|
||||
|
||||
/**
|
||||
* 运营日计算（docs/flight-state.md §3.5）：由 SODT（ddMMMyyHHmm）与机场时区计算；
|
||||
* 切日边界业务配置——不得假设等于接收日期或自然日零点（默认 0 点为占位，待业务确认 §10）。
|
||||
* 运营日计算（docs/flight-state.md §2.1）：由 SODT（ddMMMyyHHmm）与机场时区按切日边界推导；
|
||||
* OPERATION_DAY 不是接收日/落库日，一经确定不可变。切日边界业务配置，
|
||||
* 不得假设等于自然日零点（默认 0 点为占位，业务口径待确认）。
|
||||
*/
|
||||
@ConfigurationProperties("msgx.operation-day")
|
||||
class OperationDayProps {
|
||||
|
||||
@@ -23,7 +23,7 @@ interface DeliveryPort {
|
||||
*/
|
||||
fun sendKafkaSchd(topic: String, key: String, payloadJson: String)
|
||||
|
||||
/** TOMBSTONE：key=FLID、value=null——整态键缺失表示删除旧值（§7.3）。 */
|
||||
/** TOMBSTONE：key=FLID、value=null——整态键缺失表示删除旧值（flight-state.md §5）。 */
|
||||
fun sendKafkaNull(topic: String, key: String)
|
||||
|
||||
/** 连通性探测（健康检查用）；默认 true，真实 Kafka 实装时覆写为 producer metadata 校验。 */
|
||||
@@ -31,9 +31,9 @@ interface DeliveryPort {
|
||||
}
|
||||
|
||||
/**
|
||||
* 投递调度（docs/flight-state.md §7.3）：逐条 KAFKA_MSG 严格 FIFO；
|
||||
* 投递调度（docs/flight-state.md §5 + design.md §5.1/§5.2）：逐条 KAFKA_MSG 严格 FIFO；
|
||||
* KAFKA_SCHD 走 flushSchd 批量——同一 FLID 未发事件按最新 STATE_VERSION 合并输出，
|
||||
* TOMBSTONE 发 null 值消息。两主题间不保证顺序（§7.3）。
|
||||
* TOMBSTONE 发 null 值消息。两主题间不保证顺序（§5）。
|
||||
* 批量闭环：队首退避未到期不 claim；发送失败整批 attempts+1 退避，达上限整批 DEAD/DLQ。
|
||||
*/
|
||||
@Singleton
|
||||
@@ -91,7 +91,7 @@ class Dispatcher(
|
||||
}
|
||||
}
|
||||
|
||||
/** flushSchd：同 FLID 未发事件按最新 STATE_VERSION 合并（§7.3）；TOMBSTONE 发 null。 */
|
||||
/** flushSchd：同 FLID 未发事件按最新 STATE_VERSION 合并（§5）；TOMBSTONE 发 null。 */
|
||||
internal fun flushSchd() {
|
||||
val batch = try {
|
||||
msgEvents.mergePendingSchd(props.schd.flushLimit)
|
||||
@@ -116,7 +116,7 @@ class Dispatcher(
|
||||
}
|
||||
val sentIds = batch.mapNotNull { it.eventId }.toSet() - failures.mapNotNull { it.eventId }.toSet()
|
||||
if (sentIds.isNotEmpty()) msgEvents.markAllSent(sentIds.toList())
|
||||
// 被最新版本合并压掉的未发事件同样关闭（§7.3：同 FLID 只按最新 STATE_VERSION 输出一次）
|
||||
// 被最新版本合并压掉的未发事件同样关闭（§5：同 FLID 只按最新 STATE_VERSION 输出一次）
|
||||
val sentVersions = batch.associate { it.partitionKey to it.stateVersion }
|
||||
runCatching {
|
||||
while (true) {
|
||||
|
||||
@@ -1,31 +0,0 @@
|
||||
package com.gzzn.omms.msgexchange.domain
|
||||
|
||||
import com.gzzn.omms.msgexchange.domain.flight.MergeChange
|
||||
import com.gzzn.omms.msgexchange.domain.flight.ScheduleRecord
|
||||
import java.time.LocalDate
|
||||
|
||||
/**
|
||||
* 处理决策（docs/flight-state.md）——解码与校验进，落库计划出，纯数据不触碰 DB/Kafka。
|
||||
* 事务边界、PIPELINE_LOCK、事件登记由处理层执行（§5.1/§6）。
|
||||
*/
|
||||
sealed interface Decision {
|
||||
/**
|
||||
* SCHD DNLD/RESP：已过 §5.2 报文完整性校验的记录集。
|
||||
* §5.3 归属校验与 §5.4 upsert 在事务内执行（需读既有 OPERATION_DAY）。
|
||||
*/
|
||||
data class Schedule(
|
||||
val records: List<ScheduleRecord>,
|
||||
/** 报文覆盖运营日范围（单日快照时两者相等；无法确定时为 null → 整包拒绝）。 */
|
||||
val scopeStart: LocalDate?,
|
||||
val scopeEnd: LocalDate?,
|
||||
) : Decision
|
||||
|
||||
/** FLOP 增量合并（§6.1）/ ADFT（§2.1 语义待确认，按 MergeChange 承载）。 */
|
||||
data class Dynamic(val change: MergeChange) : Decision
|
||||
|
||||
/** FDEL 标记删除（§6.2）。 */
|
||||
data class Delete(val flid: String) : Decision
|
||||
|
||||
/** 幂等无变化（重复/迟到等，记成功但不推进任何状态）。 */
|
||||
data object NoOp : Decision
|
||||
}
|
||||
@@ -1,7 +1,8 @@
|
||||
package com.gzzn.omms.msgexchange.domain
|
||||
|
||||
/**
|
||||
* 解码后的入站报文（统一内部模型）。META 字段实名 SNDR/SEQN/DTTM（legacy META.java，I3 幂等键来源）。
|
||||
* 解码后的入站报文（docs/design.md §2.2 统一载荷）。META 字段实名
|
||||
* SNDR/SEQN/DTTM，其中 SNDR|TYPE|STYP|SEQN 是业务幂等键来源（Identity.of）。
|
||||
*/
|
||||
data class MetaFields(
|
||||
val sndr: String,
|
||||
@@ -11,13 +12,13 @@ data class MetaFields(
|
||||
val dttm: Long,
|
||||
)
|
||||
|
||||
/** 消息分派（sealed + 穷尽 when；FDEL 一等公民——终止航班实例 §6.2）。 */
|
||||
/** 消息分派（design.md §2.2：一等分派键，穷尽 when；FDEL 一等公民——终止航班实例）。 */
|
||||
sealed interface MsgKind {
|
||||
data class Schd(val subtype: SchdSubtype) : MsgKind
|
||||
data class Flop(val subtype: String) : MsgKind // 运行动态 STYP（FDEL 除外）
|
||||
data object Fdel : MsgKind
|
||||
|
||||
/** 未支持类型（§9：FAILED(UNSUPPORTED)，达阈值转 DEAD）。 */
|
||||
/** 未支持类型（design.md §2.3：FAILED(UNSUPPORTED)，达阈值转 DEAD）。 */
|
||||
data class Unsupported(val tag: String) : MsgKind
|
||||
|
||||
enum class SchdSubtype { RESP, DNLD, ADFT }
|
||||
|
||||
@@ -1,22 +1,23 @@
|
||||
package com.gzzn.omms.msgexchange.domain
|
||||
|
||||
/** MSG_EVENT 事件形态（§7.3）。 */
|
||||
/** MSG_EVENT 事件形态（design.md §2.1：UPSERT 整态/通知，TOMBSTONE 删除）。 */
|
||||
enum class EventType { UPSERT, TOMBSTONE }
|
||||
|
||||
/** MSG_EVENT 投递状态（outbox 状态机）。 */
|
||||
/** MSG_EVENT 投递状态机（design.md §2.3：PENDING → SENT；失败退避重试；耗尽转 DEAD 保留作 DLQ）。 */
|
||||
enum class EventStatus { PENDING, SENT, DEAD }
|
||||
|
||||
/**
|
||||
* MSG_EVENT（outbox，§3.2）：状态、变更、删除通知。
|
||||
* KAFKA_SCHD 整态 + KAFKA_MSG 变化通知；TOMBSTONE 仅在 ACTIVE→DELETED 时
|
||||
* 与删除同事务登记（§7.3），投递失败持续重试。
|
||||
* MSG_EVENT（outbox，design.md §2.1）：状态、变更、删除通知。
|
||||
* KAFKA_SCHD 发整态、KAFKA_MSG 只通知变化（docs/flight-state.md §5，两主题不承诺顺序）；
|
||||
* TOMBSTONE 仅在 ACTIVE→DELETED（§3.3）或生命周期清理前补发（§6），与删除同事务登记。
|
||||
* stateVersion：发布时的航班版本；KAFKA_SCHD 聚合按 FLID 取最新（§5）。
|
||||
*/
|
||||
data class MsgEvent(
|
||||
val eventId: Long? = null,
|
||||
val target: String,
|
||||
val partitionKey: String, // 恒为 FLID
|
||||
val eventType: EventType = EventType.UPSERT,
|
||||
val stateVersion: Long = 0, // 发布时航班版本；Dispatcher 合并同 FLID 未发事件取最新（§7.3）
|
||||
val stateVersion: Long = 0,
|
||||
val payloadJson: String,
|
||||
val state: EventStatus = EventStatus.PENDING,
|
||||
val attempts: Int = 0,
|
||||
|
||||
@@ -5,15 +5,16 @@ import java.time.LocalDateTime
|
||||
import java.time.ZoneId
|
||||
|
||||
/**
|
||||
* 运营日计算（docs/flight-state.md §3.5）：由计划运行时间字段 SODT（ddMMMyyHHmm）
|
||||
* 与机场时区计算；切日边界业务配置——默认 0 点为占位，待业务确认（§10）。
|
||||
* 运营日计算（docs/flight-state.md §2.1）：由计划运行时间字段 SODT（ddMMMyyHHmm）
|
||||
* 与机场时区按切日边界推导；OPERATION_DAY 不是消息接收日或落库日，一经确定不可变。
|
||||
* 切日边界由 msgx.operation-day.cutoff-hour 配置（默认 0 = 自然日零点，业务口径待确认）。
|
||||
*/
|
||||
class OperationDayCalculator(
|
||||
zone: ZoneId,
|
||||
cutoffHour: Int,
|
||||
) {
|
||||
private val zone: ZoneId = zone
|
||||
/** 切日边界：SODT 本地时刻早于该小时的归属前一运营日（0–23，越界按 0）。 */
|
||||
/** 切日边界：SODT 本地时刻早于该小时的归属前一运营日（0–23，越界收敛到边界值）。 */
|
||||
private val cutoffHour: Int = cutoffHour.coerceIn(0, 23)
|
||||
|
||||
/** SODT → 运营日；输入 null/空/非法返回 null（不抛异常，调用方按"运营日不可计算"处理）。 */
|
||||
@@ -28,7 +29,7 @@ class OperationDayCalculator(
|
||||
/**
|
||||
* SIS SODT 线格式：ddMMMyyHHmm（如 15DEC031723），月份英文三字母、大小写不敏感。
|
||||
* 两位年显式按 2000 基准展开（java.time 的 yy reduced-value 解析跨实现不一致，
|
||||
* 显式展开保证 AODB 侧年份窗口唯一口径；基准年待真实报文验收确认 §10）。
|
||||
* 显式展开保证 AODB 侧年份窗口唯一口径；基准年待真实报文验收确认）。
|
||||
*/
|
||||
private val SODT_REGEX = Regex("(\\d{1,2})([A-Za-z]{3})(\\d{2})(\\d{2})(\\d{2})")
|
||||
|
||||
|
||||
@@ -1,27 +1,23 @@
|
||||
package com.gzzn.omms.msgexchange.domain
|
||||
|
||||
/**
|
||||
* ACMA-8 数据模型 / PROC_STATE 状态机（docs/flight-state.md §3.2：每消息一行，
|
||||
* 处理状态与重试结果；兼作快照重放判定 §5.1）。
|
||||
* PROC_STATE 处理伴生状态（docs/design.md §2.1/§2.3）：每消息一行，
|
||||
* MSG_ID = 信箱 ID 主键防重复入队；IDENTITY_KEY 唯一约束防业务重复；
|
||||
* 处理状态机与错误分类同设计文档 §2.3。SUCCEEDED 终态兼作 SCHD 快照
|
||||
* 重放判定（docs/flight-state.md §4 步骤 / design.md §4.1）。
|
||||
*/
|
||||
enum class ProcStatus { PENDING, FAILED, SUCCEEDED, SKIPPED, DEAD }
|
||||
|
||||
/** §9 错误分类：PROTOCOL = 整包拒绝（归属日不符等），不重试交人工。 */
|
||||
/** 错误分类（design.md §2.3）：MALFORMED/PROTOCOL 直接 DEAD 不重试；其余退避重试。 */
|
||||
enum class ErrorClass { MALFORMED, PROTOCOL, CODEC_ERROR, EXHAUSTED, INFRA, UNSUPPORTED }
|
||||
|
||||
data class ProcState(
|
||||
val msgId: Long,
|
||||
val state: ProcStatus,
|
||||
val identityKey: String? = null, // SNDR|TYPE|STYP|SEQN；decode 后首次绑定，FAILED 重试不重绑
|
||||
val identityKey: String? = null, // SNDR|TYPE|STYP|SEQN（design.md §2.2）；decode 后首次绑定，FAILED 重试不重绑
|
||||
val attempts: Int = 0,
|
||||
val nextAttemptAt: java.time.Instant? = null,
|
||||
val errorClass: ErrorClass? = null,
|
||||
val lastError: String? = null,
|
||||
val updatedAt: java.time.Instant = java.time.Instant.now(),
|
||||
) {
|
||||
val isTerminal: Boolean
|
||||
get() = state == ProcStatus.SUCCEEDED || state == ProcStatus.SKIPPED || state == ProcStatus.DEAD
|
||||
|
||||
/** 终态皆可归档；PENDING/FAILED 不迁。 */
|
||||
val archivable: Boolean get() = isTerminal
|
||||
}
|
||||
)
|
||||
|
||||
@@ -4,13 +4,18 @@ import java.time.Instant
|
||||
import java.time.LocalDate
|
||||
|
||||
/**
|
||||
* SCHD 快照留痕模型（docs/flight-state.md §5.5）：
|
||||
* RESULT 与 FLAGS 分列（可「成功且告警」）；一行 = 一次尝试，重放也记；
|
||||
* 留痕不参与决策；写失败只记指标；保留 90 天，按 (SCOPE_END, RECV_AT) 清理。
|
||||
* SCHD 快照留痕 SCHD_SNAP_LOG（docs/flight-state.md §2 权威模型表；
|
||||
* 清理规则见 design.md §6.2）：只追加、可重建、不参与状态决策，写失败只记指标；
|
||||
* 一行 = 一次尝试，重放也记；保留 90 天，按 (SCOPE_END, RECV_AT) 清理。
|
||||
*/
|
||||
enum class SnapshotResult { COMMITTED, REPLAY_SKIPPED, ROLLED_BACK }
|
||||
|
||||
enum class SnapshotFlag { EMPTY, RECS_DROP, SEQN_REGRESSION, DAY_MISMATCH, SCHD_REVIVE_CONFLICT }
|
||||
/**
|
||||
* 留痕告警 flags：EMPTY（空快照合法）、RECS_DROP（声明数量不符）、
|
||||
* DAY_MISMATCH（记录运营日不可计算，整包拒绝）、
|
||||
* SCHD_REVIVE_CONFLICT（日计划命中 DELETED 航班，保持 DELETED 不恢复）。
|
||||
*/
|
||||
enum class SnapshotFlag { EMPTY, RECS_DROP, DAY_MISMATCH, SCHD_REVIVE_CONFLICT }
|
||||
|
||||
data class SnapshotLogEntry(
|
||||
val msgId: Long,
|
||||
@@ -22,5 +27,5 @@ data class SnapshotLogEntry(
|
||||
val durationMs: Long,
|
||||
val result: SnapshotResult,
|
||||
val flags: Set<SnapshotFlag> = emptySet(),
|
||||
val archiveKey: String? = null, // 证据层引用（尚未交付 §3.2）
|
||||
val archiveKey: String? = null, // 证据层引用（尚未交付 §2）
|
||||
)
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
package com.gzzn.omms.msgexchange.domain
|
||||
|
||||
/**
|
||||
* MSG_EVENT 投递目标（§7.3：KAFKA_SCHD 发整态、KAFKA_MSG 只通知变化；
|
||||
* 两主题间不保证顺序）。ES 投影属阶段 B，暂不登记目标。
|
||||
* MSG_EVENT 投递目标（docs/flight-state.md §5：KAFKA_SCHD 按 FLID 发完整状态投影、
|
||||
* KAFKA_MSG 只通知变化；两主题间不保证顺序；删除用 tombstone）。
|
||||
* 常量值即 MSG_EVENT.TARGET 落库值（design.md §2.1）；topic 名由投递适配层映射
|
||||
* （KAFKA:msg → "msg"，KAFKA:schd → "schd"）。ES 投影属阶段 B，暂不登记目标。
|
||||
*/
|
||||
object Targets {
|
||||
const val KAFKA_MSG = "KAFKA:msg"
|
||||
|
||||
@@ -4,14 +4,15 @@ import java.time.Instant
|
||||
import java.time.LocalDate
|
||||
|
||||
/**
|
||||
* 航班实例当前态模型（docs/flight-state.md §3）。
|
||||
* 航班实例当前态模型（docs/flight-state.md §2/§3）。
|
||||
*
|
||||
* 身份：FLID 唯一关联键；OPERATION_DAY 一经确定不可变（§3.5/§5.3）；
|
||||
* STATE 仅 ACTIVE/DELETED（§3.1，无 ARCHIVED——物理清除只发生在历史归档成功之后 §8.2）。
|
||||
* 身份：FLID 是唯一关联键（§2.1）；OPERATION_DAY 一经确定不可变（§2.1），
|
||||
* 未被日计划收录前可为 NULL。STATE 仅 ACTIVE/DELETED（§3.3，无 ARCHIVED——
|
||||
* 物理清除只发生在历史归档成功之后 §6）。
|
||||
*/
|
||||
enum class FlightState { ACTIVE, DELETED }
|
||||
|
||||
/** FLIGHT_SCHD 主行的身份与追踪字段（不含标量载荷）。 */
|
||||
/** FLIGHT_SCHD 主行的身份与追踪字段（§2 权威模型；不含标量载荷）。 */
|
||||
data class FlightMainRow(
|
||||
val flid: String,
|
||||
val operationDay: LocalDate?,
|
||||
@@ -22,35 +23,19 @@ data class FlightMainRow(
|
||||
)
|
||||
|
||||
/**
|
||||
* SCHD DNLD/RESP 单条 FLTR 记录（解码产物，§5 入口）。
|
||||
* scalars/collections 只含报文中出现的字段——完整快照语义下
|
||||
* 出现 = Set/Replace，未出现 = 清除/替换空集（§5.4）。
|
||||
* SCHD DNLD/RESP 单条 FLTR 记录（解码产物，§3.1 日计划入口）。
|
||||
* scalars/collections 只含报文中出现的字段：出现 = Set/Replace（标量空值 = 显式清空），
|
||||
* 未出现 = 按合并语义保留本地值（§3.1/§7 不变量）；集合按完整合并结果写入（§2.2）。
|
||||
*/
|
||||
data class ScheduleRecord(
|
||||
val flid: String,
|
||||
val scalars: Map<String, String>,
|
||||
val collections: Map<String, List<Map<String, String>>> = emptyMap(),
|
||||
val seqn: Long = 0,
|
||||
)
|
||||
|
||||
/**
|
||||
* 完整快照落库载荷（§5.1 步骤 6）：按记录整体替换映射内字段，
|
||||
* 仓储层负责明细先删后插与 `OPERATION_DAY` 不可变条件更新（§7.4）。
|
||||
*/
|
||||
data class SnapshotPatch(
|
||||
val flid: String,
|
||||
val operationDay: LocalDate,
|
||||
val scalars: Map<String, String>,
|
||||
val collections: Map<String, List<Map<String, String>>>,
|
||||
)
|
||||
|
||||
/** 快照 upsert 结果（§5.1 步骤 6：DELETED 航班保持 DELETED 并告警，不恢复）。 */
|
||||
enum class UpsertOutcome { INSERTED, UPDATED, REVIVE_CONFLICT }
|
||||
|
||||
/**
|
||||
* FLOP/ADFT 增量载荷（§6.1/§2.1）。
|
||||
* scalars 出现 = Set，缺失 = 保留本地值；collections 出现 = Replace
|
||||
* （FLOP 集合语义未定案前沿用 Replace，§10 当前偏差明示）。
|
||||
* FLOP/ADFT 增量载荷（§3.2/§3.3）：出现字段/集合更新，缺失保留本地值；
|
||||
* ADFT 缺失字段语义待上游确认前按保守 Set-only 处理（§3.3），不沿用全量替换。
|
||||
*/
|
||||
data class MergeChange(
|
||||
val flid: String,
|
||||
@@ -59,8 +44,9 @@ data class MergeChange(
|
||||
)
|
||||
|
||||
/**
|
||||
* 完整当前态（§3.3）：主行 + 全部明细 = 完整当前态；
|
||||
* 读取须在一致性读事务中，且展示层过滤 STATE = ACTIVE（§9）。
|
||||
* 完整当前态（§2.2/§3）：主行 + 全部明细 = 完整当前态。
|
||||
* 写入前在内存生成完整新状态再落库（引擎产物），读取须在主表与全部明细的
|
||||
* 一致性读边界内进行（§5），展示层过滤 STATE = ACTIVE。
|
||||
*/
|
||||
data class FlightSnapshot(
|
||||
val flid: String,
|
||||
@@ -71,19 +57,23 @@ data class FlightSnapshot(
|
||||
val collections: Map<String, List<Map<String, String>>>,
|
||||
)
|
||||
|
||||
/** §8.1 历史判定窗口（按机场时区计算；窗口值业务配置）。 */
|
||||
/** §6 生命周期判定窗口（按机场时区计算；窗口值业务配置）。 */
|
||||
data class HistoryRules(
|
||||
val cancelledHours: Long = 48, // CNCL 非空超过 N 小时
|
||||
val terminalHours: Long = 48, // NAAT/NEAT 终态超过 N 小时（字段含义待术语表确认 §10）
|
||||
val terminalHours: Long = 48, // NAAT/NEAT 终态超过 N 小时（字段含义待术语表确认 §6）
|
||||
val deletedHours: Long = 48, // STATE = DELETED 超过 N 小时
|
||||
val idleHours: Long = 24 * 7, // 无终态字段：最后更新超过兜底期限
|
||||
)
|
||||
|
||||
/** §8.1 命中历史判定的航班（归档 → 物理清除候选）。 */
|
||||
/**
|
||||
* §6 命中生命周期判定的航班（先归档 → 后物理清除；历史存储失败必须删 0 行）。
|
||||
* wasNeverFdel：是否未经 FDEL 而被生命周期清除——清除前须补发一次删除事件（§3.3/§5）。
|
||||
* 当前为推断口径（无持久化"曾 FDEL"痕迹）：state=ACTIVE 视为未经 FDEL，
|
||||
* FDEL→ADFT 重激活后的航班会被误判并重复补发，收敛见 §6 开放项。
|
||||
*/
|
||||
data class HistoryCandidate(
|
||||
val flid: String,
|
||||
val state: FlightState,
|
||||
val stateVersion: Long,
|
||||
/** 是否未经 FDEL 而被生命周期清除——清除前须补发一次删除事件（§7.3/§8.2）。 */
|
||||
val wasNeverFdel: Boolean,
|
||||
)
|
||||
|
||||
@@ -5,33 +5,40 @@ import com.gzzn.omms.msgexchange.domain.SnapshotFlag
|
||||
import java.time.LocalDate
|
||||
|
||||
/**
|
||||
* 航班状态引擎（docs/flight-state.md §3.3/§5.2/§5.4/§6.1）——纯函数，
|
||||
* 内存生成完整新状态再落库；不触碰 DB/Kafka。
|
||||
* 航班状态引擎（docs/flight-state.md §2–§4）——纯函数，内存推导完整新状态，
|
||||
* 不触碰 DB/Kafka：
|
||||
* - [validateMessage]：日计划整包校验（§4 步骤 2：声明数量、航班标识、运营日推导）；
|
||||
* - [snapshotState]：DNLD/RESP 日计划合并（§3.1）——出现 Set/Replace、缺失保留、
|
||||
* 标量显式空值清空；DELETED 航班保持 DELETED（§3.3，不承担恢复语义）；
|
||||
* - [mergedState]：FLOP/ADFT 增量合并（§3.2/§3.3）——只修改报文表达的字段/集合。
|
||||
*/
|
||||
object FlightStateEngine {
|
||||
|
||||
/** 集合键白名单：10 类集合 ↔ 9 张明细表（ROUT/ERUT 共用 FLIGHT_ROUTE_POINT，§3.2）。 */
|
||||
/** 集合键白名单：10 类集合 ↔ 9 张明细表（8 张资源表 + FLIGHT_ROUTE_POINT，
|
||||
* ROUT/ERUT 共用路线表，flight-state.md §2.2）。 */
|
||||
val COLLECTION_KEYS: Set<String> = setOf(
|
||||
"GTDT", "CKDT", "CLDT", "PSDT", "CHDT", "DELY", "ABTM", "CHOT", "ROUT", "ERUT",
|
||||
)
|
||||
|
||||
/**
|
||||
* §5.2 报文完整性五项校验：RECS 0–9999 且等于实收数；每条含合法数字型 FLID；
|
||||
* 快照内不重复；每条记录运营日可计算且在报文覆盖范围内。任一失败整包不落地。
|
||||
* 日计划整包校验（§4 步骤 2 / §3.1）：RECS 0–9999 且等于实收数；每条含合法
|
||||
* 数字型 FLID（SIS Number(1-12)）；快照内不重复；每条记录运营日可计算。
|
||||
* 任一失败整包不落地（DEAD(PROTOCOL)）。
|
||||
*/
|
||||
fun validateMessage(
|
||||
recsDeclared: Int,
|
||||
records: List<ScheduleRecord>,
|
||||
scopeStart: LocalDate?,
|
||||
scopeEnd: LocalDate?,
|
||||
opDay: OperationDayCalculator,
|
||||
): SnapshotValidation {
|
||||
val flags = linkedSetOf<SnapshotFlag>()
|
||||
if (recsDeclared !in 0..9999) return SnapshotValidation.Invalid("RECS out of range: $recsDeclared", setOf(SnapshotFlag.RECS_DROP))
|
||||
if (records.size != recsDeclared) return SnapshotValidation.Invalid(
|
||||
"RECS ($recsDeclared) != received FLTR count (${records.size})",
|
||||
setOf(SnapshotFlag.RECS_DROP),
|
||||
)
|
||||
if (recsDeclared !in 0..9999) {
|
||||
return SnapshotValidation.Invalid("RECS out of range: $recsDeclared", setOf(SnapshotFlag.RECS_DROP))
|
||||
}
|
||||
if (records.size != recsDeclared) {
|
||||
return SnapshotValidation.Invalid(
|
||||
"RECS ($recsDeclared) != received FLTR count (${records.size})",
|
||||
setOf(SnapshotFlag.RECS_DROP),
|
||||
)
|
||||
}
|
||||
if (records.isEmpty()) return SnapshotValidation.Ok(emptyMap(), setOf(SnapshotFlag.EMPTY))
|
||||
|
||||
val perRecordDay = linkedMapOf<String, LocalDate>()
|
||||
@@ -45,24 +52,22 @@ object FlightStateEngine {
|
||||
}
|
||||
val day = opDay.compute(record.scalars["SODT"])
|
||||
if (day == null) {
|
||||
return SnapshotValidation.Invalid("operation day not computable, flid=${record.flid}", setOf(SnapshotFlag.DAY_MISMATCH))
|
||||
}
|
||||
val inScope = (scopeStart == null || !day.isBefore(scopeStart)) &&
|
||||
(scopeEnd == null || !day.isAfter(scopeEnd))
|
||||
if (!inScope) {
|
||||
return SnapshotValidation.Invalid(
|
||||
"operation day $day outside coverage [$scopeStart, $scopeEnd], flid=${record.flid}",
|
||||
"operation day not computable, flid=${record.flid}",
|
||||
setOf(SnapshotFlag.DAY_MISMATCH),
|
||||
)
|
||||
}
|
||||
perRecordDay[record.flid] = day
|
||||
}
|
||||
return SnapshotValidation.Ok(perRecordDay, flags)
|
||||
return SnapshotValidation.Ok(perRecordDay, emptySet())
|
||||
}
|
||||
|
||||
/**
|
||||
* 完整快照语义（§5.4 DNLD/RESP）：标量出现 Set、缺失 Clear；集合出现 Replace、缺失 Replace 空集。
|
||||
* `keepDeleted = true` 时 STATE 保持 DELETED（§5.1 步骤 6：普通 SCHD 不恢复）。
|
||||
* 日计划合并（§3.1）：标量出现 Set（空值 = 显式清空）、缺失保留本地值；
|
||||
* 集合出现 Replace、缺失保留。新航班（current == null）无本地值可保留，
|
||||
* 集合按全键输出空集（§2.2 完整合并结果写入口径）。
|
||||
* `keepDeleted = true` 时 STATE 保持 DELETED（§3.3：日计划不承担恢复，
|
||||
* 恢复入口只有 ADFT）；冲突告警由调用方按 SCHD_REVIVE_CONFLICT 记录。
|
||||
*/
|
||||
fun snapshotState(
|
||||
current: FlightSnapshot?,
|
||||
@@ -75,28 +80,33 @@ object FlightStateEngine {
|
||||
keepDeleted -> FlightState.DELETED
|
||||
else -> current.state
|
||||
}
|
||||
// §5.4 完整快照：标量出现 Set、缺失 Clear；集合出现 Replace、缺失 Replace 空集。
|
||||
// 完整替换 = 新状态只由记录决定，不从 current 继承任何字段。
|
||||
val scalars: Map<String, String> = record.scalars
|
||||
val scalars = buildMap {
|
||||
current?.scalars?.let(::putAll) // 缺失 = 保留（§3.1）
|
||||
putAll(record.scalars) // 出现 = Set；空串 = 显式清空
|
||||
}
|
||||
val collections = buildMap {
|
||||
COLLECTION_KEYS.forEach { key -> put(key, emptyList()) } // 缺失 = Replace 空集（§5.4）
|
||||
if (current == null) {
|
||||
COLLECTION_KEYS.forEach { key -> put(key, emptyList()) } // 新航班全键输出
|
||||
} else {
|
||||
putAll(current.collections) // 缺失 = 保留（§3.1）
|
||||
}
|
||||
record.collections.forEach { (key, items) ->
|
||||
if (key in COLLECTION_KEYS) put(key, items)
|
||||
if (key in COLLECTION_KEYS) put(key, items) // 出现 = Replace（§3.1）
|
||||
}
|
||||
}
|
||||
return FlightSnapshot(
|
||||
flid = record.flid,
|
||||
operationDay = operationDay,
|
||||
state = state,
|
||||
stateVersion = (current?.stateVersion ?: 0L) + 1,
|
||||
stateVersion = (current?.stateVersion ?: 0L) + 1, // 每次成功写入 +1（§3.1/§4）
|
||||
scalars = scalars,
|
||||
collections = collections,
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* FLOP 增量合并（§6.1）：标量出现覆盖、缺失保留；集合出现 Replace、缺失保留
|
||||
* ——FLOP 集合语义未定案，Replace 为当前偏差明示沿用（§10）。
|
||||
* FLOP/ADFT 增量合并（§3.2/§3.3）：标量出现覆盖、缺失保留；集合出现 Replace、
|
||||
* 缺失保留——只修改报文表达的字段/集合，其余航班状态保持不变；不推导运营日。
|
||||
*/
|
||||
fun mergedState(current: FlightSnapshot, change: MergeChange): FlightSnapshot {
|
||||
val scalars = buildMap {
|
||||
@@ -116,11 +126,11 @@ object FlightStateEngine {
|
||||
)
|
||||
}
|
||||
|
||||
/** FLID：数字型，SIS §3.16.2 Number(1-12)。 */
|
||||
/** FLID：数字型（SIS Number(1-12)）。 */
|
||||
private val FLID_REGEX = Regex("\\d{1,12}")
|
||||
}
|
||||
|
||||
/** §5.2 校验结果：Ok 携带每条记录的归属运营日与观测 flags；Invalid 整包拒绝。 */
|
||||
/** 整包校验结果：Ok 携带每条记录的归属运营日与观测 flags；Invalid 整包拒绝。 */
|
||||
sealed interface SnapshotValidation {
|
||||
data class Ok(
|
||||
val perRecordDay: Map<String, LocalDate>,
|
||||
|
||||
@@ -14,22 +14,22 @@ import java.time.LocalDate
|
||||
import java.time.ZoneId
|
||||
|
||||
// =====================================================================
|
||||
// 仓储契约（docs/flight-state.md §3.2 表职责）。
|
||||
// 仓储契约（docs/flight-state.md §2 权威模型/§3 写入语义，design.md §2.1）。
|
||||
// 决策层写路径约定：所有 FLIGHT_SCHD 及明细写操作必须发生在
|
||||
// 「持有 PIPELINE_LOCK 的同一事务」内（§1：锁只串行化 DB 事务）。
|
||||
// 「持有 PIPELINE_LOCK 的同一事务」内（§1/§4：锁只串行化 DB 事务）。
|
||||
// =====================================================================
|
||||
|
||||
/** 自有 PG 单事务原子保障（§1 目标 3：状态、事件、处理终态同事务提交）。 */
|
||||
/** 自有 PG 单事务原子保障（flight-state.md §1/§4：状态、事件、处理终态同事务提交）。 */
|
||||
interface PipelineTransactionManager {
|
||||
fun <T> inTransaction(block: () -> T): T
|
||||
}
|
||||
|
||||
/** 单行锁（§3.2 PIPELINE_LOCK）：事务内第一步 SELECT ... FOR UPDATE，串行化状态写事务。 */
|
||||
/** 单行锁 PIPELINE_LOCK：事务内第一步 SELECT ... FOR UPDATE，串行化状态写事务（§1/§4）。 */
|
||||
interface PipelineLockRepository {
|
||||
fun lock()
|
||||
}
|
||||
|
||||
/** PROC_STATE：每消息一行；兼作快照重放判定（§5.1 步骤 2：MSG_ID 已有成功终态 → 重放）。 */
|
||||
/** PROC_STATE：每消息一行（design.md §2.1）；SUCCEEDED 终态兼作日计划重放判定。 */
|
||||
interface ProcStateRepository {
|
||||
fun insert(msgId: Long, state: ProcStatus = ProcStatus.PENDING)
|
||||
|
||||
@@ -61,7 +61,7 @@ interface ProcStateRepository {
|
||||
fun requeueByErrorClasses(errorClasses: List<ErrorClass>): Int
|
||||
}
|
||||
|
||||
/** MSG_EVENT outbox（§7.3）。KAFKA_SCHD 合并同 FLID 未发事件按最新 STATE_VERSION 输出。 */
|
||||
/** MSG_EVENT outbox（flight-state.md §5）。KAFKA_SCHD 合并同 FLID 未发事件按最新 STATE_VERSION 输出。 */
|
||||
interface MsgEventRepository {
|
||||
fun insertAll(events: List<MsgEvent>): List<Long>
|
||||
|
||||
@@ -69,7 +69,7 @@ interface MsgEventRepository {
|
||||
|
||||
fun claimBatch(target: String, limit: Int): List<MsgEvent>
|
||||
|
||||
/** 同 FLID 未发 KAFKA_SCHD 事件合并：每 FLID 取最新 STATE_VERSION 一条（§7.3）。 */
|
||||
/** 同 FLID 未发 KAFKA_SCHD 事件合并：每 FLID 取最新 STATE_VERSION 一条（§5）。 */
|
||||
fun mergePendingSchd(limit: Int): List<MsgEvent>
|
||||
|
||||
fun markSent(eventId: Long)
|
||||
@@ -81,62 +81,63 @@ interface MsgEventRepository {
|
||||
fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String, attempts: Int? = null)
|
||||
}
|
||||
|
||||
/** 完整态落库结果：DAY_GUARD_VIOLATION = OPERATION_DAY 不可变条件更新未命中（§7.4）。 */
|
||||
/** 完整态落库结果：DAY_GUARD_VIOLATION = OPERATION_DAY 不可变条件更新未命中（§2.1）。 */
|
||||
enum class PersistOutcome { INSERTED, UPDATED, DAY_GUARD_VIOLATION }
|
||||
|
||||
/**
|
||||
* 航班当前态权威（§3.2 FLIGHT_SCHD + 9 张明细表）。
|
||||
* 航班当前态权威（§2 FLIGHT_SCHD + 9 张明细表）。
|
||||
* 唯一写路径 = persistFullState：写入前在内存生成完整新状态（引擎产物）再落库，
|
||||
* 明细集合按组先删后插（§3.3）。
|
||||
* 明细集合按组先删后插（§2.2 完整合并结果为准）。
|
||||
*/
|
||||
interface FlightStateRepository {
|
||||
/** 主行点查（身份/版本判定；批量用于快照归属校验 §5.3）。 */
|
||||
/** 主行点查（身份/版本判定；批量用于日计划归属校验 §2.1）。 */
|
||||
fun findMainRow(flid: String): FlightMainRow?
|
||||
|
||||
fun findMainRows(flids: Collection<String>): Map<String, FlightMainRow>
|
||||
|
||||
/** 完整当前态：主行 + 全部明细（一致性读边界由调用方事务保证，§9）。 */
|
||||
/** 完整当前态：主行 + 全部明细（一致性读边界由调用方事务保证，§5）。 */
|
||||
fun loadFullSnapshot(flid: String): FlightSnapshot?
|
||||
|
||||
/**
|
||||
* 完整当前态落库（§3.3）：主行 upsert + 全部明细按组先删后插；
|
||||
* 完整当前态落库（§3）：主行 upsert + 全部明细按组先删后插；
|
||||
* STATE_VERSION 以 snapshot.stateVersion 落库。
|
||||
* SCHD（整体替换/清除）、FLOP/ADFT（合并后全量写）共用此唯一写路径。
|
||||
* 条件更新带 `WHERE operation_day IS NULL OR operation_day = :day`（§7.4 不可变强化）。
|
||||
* SCHD（§3.1）、FLOP/ADFT（§3.2/§3.3）合并后的全量结果共用此唯一写路径。
|
||||
* 条件更新带 `WHERE operation_day IS NULL OR operation_day = :day`（§2.1 不可变强化）。
|
||||
*/
|
||||
fun persistFullState(snapshot: FlightSnapshot, msgId: Long, now: Instant): PersistOutcome
|
||||
|
||||
/**
|
||||
* FDEL（§6.2）：ACTIVE → 置 DELETED、推进版本、明细保留、返回 true（发布删除事件）；
|
||||
* FDEL（§3.3）：ACTIVE → 置 DELETED、推进版本、明细保留、返回 true（发布删除事件）；
|
||||
* 已 DELETED 或不存在 → 返回 false（幂等成功，不推进版本不重复发布）。
|
||||
*/
|
||||
fun markDeleted(flid: String, msgId: Long, now: Instant): Boolean
|
||||
|
||||
/** ADFT 生命周期重激活（§6.3）：DELETED → ACTIVE，推进版本；非 DELETED 返回 false。 */
|
||||
/** ADFT 生命周期重激活（§3.3）：DELETED → ACTIVE，推进版本；非 DELETED 返回 false。 */
|
||||
fun revive(flid: String, msgId: Long, now: Instant): Boolean
|
||||
|
||||
/** §8.1：按窗口规则选出历史候选（含 DELETED；候选时间按机场时区折算）。 */
|
||||
/** §6：按保留期与终态/静默判据选出历史候选（含 DELETED；窗口按机场时区折算）。 */
|
||||
fun findHistoryCandidates(rules: HistoryRules, zone: ZoneId, now: Instant): List<HistoryCandidate>
|
||||
|
||||
/**
|
||||
* §8.2 步骤 3：物理删除主行与明细（仅历史存储确认成功后调用；
|
||||
* §6：物理删除主行与明细（仅历史存储确认成功后调用；
|
||||
* 历史存储未接通时调用方必须传空集合——删 0 条）。
|
||||
*/
|
||||
fun purgeArchived(flids: Collection<String>): Int
|
||||
|
||||
/** §8.3 观测：OPERATION_DAY 仍为 NULL 的航班数（只增不删，终止规则未定 §10）。 */
|
||||
/** §2.1 观测：OPERATION_DAY 仍为 NULL 的航班数（只增不删，终止规则未定 §6）。 */
|
||||
fun countOperationDayNull(): Int
|
||||
}
|
||||
|
||||
/** §5.5 SCHD_SNAP_LOG：事务外追加留痕，不参与决策；一行 = 一次尝试（重放也记）。 */
|
||||
/** SCHD_SNAP_LOG（design.md §6.2）：事务外追加留痕，不参与决策；一行 = 一次尝试（重放也记）。 */
|
||||
interface SnapshotLogRepository {
|
||||
fun append(entry: SnapshotLogEntry)
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 请求状态机（§7.1）：只有 RESP 完成 RQFD 请求，按（运营日、发送方、请求类型）
|
||||
* 匹配最新一条 PENDING；同类请求只留一条有效，新请求置旧为 EXPIRED。
|
||||
* 请求状态机 REQ_TRACK（design.md §4.2 目标机制，尚无运行时协调器）：
|
||||
* 只有 RESP 完成 RQFD 请求，按（运营日、发送方、请求类型）匹配最新一条 PENDING；
|
||||
* 同类请求只留一条有效，新请求置旧为 EXPIRED。
|
||||
*/
|
||||
interface ReqTrackRepository {
|
||||
enum class ReqState { PENDING, SENT, DONE, EXPIRED }
|
||||
@@ -156,7 +157,7 @@ interface ReqTrackRepository {
|
||||
|
||||
fun findLatest(reqType: String, operationDay: LocalDate, sender: String, states: List<ReqState>): Req?
|
||||
|
||||
/** RESP 完成请求（§7.1）：匹配最新一条 PENDING/SENT；无匹配返回 false（迟到不报错）。 */
|
||||
/** RESP 完成请求（design.md §4.2）：匹配最新一条 PENDING/SENT；无匹配返回 false（迟到不报错）。 */
|
||||
fun completeLatest(reqType: String, operationDay: LocalDate, sender: String): Boolean
|
||||
|
||||
fun linkCoutmsgs(reqId: Long, coutmsgsId: Long)
|
||||
@@ -167,7 +168,7 @@ interface ReqTrackRepository {
|
||||
}
|
||||
|
||||
/**
|
||||
* 共享信箱回填补偿待办（§7.2）：业务事务内预登记（消除崩溃窗口 §10 偏差），
|
||||
* 共享信箱回填补偿待办（design.md §3.3/§6.1）：业务事务内预登记，
|
||||
* 提交后由 BackfillSweepJob 重试；回填失败不得把 SUCCEEDED 改回 FAILED。
|
||||
*/
|
||||
interface BackfillTodoRepository {
|
||||
|
||||
@@ -1,72 +0,0 @@
|
||||
package com.gzzn.omms.msgexchange.infra.persistence.dialect
|
||||
|
||||
/**
|
||||
* v2 §6（ACM2-29 P3-B）：数据库方言接缝。
|
||||
*
|
||||
* 原则（v2 §1/§6）：同一业务模型支持 PostgreSQL 与 Oracle 11g，方言只存在于基础设施适配层，
|
||||
* 业务层与仓储语义不按数据库类型分支。
|
||||
*
|
||||
* 接缝范围（随仓储 SQL 演进扩充）：
|
||||
* - 航班主行快照 upsert：PG `INSERT .. ON CONFLICT` vs 11g `MERGE INTO`（11g 无 ON CONFLICT）
|
||||
* - 增量路径主行存在性保障：同上
|
||||
*
|
||||
* 激活门控（本接缝为编译级交付，**未经目标库验证**）：
|
||||
* Oracle 11.2 补丁级别 × JDK 25 × ojdbc 驱动 × Flyway（db/migration/oracle11g location）
|
||||
* × 连接池组合必须在现场 11g 实测通过后，才允许把生产方言切到 [Oracle11gDialect]；
|
||||
* MERGE 绑定顺序适配、CLOB（SRVT/VIPF/MAFL_TEXT）与空串=NULL 语义回归一并纳入激活清单。
|
||||
* 验收证据要求见 docs/flight-state.md §8。
|
||||
*/
|
||||
interface SqlDialect {
|
||||
/** 航班主行快照 upsert：按 SCALAR_WRITE_COLUMNS 生成整体替换 SQL。 */
|
||||
fun flightSnapshotUpsertSql(columns: List<String>): String
|
||||
|
||||
/** 增量路径主行存在性保障（新插 FDAY=NULL，已有行仅推进 updated_at）。 */
|
||||
fun flightRowEnsureSql(): String
|
||||
}
|
||||
|
||||
/** PostgreSQL 方言：当前生产路径。 */
|
||||
object PostgreSqlDialect : SqlDialect {
|
||||
|
||||
override fun flightSnapshotUpsertSql(columns: List<String>): String = buildString {
|
||||
append("INSERT INTO flight_schd (flid, fday, ")
|
||||
append(columns.joinToString(", "))
|
||||
append(", last_message_id, state_version, created_at, updated_at) VALUES (?, ?, ")
|
||||
append(columns.joinToString(", ") { "?" })
|
||||
append(", ?, ?, ?, ?) ON CONFLICT (flid) DO UPDATE SET fday = EXCLUDED.fday, ")
|
||||
append(columns.joinToString(", ") { "$it = EXCLUDED.$it" })
|
||||
append(", last_message_id = EXCLUDED.last_message_id, state_version = EXCLUDED.state_version, updated_at = EXCLUDED.updated_at")
|
||||
}
|
||||
|
||||
override fun flightRowEnsureSql(): String =
|
||||
"INSERT INTO flight_schd (flid, created_at, updated_at) VALUES (?, ?, ?) " +
|
||||
"ON CONFLICT (flid) DO UPDATE SET updated_at = EXCLUDED.updated_at"
|
||||
}
|
||||
|
||||
/**
|
||||
* Oracle 11g 方言：`MERGE INTO` 单语句 upsert（11g 无 `ON CONFLICT`）。
|
||||
*
|
||||
* **编译级交付，未在目标 11g 上验收**（见 [SqlDialect] 激活门控）。
|
||||
* 绑定顺序与 PG 不同（USING 子句的 flid 绑定 + MATCHED/NOT MATCHED 两份列绑定），
|
||||
* 激活时需同步实现 11g 绑定适配，禁止直接复用 PG 绑定序列。
|
||||
*/
|
||||
object Oracle11gDialect : SqlDialect {
|
||||
|
||||
override fun flightSnapshotUpsertSql(columns: List<String>): String = buildString {
|
||||
append("MERGE INTO flight_schd t ")
|
||||
append("USING (SELECT ? AS FLID FROM dual) s ON (t.flid = s.FLID) ")
|
||||
append("WHEN MATCHED THEN UPDATE SET t.fday = ?, ")
|
||||
append(columns.joinToString(", ") { "t.$it = ?" })
|
||||
append(", t.last_message_id = ?, t.state_version = ?, t.updated_at = ? ")
|
||||
append("WHEN NOT MATCHED THEN INSERT (flid, fday, ")
|
||||
append(columns.joinToString(", "))
|
||||
append(", last_message_id, state_version, created_at, updated_at) VALUES (")
|
||||
append((listOf("?", "?") + columns.map { "?" } + listOf("?", "?", "?")).joinToString(", "))
|
||||
append(")")
|
||||
}
|
||||
|
||||
override fun flightRowEnsureSql(): String =
|
||||
"MERGE INTO flight_schd t " +
|
||||
"USING (SELECT ? AS FLID FROM dual) s ON (t.flid = s.FLID) " +
|
||||
"WHEN MATCHED THEN UPDATE SET t.updated_at = ? " +
|
||||
"WHEN NOT MATCHED THEN INSERT (flid, created_at, updated_at) VALUES (?, ?, ?)"
|
||||
}
|
||||
+12
-12
@@ -36,9 +36,9 @@ import java.util.Locale
|
||||
import javax.sql.DataSource
|
||||
|
||||
// =====================================================================
|
||||
// 自有 PostgreSQL 仓储实现（docs/flight-state.md §3.2 表职责；schema 见
|
||||
// 自有 PostgreSQL 仓储实现（docs/flight-state.md §2 权威模型/§3 写入语义；schema 见
|
||||
// db/migration/V1__flight_state_baseline.sql）。全部写路径约定在
|
||||
// withTransaction + PIPELINE_LOCK 内调用（§1 目标 3/§3.2）。
|
||||
// withTransaction + PIPELINE_LOCK 内调用（§1/§4）。
|
||||
// =====================================================================
|
||||
|
||||
@Singleton
|
||||
@@ -56,7 +56,7 @@ class JdbcPipelineTransactionManager(
|
||||
class JdbcPipelineLockRepository(
|
||||
private val ds: DataSource,
|
||||
) : PipelineLockRepository {
|
||||
/** 事务内第一步：单行 FOR UPDATE 串行化状态写事务（§3.2 PIPELINE_LOCK）。 */
|
||||
/** 事务内第一步：单行 FOR UPDATE 串行化状态写事务（§1/§4 PIPELINE_LOCK）。 */
|
||||
override fun lock() {
|
||||
ds.queryOne("SELECT lock_id FROM pipeline_lock WHERE lock_id = 1 FOR UPDATE", {}) { 1 } ?: error("PIPELINE_LOCK row missing")
|
||||
}
|
||||
@@ -211,7 +211,7 @@ class JdbcMsgEventRepository(
|
||||
::mapEvent,
|
||||
)
|
||||
|
||||
/** §7.3：同 FLID 未发事件按最新 STATE_VERSION 合并；PG DISTINCT ON 方言（注释明示）。 */
|
||||
/** §5：同 FLID 未发事件按最新 STATE_VERSION 合并；PG DISTINCT ON 方言（注释明示）。 */
|
||||
override fun mergePendingSchd(limit: Int): List<MsgEvent> =
|
||||
ds.query(
|
||||
"""
|
||||
@@ -348,7 +348,7 @@ class JdbcFlightStateRepository(
|
||||
},
|
||||
)
|
||||
if (updated == 0) {
|
||||
// §7.4 不可变强化：条件更新未命中 = 归属日冲突
|
||||
// §2.1 不可变强化：条件更新未命中 = 归属日冲突
|
||||
return if (existed != null) PersistOutcome.DAY_GUARD_VIOLATION else PersistOutcome.UPDATED
|
||||
}
|
||||
replaceDetails(snapshot, now)
|
||||
@@ -370,7 +370,7 @@ class JdbcFlightStateRepository(
|
||||
) == 1
|
||||
|
||||
override fun findHistoryCandidates(rules: HistoryRules, zone: ZoneId, now: Instant): List<HistoryCandidate> {
|
||||
// §8.1 四条判定在应用层执行（SIS 时间串解析无法下推 SQL）；
|
||||
// §6 四条判定在应用层执行（SIS 时间串解析无法下推 SQL）；
|
||||
// 不按 updated_at 粗筛——取消/终态时间可能早于最近一次更新，粗筛会漏删。
|
||||
val rows = ds.query(
|
||||
"SELECT flid, state, state_version, operation_day, cncl, naat, neat, updated_at FROM flight_schd",
|
||||
@@ -399,7 +399,7 @@ class JdbcFlightStateRepository(
|
||||
r.updatedAt < now.minus(Duration.ofHours(rules.idleHours)) -> true
|
||||
else -> false
|
||||
}
|
||||
// NAAT/NEAT 业务含义待术语表确认（§10）：解析失败视为不命中，不误删
|
||||
// NAAT/NEAT 业务含义待术语表确认（§6 开放项）：解析失败视为不命中，不误删
|
||||
if (hit) HistoryCandidate(r.flid, r.state, r.version, wasNeverFdel = r.state == FlightState.ACTIVE) else null
|
||||
}
|
||||
}
|
||||
@@ -456,7 +456,7 @@ class JdbcFlightStateRepository(
|
||||
items.forEachIndexed { ordinal, item ->
|
||||
val cols = mutableListOf("flid", "ordinal", "source_seq")
|
||||
val vals = mutableListOf<Any?>(flid)
|
||||
vals.add(ordinal + 1) // ORDINAL 保留输入顺序（§3.2）
|
||||
vals.add(ordinal + 1) // ORDINAL 保留输入顺序（§2.2）
|
||||
vals.add(item[spec.seqAttr])
|
||||
spec.columns.forEach { col ->
|
||||
cols.add(col)
|
||||
@@ -488,7 +488,7 @@ class JdbcFlightStateRepository(
|
||||
|
||||
private fun loadDetails(flid: String, key: String, spec: DetailSpec): List<Map<String, String>> {
|
||||
val sql = if (spec.routeKind != null) {
|
||||
// ROUT/ERUT 共表：按 ROUTE_KIND 过滤（§3.2）
|
||||
// ROUT/ERUT 共表：按 ROUTE_KIND 过滤（§2.2）
|
||||
"SELECT * FROM ${spec.table} WHERE flid = ? AND route_kind = ? ORDER BY ordinal ASC"
|
||||
} else {
|
||||
"SELECT * FROM ${spec.table} WHERE flid = ? ORDER BY ordinal ASC"
|
||||
@@ -544,7 +544,7 @@ class JdbcFlightStateRepository(
|
||||
"flight_delay", "flight_bridge_op", "flight_chock_op", "flight_route_point",
|
||||
)
|
||||
|
||||
/** 10 类集合 ↔ 明细表/列映射（§3.2；列名与基线一致）。 */
|
||||
/** 10 类集合 ↔ 明细表/列映射（§2.2；列名与基线一致）。 */
|
||||
internal val COLLECTIONS: Map<String, DetailSpec> = mapOf(
|
||||
"GTDT" to DetailSpec("flight_gate", listOf("gate", "pgot", "pgct", "gotm", "gctm", "gtyp"), "GTNO"),
|
||||
"CKDT" to DetailSpec("flight_checkin", listOf("chkc", "ccls", "pcot", "pcct", "cotm", "cctm", "ctyp"), "CKNO"),
|
||||
@@ -566,7 +566,7 @@ class JdbcFlightStateRepository(
|
||||
class JdbcSnapshotLogRepository(
|
||||
private val ds: DataSource,
|
||||
) : SnapshotLogRepository {
|
||||
/** §5.5 只追加；写失败由调用方捕获记指标（append 自身不抛出）。 */
|
||||
/** design.md §6.2：只追加；写失败由调用方捕获记指标（append 自身不抛出）。 */
|
||||
override fun append(entry: SnapshotLogEntry) {
|
||||
ds.update(
|
||||
"""
|
||||
@@ -628,7 +628,7 @@ class JdbcReqTrackRepository(
|
||||
)
|
||||
}
|
||||
|
||||
/** §7.1 RESP 完成请求：匹配最新一条 PENDING/SENT；无匹配返回 false（迟到不报错）。 */
|
||||
/** design.md §4.2 RESP 完成请求：匹配最新一条 PENDING/SENT；无匹配返回 false（迟到不报错）。 */
|
||||
override fun completeLatest(reqType: String, operationDay: LocalDate, sender: String): Boolean =
|
||||
ds.update(
|
||||
"""
|
||||
|
||||
@@ -6,7 +6,7 @@ import jakarta.inject.Singleton
|
||||
|
||||
/**
|
||||
* stub 适配层——DeliveryPort 内存实现，仅在 msgx.stubs=true 时生效。
|
||||
* 记录 (topic, key, payload)：payload = null 表示 TOMBSTONE（§7.3 键缺失=删除旧值）。
|
||||
* 记录 (topic, key, payload)：payload = null 表示 TOMBSTONE（flight-state.md §5 键缺失=删除旧值）。
|
||||
*/
|
||||
@Requires(property = "msgx.stubs", value = "true")
|
||||
@Singleton
|
||||
|
||||
@@ -132,7 +132,7 @@ class StubMsgEvents : MsgEventRepository {
|
||||
override fun claimBatch(target: String, limit: Int): List<MsgEvent> =
|
||||
rows.values.filter { it.target == target && it.state == EventStatus.PENDING }.sortedBy { it.eventId!! }.take(limit)
|
||||
|
||||
/** §7.3：同 FLID 取最新 STATE_VERSION，按事件序输出。 */
|
||||
/** §5：同 FLID 取最新 STATE_VERSION，按事件序输出。 */
|
||||
override fun mergePendingSchd(limit: Int): List<MsgEvent> =
|
||||
rows.values
|
||||
.filter { it.target == "KAFKA:schd" && it.state == EventStatus.PENDING }
|
||||
@@ -178,7 +178,7 @@ class StubFlightState : FlightStateRepository {
|
||||
|
||||
override fun loadFullSnapshot(flid: String): FlightSnapshot? = snapshots[flid]
|
||||
|
||||
/** §7.4 不可变条件：已有非空 OPERATION_DAY 且与新值不同 → DAY_GUARD_VIOLATION。 */
|
||||
/** §2.1 不可变条件：已有非空 OPERATION_DAY 且与新值不同 → DAY_GUARD_VIOLATION。 */
|
||||
override fun persistFullState(snapshot: FlightSnapshot, msgId: Long, now: Instant): PersistOutcome {
|
||||
val existing = mains[snapshot.flid]
|
||||
if (existing?.operationDay != null && existing.operationDay != snapshot.operationDay) {
|
||||
@@ -260,7 +260,7 @@ class StubReqTrack : ReqTrackRepository {
|
||||
fun clear() = rows.clear()
|
||||
|
||||
override fun insert(reqType: String, operationDay: LocalDate, sender: String): Long {
|
||||
// §7.1：同类（类型+运营日+发送方）旧有效请求先置 EXPIRED
|
||||
// design.md §4.2：同类（类型+运营日+发送方）旧有效请求先置 EXPIRED
|
||||
rows.values.filter {
|
||||
it.reqType == reqType && it.operationDay == operationDay && it.sender == sender &&
|
||||
(it.state == ReqTrackRepository.ReqState.PENDING || it.state == ReqTrackRepository.ReqState.SENT)
|
||||
|
||||
@@ -14,13 +14,13 @@ import java.time.Instant
|
||||
import java.time.ZoneId
|
||||
|
||||
/**
|
||||
* 历史归档与物理清除（docs/flight-state.md §8.2，顺序不可颠倒）：
|
||||
* 1. HISTORY_SWEEP 选出满足 §8.1 的航班（含 DELETED）；
|
||||
* 历史归档与物理清除（docs/flight-state.md §6 生命周期，顺序不可颠倒；design.md §6.2）：
|
||||
* 1. 选出满足保留期 + 终态/静默判据的航班（含 DELETED）；
|
||||
* 2. 写入历史存储；
|
||||
* 3. 历史存储返回成功的 FLID 集合 → 物理删除主行与明细（归档结果只记录在历史存储，当前态无 ARCHIVED 态）；
|
||||
* 4. 未经 FDEL 的航班在清除前补发一次删除事件（§7.3）；其余不发；
|
||||
* 4. 未经 FDEL 的航班在清除前补发一次删除事件（§3.3/§5）；其余不发；
|
||||
* 5. 失败或不明确的保留重试。
|
||||
* 前提红线：历史存储未接通时必须删 0 条（§8.2）。
|
||||
* 前提红线：历史存储未接通时必须删 0 条（§6）。
|
||||
*/
|
||||
@Singleton
|
||||
class HistorySweepJob(
|
||||
@@ -29,7 +29,7 @@ class HistorySweepJob(
|
||||
private val props: HistoryProps,
|
||||
/** 历史存储端口：返回归档成功的 FLID 集合；未接通时不注入（null）。 */
|
||||
private val historyStore: HistoryStore? = null,
|
||||
/** §8.3 留痕清理端口（90 天，按 (SCOPE_END, RECV_AT)）；未接通时不注入。 */
|
||||
/** design.md §6.2 留痕清理端口（90 天，按 (SCOPE_END, RECV_AT)）；未接通时不注入。 */
|
||||
private val snapLogPurge: SnapshotLogPurge? = null,
|
||||
) {
|
||||
/** 历史存储端口（由部署侧适配实现；脚手架默认未接通）。 */
|
||||
@@ -46,12 +46,12 @@ class HistorySweepJob(
|
||||
|
||||
fun run(now: Instant = Instant.now()): SweepOutcome {
|
||||
if (!props.historyStoreEnabled || historyStore == null) {
|
||||
// 红线：历史存储未接通必须删 0 条；绝不允许先删当前态再补历史（§8.2）
|
||||
// 红线：历史存储未接通必须删 0 条；绝不允许先删当前态再补历史（§6）
|
||||
return SweepOutcome(selected = 0, archived = 0, purged = 0)
|
||||
}
|
||||
|
||||
val rules = HistoryRules(props.cancelledHours, props.terminalHours, props.deletedHours, props.idleHours)
|
||||
val zone = ZoneId.of("Asia/Shanghai") // §8.1：窗口按机场时区计算
|
||||
val zone = ZoneId.of("Asia/Shanghai") // §6：窗口按机场时区计算
|
||||
val candidates = flightState.findHistoryCandidates(rules, zone, now)
|
||||
if (candidates.isEmpty()) return SweepOutcome(0, 0, 0)
|
||||
|
||||
@@ -59,7 +59,7 @@ class HistorySweepJob(
|
||||
if (archivedFlids.isEmpty()) return SweepOutcome(candidates.size, archived = 0, purged = 0)
|
||||
|
||||
val toPurge = candidates.filter { it.flid in archivedFlids }
|
||||
// §8.2 步骤 4：未经 FDEL、由生命周期直接清除的航班，清除前补发一次删除事件
|
||||
// §6/§3.3：未经 FDEL、由生命周期直接清除的航班，清除前补发一次删除事件
|
||||
val preDelete = toPurge.filter { it.wasNeverFdel }
|
||||
if (preDelete.isNotEmpty()) {
|
||||
msgEvents.insertAll(preDelete.map { tombstone(it) })
|
||||
|
||||
@@ -9,10 +9,9 @@ import java.time.LocalDate
|
||||
import java.time.ZoneId
|
||||
|
||||
/**
|
||||
* 维护作业调度（docs/flight-state.md §8）：单 daemon 线程，独立于主泵——
|
||||
* 旧「作业不插队 PUMP_JOB 队列」机制随审计口径移除（§3.2 表清单无 PUMP_JOB）；
|
||||
* 历史归档/留痕清理均为内部清理路径，不参与 FIFO 消息序。
|
||||
* 触发：回填补偿 30s 固定间隔；历史归档/留痕清理每日 03:30（机场时区）后首个 tick。
|
||||
* 维护作业调度（docs/design.md §6.1/§6.2）：单 daemon 线程，独立于主泵——
|
||||
* 作业不参与消息 FIFO，也不使到期消息饥饿（PUMP_JOB 队列机制已随审计口径移除）。
|
||||
* 触发：回填补偿扫描 30s 固定间隔；历史归档/留痕清理每日机场时区 03:30 后首个 tick。
|
||||
*/
|
||||
@Singleton
|
||||
class JobRunner(
|
||||
|
||||
@@ -10,8 +10,10 @@ import com.gzzn.omms.msgexchange.domain.OperationDayCalculator
|
||||
import com.gzzn.omms.msgexchange.domain.ProcState
|
||||
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.domain.flight.FlightStateEngine
|
||||
import com.gzzn.omms.msgexchange.domain.flight.MergeChange
|
||||
import com.gzzn.omms.msgexchange.domain.flight.ScheduleRecord
|
||||
import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository
|
||||
import com.gzzn.omms.msgexchange.infra.persistence.FlightStateRepository
|
||||
import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository
|
||||
@@ -22,8 +24,9 @@ import java.time.Instant
|
||||
import java.time.ZoneId
|
||||
|
||||
/**
|
||||
* FLOP（§6.1）：读取完整当前态 → 合并变化 → 保留运营日 → STATE_VERSION+1 →
|
||||
* 同事务登记 KAFKA_MSG / KAFKA_SCHD 与处理终态。
|
||||
* FLOP（docs/flight-state.md §3.2 动态运行事件）：读取完整当前态 → 合并变化 →
|
||||
* 保留运营日 → STATE_VERSION+1 → 同事务登记 KAFKA_MSG / KAFKA_SCHD 与处理终态。
|
||||
* 未知/迟到航班按幂等成功处理，不创建实例（创建入口只有 SCHD/ADFT）。
|
||||
*/
|
||||
@Singleton
|
||||
class FlopProcessor(
|
||||
@@ -49,8 +52,8 @@ class FlopProcessor(
|
||||
}
|
||||
|
||||
/**
|
||||
* FDEL（§6.2）：ACTIVE → 置 DELETED、推进版本、明细保留、发布 tombstone；
|
||||
* 已 DELETED / 不存在 → 幂等成功，不推进版本、不重复发布。
|
||||
* FDEL（docs/flight-state.md §3.3 删除）：ACTIVE → 置 DELETED、推进版本、明细保留、
|
||||
* 与删除同事务登记 tombstone（§5）；已 DELETED / 不存在 → 幂等成功，不推进版本、不重复发布。
|
||||
*/
|
||||
@Singleton
|
||||
class FdelProcessor(
|
||||
@@ -66,7 +69,7 @@ class FdelProcessor(
|
||||
val deleted = flightState.markDeleted(payload.flid, msgId = head.msgId, now = Instant.now())
|
||||
if (deleted) {
|
||||
val current = flightState.loadFullSnapshot(payload.flid)
|
||||
// tombstone 仅在 ACTIVE→DELETED 时登记（§7.3），与删除同事务
|
||||
// tombstone 仅在 ACTIVE→DELETED 时登记（§3.3/§5），与删除同事务
|
||||
msgEvents.insertAll(
|
||||
listOf(
|
||||
MsgEvent(
|
||||
@@ -94,14 +97,14 @@ class FdelProcessor(
|
||||
)
|
||||
preRegisterBackfill(head, msg, backfillTodo)
|
||||
}
|
||||
ApplyResult.Succeeded // 未命中 = 迟到/重复，幂等成功（§6.2 步骤 3/4）
|
||||
ApplyResult.Succeeded // 未命中 = 迟到/重复，幂等成功（§3.3）
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* ADFT（§6.3 + §2.1）：字段缺失语义待确认——确认前按保守 Set-only 处理
|
||||
* （出现字段覆盖，缺失不 Clear，不沿用 FLOP 全量合并规则）。
|
||||
* FLID 已存在且 DELETED → 生命周期重激活；不存在 → 新实例建立（含运营日计算 §3.5）。
|
||||
* ADFT（docs/flight-state.md §3.3）：字段缺失语义待上游确认——确认前按保守 Set-only
|
||||
* 处理（出现字段覆盖、缺失不清空）。FLID 已存在且 DELETED → 生命周期重激活；不存在 →
|
||||
* 新实例建立（含 SODT 时直接计算 OPERATION_DAY，§2.1；否则保留 NULL 待日计划收录）。
|
||||
*/
|
||||
@Singleton
|
||||
class AdftProcessor(
|
||||
@@ -118,12 +121,12 @@ class AdftProcessor(
|
||||
cutoffHour = operationDayProps.cutoffHour,
|
||||
)
|
||||
|
||||
fun apply(head: ProcState, msg: DecodedMessage, record: com.gzzn.omms.msgexchange.domain.flight.ScheduleRecord): ApplyResult =
|
||||
fun apply(head: ProcState, msg: DecodedMessage, record: ScheduleRecord): ApplyResult =
|
||||
txManager.inTransaction {
|
||||
lock.lock()
|
||||
val main = flightState.findMainRow(record.flid)
|
||||
if (main != null && main.state == com.gzzn.omms.msgexchange.domain.flight.FlightState.DELETED) {
|
||||
// §6.3 重激活：DELETED → ACTIVE，推进版本，登记状态事件
|
||||
if (main != null && main.state == FlightState.DELETED) {
|
||||
// §3.3 重激活：DELETED → ACTIVE，推进版本，登记状态事件
|
||||
if (flightState.revive(record.flid, msgId = head.msgId, now = Instant.now())) {
|
||||
val current = flightState.loadFullSnapshot(record.flid)
|
||||
if (current != null) {
|
||||
@@ -138,15 +141,15 @@ class AdftProcessor(
|
||||
|
||||
val current = flightState.loadFullSnapshot(record.flid)
|
||||
val next: FlightSnapshot = if (current == null) {
|
||||
// 新实例建立：ADFT 含 SODT 时直接计算运营日（§2.1），不可算则置 null 待快照收录
|
||||
// 新实例建立：ADFT 含 SODT 时直接计算运营日（§2.1），不可算则置 null 待日计划收录
|
||||
val day = opDay.compute(record.scalars["SODT"])
|
||||
FlightSnapshot(
|
||||
flid = record.flid,
|
||||
operationDay = day, // 待确认项 §2.1：不可算时不得默认写接收日
|
||||
state = com.gzzn.omms.msgexchange.domain.flight.FlightState.ACTIVE,
|
||||
operationDay = day, // §2.1：不可算时不得默认写接收日
|
||||
state = FlightState.ACTIVE,
|
||||
stateVersion = 1L,
|
||||
scalars = record.scalars,
|
||||
collections = com.gzzn.omms.msgexchange.domain.flight.FlightStateEngine.COLLECTION_KEYS.associateWith { key ->
|
||||
collections = FlightStateEngine.COLLECTION_KEYS.associateWith { key ->
|
||||
record.collections[key] ?: emptyList()
|
||||
},
|
||||
)
|
||||
@@ -162,8 +165,8 @@ class AdftProcessor(
|
||||
ApplyResult.Succeeded
|
||||
}
|
||||
|
||||
/** §2.1 保守语义：仅出现字段 Set；集合出现 Replace、缺失保留。 */
|
||||
private fun setOnly(record: com.gzzn.omms.msgexchange.domain.flight.ScheduleRecord) = MergeChange(
|
||||
/** §3.3 保守语义：仅出现字段 Set；集合出现 Replace、缺失保留。 */
|
||||
private fun setOnly(record: ScheduleRecord) = MergeChange(
|
||||
flid = record.flid,
|
||||
scalars = record.scalars,
|
||||
collections = record.collections,
|
||||
@@ -174,10 +177,10 @@ class AdftProcessor(
|
||||
// 共享小工具（处理器层私有约定）
|
||||
// =====================================================================
|
||||
|
||||
/** 航班不存在/迟到：幂等成功（§9 队头不阻塞；不创建实例——创建入口只有 SCHD/ADFT）。 */
|
||||
/** 航班不存在/迟到：幂等成功（§3.2/§3.3；不阻塞队头，不创建实例——创建入口只有 SCHD/ADFT）。 */
|
||||
private fun idempotentAbsent(head: ProcState, msg: DecodedMessage): ApplyResult = ApplyResult.Succeeded
|
||||
|
||||
/** KAFKA_SCHD 整态 + KAFKA_MSG 变化通知（§7.3）。 */
|
||||
/** KAFKA_SCHD 整态 + KAFKA_MSG 变化通知（flight-state.md §3.1/§5）。 */
|
||||
internal fun eventsFor(next: FlightSnapshot, mapper: ObjectMapper): List<MsgEvent> {
|
||||
val payload = linkedMapOf<String, Any>(
|
||||
"flid" to next.flid,
|
||||
@@ -201,7 +204,7 @@ internal fun eventsFor(next: FlightSnapshot, mapper: ObjectMapper): List<MsgEven
|
||||
)
|
||||
}
|
||||
|
||||
/** §7.2 业务事务内预登记回填待办（与终态同事务）。 */
|
||||
/** 回填待办与业务终态同事务预登记（design.md §3.3/§6.1；提交后回填并删待办）。 */
|
||||
internal fun preRegisterBackfill(head: ProcState, msg: DecodedMessage, backfillTodo: BackfillTodoRepository?) {
|
||||
backfillTodo?.record(
|
||||
BackfillTodoRepository.BackfillTask(
|
||||
@@ -211,4 +214,3 @@ internal fun preRegisterBackfill(head: ProcState, msg: DecodedMessage, backfillT
|
||||
null,
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -19,8 +19,9 @@ import java.time.Duration
|
||||
import java.time.Instant
|
||||
|
||||
/**
|
||||
* 处理主泵（docs/flight-state.md §1 目标 2）：单活动主泵严格 FIFO + HOL + 毒丸 DEAD 升级；
|
||||
* 航班状态、事件、处理终态在处理器事务内原子提交（§1 目标 3）。
|
||||
* 处理主泵（docs/flight-state.md §1 严格有序 + §4 处理事务与失败规则）：
|
||||
* 单活动主泵严格 FIFO + HOL 阻塞 + 队头滞留转 DEAD；航班状态、事件、处理终态
|
||||
* 与回填待办在处理器事务内原子提交。
|
||||
*/
|
||||
@Singleton
|
||||
class Pump(
|
||||
@@ -84,7 +85,7 @@ class Pump(
|
||||
/**
|
||||
* processOne：解码 → 绑定 → 处理器（事务内决策+落库）→ 终态迁移 → 回填。
|
||||
* 边界化失败迁移（ProcFailure）：任何意外异常归于本条 head，FAILED(INFRA)+退避，不穿出杀泵；
|
||||
* MALFORMED / PROTOCOL 直接 DEAD 不重试（§9）。
|
||||
* MALFORMED / PROTOCOL 直接 DEAD 不重试（docs/design.md §2.3 错误分类）。
|
||||
*/
|
||||
@Singleton
|
||||
class MessageProcessor(
|
||||
@@ -156,7 +157,7 @@ class MessageProcessor(
|
||||
}
|
||||
}
|
||||
|
||||
// 处理器分派：SCHD 快照主链路（§5.1）/ FLOP / FDEL / ADFT；缺载荷按 MALFORMED 终态
|
||||
// 处理器分派：SCHD 日计划主链路（§3.1）/ FLOP / FDEL / ADFT；缺载荷按 MALFORMED 终态
|
||||
val result: ApplyResult = when (val kind = decoded.kind) {
|
||||
is MsgKind.Schd -> {
|
||||
val body = decoded.body as? ScheduleBody
|
||||
@@ -194,7 +195,7 @@ class MessageProcessor(
|
||||
flopProcessor.apply(head, decoded, payload)
|
||||
}
|
||||
is MsgKind.Unsupported -> {
|
||||
// §9：未支持类型 → FAILED(UNSUPPORTED) 退避重试，达阈值转 DEAD；绝不写终态
|
||||
// design.md §2.3：未支持类型 → FAILED(UNSUPPORTED) 退避重试，达阈值转 DEAD；绝不写终态
|
||||
log.warn("unsupported type -> FAILED(UNSUPPORTED) msgId={} tag={}", head.msgId, kind.tag)
|
||||
procFailure.fail(head, ErrorClass.UNSUPPORTED, "no-handler:${kind.tag}")
|
||||
return
|
||||
@@ -205,7 +206,7 @@ class MessageProcessor(
|
||||
is ApplyResult.Succeeded, ApplyResult.ReplaySkipped ->
|
||||
procState.update(head.msgId, ProcStatus.SUCCEEDED)
|
||||
is ApplyResult.DeadProtocol -> {
|
||||
// §9：整包拒绝 DEAD(PROTOCOL)，立即释放队头，交人工确认
|
||||
// design.md §2.3：整包拒绝 DEAD(PROTOCOL)，立即释放队头，交人工确认
|
||||
log.error("DEAD(PROTOCOL) msgId={} reason={} flags={}", head.msgId, result.reason, result.flags)
|
||||
procState.update(head.msgId, ProcStatus.DEAD, errorClass = ErrorClass.PROTOCOL, lastError = result.reason.take(1000))
|
||||
compensateBackfill(head, decoded) // 拒绝包同样要回填信箱，防止反复轮询
|
||||
@@ -217,7 +218,7 @@ class MessageProcessor(
|
||||
log.info("SUCCEEDED msgId={} kind={}", head.msgId, decoded.typeTag)
|
||||
}
|
||||
|
||||
/** §7.2：提交后回填共享信箱；失败不得把 SUCCEEDED 改回 FAILED，待办已事务内预登记。 */
|
||||
/** design.md §3.3/§6.1：提交后回填共享信箱；失败不得把 SUCCEEDED 改回 FAILED，待办已事务内预登记。 */
|
||||
private fun backfill(head: ProcState, decoded: DecodedMessage) {
|
||||
try {
|
||||
inbox.backfillOnSuccess(head.msgId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn)
|
||||
|
||||
@@ -6,7 +6,6 @@ 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
|
||||
@@ -30,24 +29,25 @@ import java.time.Instant
|
||||
import java.time.LocalDate
|
||||
import java.time.ZoneId
|
||||
|
||||
/** 处理器执行结果——终态迁移由 MessageProcessor 统一落库。 */
|
||||
/** 处理器执行结果——终态迁移由 MessageProcessor 统一落库（docs/design.md §2.3）。 */
|
||||
sealed interface ApplyResult {
|
||||
/** 业务成功（含幂等成功）。 */
|
||||
data object Succeeded : ApplyResult
|
||||
|
||||
/** §5.1 步骤 2：MSG_ID 已有成功终态 → 重放，直接记幂等成功。 */
|
||||
/** MSG_ID 已有成功终态 → 重放，直接记幂等成功（docs/flight-state.md §4 重放判定）。 */
|
||||
data object ReplaySkipped : ApplyResult
|
||||
|
||||
/** §9：整包拒绝 DEAD(PROTOCOL)，不重试，交人工确认。 */
|
||||
/** 整包拒绝 DEAD(PROTOCOL)（flight-state.md §3.1/§4），不重试，交人工确认。 */
|
||||
data class DeadProtocol(val reason: String, val flags: Set<SnapshotFlag> = emptySet()) : ApplyResult
|
||||
}
|
||||
|
||||
/** §5.3 第四行：归属日不符 = 串日/错发/污染，整包拒绝。 */
|
||||
/** 归属日冲突（flight-state.md §2.1：OPERATION_DAY 一经确定不可变）= 串日/错发/污染，整包拒绝。 */
|
||||
class ProtocolViolation(message: String) : RuntimeException(message)
|
||||
|
||||
/**
|
||||
* SCHD 快照主链路（docs/flight-state.md §5.1 applyScheduleRecords，同一事务）：
|
||||
* 对单日快照与滚动窗口统一适用，不做名单层面的处理。
|
||||
* SCHD 日计划主链路（docs/flight-state.md §3.1/§4）：对 DNLD 与 RESP 统一适用。
|
||||
* 同一事务内：锁 → 归属日校验 → 逐条合并写完整当前态 → 登记事件与回填待办；
|
||||
* 整包校验失败或运营日冲突整包不落地。
|
||||
*/
|
||||
@Singleton
|
||||
class ScheduleProcessor(
|
||||
@@ -70,18 +70,16 @@ class ScheduleProcessor(
|
||||
val body = msg.body as? ScheduleBody ?: return ApplyResult.DeadProtocol("missing-schd-body")
|
||||
val started = System.nanoTime()
|
||||
|
||||
// ② 重放判定：MSG_ID 已有成功终态 → 幂等成功（§5.1 步骤 2，重放也记留痕）
|
||||
// 重放判定：MSG_ID 已有成功终态 → 幂等成功（flight-state.md §4），重放也记留痕
|
||||
if (procState.findSuccessTerminal(head.msgId)) {
|
||||
logSnapshot(head, body, SnapshotResult.REPLAY_SKIPPED, upserted = 0, flags = emptySet(), started)
|
||||
return ApplyResult.ReplaySkipped
|
||||
}
|
||||
|
||||
// ③ 报文完整性（§5.2 五项）：任一失败整包不落地 → DEAD(PROTOCOL)
|
||||
// 整包校验（§4 步骤 2 / design.md §4.1）：任一失败整包不落地 → DEAD(PROTOCOL)
|
||||
val validation = FlightStateEngine.validateMessage(
|
||||
recsDeclared = body.recsDeclared,
|
||||
records = body.records,
|
||||
scopeStart = body.scopeStart,
|
||||
scopeEnd = body.scopeEnd,
|
||||
opDay = opDay,
|
||||
)
|
||||
if (validation is SnapshotValidation.Invalid) {
|
||||
@@ -91,18 +89,18 @@ class ScheduleProcessor(
|
||||
val ok = validation as SnapshotValidation.Ok
|
||||
|
||||
if (ok.perRecordDay.isEmpty()) {
|
||||
// 空快照：合法但无写入（§5.5 EMPTY），仍算成功终态
|
||||
// 空快照：合法但无写入，仍算成功终态
|
||||
logSnapshot(head, body, SnapshotResult.COMMITTED, upserted = 0, setOf(SnapshotFlag.EMPTY), started)
|
||||
return ApplyResult.Succeeded
|
||||
}
|
||||
|
||||
val flags = linkedSetOf<SnapshotFlag>()
|
||||
return try {
|
||||
// ①④⑤⑥⑦ 同一事务：锁 → 归属校验 → upsert → 版本/事件/待办预登记/终态
|
||||
// 同一事务（flight-state.md §4）：锁 → 归属校验 → 逐条合并写 → 事件/待办预登记
|
||||
val upserted = txManager.inTransaction {
|
||||
lock.lock()
|
||||
|
||||
// ⑤ 归属校验（§5.3）：OPERATION_DAY 不可变，批量点查避免逐航班往返
|
||||
// 归属校验（§2.1）：OPERATION_DAY 不可变，批量点查避免逐航班往返
|
||||
val mains = flightState.findMainRows(ok.perRecordDay.keys)
|
||||
ok.perRecordDay.forEach { (flid, day) ->
|
||||
val existing = mains[flid] ?: return@forEach
|
||||
@@ -119,7 +117,7 @@ class ScheduleProcessor(
|
||||
val record = body.records.first { it.flid == flid }
|
||||
val existingMain = mains[flid]
|
||||
val keepDeleted = existingMain?.state == FlightState.DELETED
|
||||
if (keepDeleted) flags.add(SnapshotFlag.SCHD_REVIVE_CONFLICT) // §5.1 步骤 6：不恢复
|
||||
if (keepDeleted) flags.add(SnapshotFlag.SCHD_REVIVE_CONFLICT) // §3.3：日计划不复活 DELETED
|
||||
val current = if (existingMain != null) flightState.loadFullSnapshot(flid) else null
|
||||
val next = FlightStateEngine.snapshotState(
|
||||
current = current,
|
||||
@@ -135,7 +133,7 @@ class ScheduleProcessor(
|
||||
events += snapshotEvents(next)
|
||||
}
|
||||
if (events.isNotEmpty()) msgEvents.insertAll(events)
|
||||
// §7.2 目标形态：业务事务内预登记回填待办（提交后由 MessageProcessor 回填并删待办）
|
||||
// 回填待办与业务终态同事务预登记（design.md §3.3/§6.1；提交后由 MessageProcessor 回填并删待办）
|
||||
backfillTodo?.record(
|
||||
BackfillTodoRepository.BackfillTask(
|
||||
msgId = head.msgId, sndr = msg.meta.sndr, type = msg.meta.type,
|
||||
@@ -153,9 +151,8 @@ class ScheduleProcessor(
|
||||
}
|
||||
}
|
||||
|
||||
/** 状态事件：整态出站（§7.3 KAFKA_SCHD）+ 变化通知（KAFKA_MSG）。 */
|
||||
/** 状态事件：KAFKA_SCHD 整态 + KAFKA_MSG 变化通知（flight-state.md §3.1 登记、§5 语义）。 */
|
||||
private fun snapshotEvents(next: FlightSnapshot): List<MsgEvent> {
|
||||
// KAFKA_SCHD 整态：缺失集合输出空集 → 消费者删除旧值（§5.4/§7.3）
|
||||
val payload = linkedMapOf<String, Any>(
|
||||
"flid" to next.flid,
|
||||
"stateVersion" to next.stateVersion,
|
||||
@@ -174,7 +171,8 @@ class ScheduleProcessor(
|
||||
)
|
||||
}
|
||||
|
||||
/** §5.5 留痕：事务外追加，失败只记 error 不阻塞；scope 按记录归属运营日推导。 */
|
||||
/** 留痕 SCHD_SNAP_LOG（design.md §6.2）：事务外追加，失败只记 error 不阻塞；
|
||||
* scope 为报文各记录归属运营日的最小/最大（单日快照两者相等）。 */
|
||||
private fun logSnapshot(
|
||||
head: ProcState,
|
||||
body: ScheduleBody,
|
||||
|
||||
@@ -7,17 +7,19 @@ Flyway 配置**；PG 路径使用 `classpath:db/migration`，两者互不混用
|
||||
|
||||
1. 现场 11.2 补丁级别、数据库字符集、DBA 权限清单拿到，且可提供可测试的目标库。
|
||||
2. JDK 25 × ojdbc 驱动（具体版本）× Flyway Oracle 支持 × 连接池组合在目标库实测通过
|
||||
——不能以"PG 通过"代替 Oracle 验收（docs/flight-state.md）。
|
||||
3. `SqlDialect` 切到 `Oracle11gDialect` 前，MERGE 绑定顺序适配完成并通过
|
||||
`FlightSchdJdbcPgTest` 同等粒度的 11g 集成测试。
|
||||
——不能以"PG 通过"代替 Oracle 验收（Oracle 11g 适配为开放项，docs/flight-state.md §6）。
|
||||
3. 11g 的 upsert（MERGE INTO）与绑定顺序适配完成并通过与 PG 同粒度的集成测试后，
|
||||
才允许把仓储 SQL 切到 11g 方言。先前编译级交付的 `SqlDialect` 方言接缝已随
|
||||
V1 基线列名更替（fday/last_message_id → operation_day/last_msg_id）退役删除；
|
||||
接入时必须按 V1 基线现有列名重建，禁止直接复用旧接缝。
|
||||
|
||||
## 计划内容
|
||||
|
||||
- V1 起：`V1__flight_state_baseline.sql` 的 11g 等价 DDL——
|
||||
PIPELINE_LOCK/PROC_STATE/MSG_EVENT/REQ_TRACK/BACKFILL_TODO/FLIGHT_SCHD + 9 张明细表/SCHD_SNAP_LOG；
|
||||
`NUMBER`/序列替代 `BIGSERIAL`、`TIMESTAMP WITH TIME ZONE`、`VARCHAR2` BYTE/CHAR 语义钉死。
|
||||
- JdbcFlightStateRepository 的 `INSERT ... ON CONFLICT` 改 MERGE（OPERATION_DAY 不可变条件 §7.4）。
|
||||
- 空串按 NULL 的语义回归：命令层 presence 信息不得被 11g 空串语义吞掉
|
||||
（docs/flight-state.md §5/§10）。
|
||||
- `INSERT ... ON CONFLICT` 改 11g MERGE（OPERATION_DAY 不可变条件，flight-state.md §2.1）。
|
||||
- 空串按 NULL 的语义回归：显式清空的 presence 信息不得被 11g 空串语义吞掉
|
||||
（flight-state.md §3.1 显式清空口径）。
|
||||
|
||||
版本号与 PG location 各自独立推进，禁止复用版本号语义。
|
||||
|
||||
@@ -58,7 +58,7 @@ class PipelineSmokeTest {
|
||||
}
|
||||
|
||||
companion object {
|
||||
/** 合法 META，但 TYPE 未在支持范围 → FAILED(UNSUPPORTED)（§9） */
|
||||
/** 合法 META，但 TYPE 未在支持范围 → FAILED(UNSUPPORTED)（design.md §2.3） */
|
||||
val UNSUPPORTED_XML = """
|
||||
<MSG>
|
||||
<META><SNDR>AODB</SNDR><SEQN>1</SEQN><DTTM>20260908120000</DTTM><TYPE>XYZQ</TYPE><STYP>FOO</STYP></META>
|
||||
|
||||
@@ -3,6 +3,7 @@ package com.gzzn.omms.msgexchange.codec
|
||||
import com.gzzn.omms.msgexchange.domain.MsgKind
|
||||
import org.junit.jupiter.api.Assertions.assertEquals
|
||||
import org.junit.jupiter.api.Assertions.assertInstanceOf
|
||||
import org.junit.jupiter.api.Assertions.assertNull
|
||||
import org.junit.jupiter.api.Assertions.assertTrue
|
||||
import org.junit.jupiter.api.Test
|
||||
|
||||
@@ -57,7 +58,8 @@ class JacksonXmlCodecTest {
|
||||
|
||||
val body = msg.body as FlopPayload
|
||||
assertEquals("121112312", body.flid)
|
||||
assertEquals("CA-CA101-A-12DEC261345-D", body.scalars["FFID"])
|
||||
// FFID（友好航班号）是 wire 必填字段，但当前状态模型不消费它，映射层应忽略。
|
||||
assertNull(body.scalars["FFID"])
|
||||
assertEquals(3, body.collections["GTDT"]!!.size)
|
||||
assertEquals("G28", body.collections["GTDT"]!![0]["GATE"])
|
||||
assertEquals("1", body.collections["GTDT"]!![0]["GTNO"])
|
||||
|
||||
@@ -16,7 +16,7 @@ import org.junit.jupiter.api.Test
|
||||
import java.time.Instant
|
||||
|
||||
/**
|
||||
* 投递调度（docs/flight-state.md §7.3）：
|
||||
* 投递调度（docs/flight-state.md §5 + design.md §5.1/§5.2）：
|
||||
* KAFKA_MSG 逐条 FIFO；KAFKA_SCHD 唯一出口 flushSchd——同 FLID 未发事件按最新
|
||||
* STATE_VERSION 合并；TOMBSTONE 发 null 值消息；失败退避重试、达上限 DEAD。
|
||||
*/
|
||||
@@ -84,7 +84,7 @@ class DispatcherTickTest {
|
||||
d.flushSchd()
|
||||
|
||||
assertEquals(1, port.tombstones.size)
|
||||
assertEquals("F1", port.tombstones.single().key) // 整态键缺失表示删除旧值（§7.3）
|
||||
assertEquals("F1", port.tombstones.single().key) // §5：整态键缺失表示删除旧值
|
||||
assertNull(port.tombstones.single().payload)
|
||||
}
|
||||
|
||||
|
||||
@@ -8,7 +8,7 @@ import java.time.LocalDate
|
||||
import java.time.ZoneId
|
||||
|
||||
/**
|
||||
* 运营日计算（docs/flight-state.md §3.5）：SODT（ddMMMyyHHmm）+ 机场时区 + 业务切日边界。
|
||||
* 运营日计算（docs/flight-state.md §2.1）：SODT（ddMMMyyHHmm）+ 机场时区 + 业务切日边界。
|
||||
*/
|
||||
class OperationDayTest {
|
||||
|
||||
|
||||
@@ -10,9 +10,10 @@ import java.time.LocalDate
|
||||
import java.time.ZoneId
|
||||
|
||||
/**
|
||||
* 航班状态引擎（docs/flight-state.md §3.3/§5.2/§5.4/§6.1）不变量：
|
||||
* 快照整体替换（缺失=清除）、DELETED 不被 SCHD 恢复、FLOP 增量保留缺失字段、
|
||||
* §5.2 五项校验整包拒绝语义。
|
||||
* 航班状态引擎（docs/flight-state.md §2–§4）不变量：
|
||||
* 日计划合并（§3.1）——出现 Set/Replace、缺失保留、标量空值显式清空；
|
||||
* DELETED 不被日计划恢复（§3.3）；FLOP/ADFT 增量合并（§3.2/§3.3）；
|
||||
* 整包校验（§4 步骤 2：声明数量、航班标识、运营日推导）失败整包拒绝。
|
||||
*/
|
||||
class FlightStateEngineTest {
|
||||
|
||||
@@ -20,7 +21,7 @@ class FlightStateEngineTest {
|
||||
private val day = LocalDate.of(2026, 12, 15)
|
||||
|
||||
@Test
|
||||
fun `snapshot replaces scalars and clears absent ones`() {
|
||||
fun `day plan merges scalars overwriting overlaps retaining absent and clearing on explicit empty`() {
|
||||
val current = FlightSnapshot(
|
||||
"121", day, FlightState.ACTIVE, 5,
|
||||
scalars = mapOf("FLNO" to "CA001", "REMC" to "old-note"),
|
||||
@@ -28,25 +29,66 @@ class FlightStateEngineTest {
|
||||
)
|
||||
val next = FlightStateEngine.snapshotState(
|
||||
current,
|
||||
ScheduleRecord("121", scalars = mapOf("FLNO" to "CA002")),
|
||||
ScheduleRecord("121", scalars = mapOf("FLNO" to "CA002", "CNCL" to "")),
|
||||
operationDay = day,
|
||||
keepDeleted = false,
|
||||
)
|
||||
|
||||
assertEquals("CA002", next.scalars["FLNO"])
|
||||
assertFalse(next.scalars.containsKey("REMC")) // 缺失 = Clear（§5.4）
|
||||
assertEquals(emptyList<Map<String, String>>(), next.collections["GTDT"]) // 缺失 = Replace 空集
|
||||
assertEquals(6, next.stateVersion) // 每次成功写入 +1（§4）
|
||||
assertEquals("CA002", next.scalars["FLNO"]) // 重叠字段被覆盖（§3.1）
|
||||
assertEquals("old-note", next.scalars["REMC"]) // 缺失标量保留（§3.1/§7）
|
||||
assertEquals("", next.scalars["CNCL"]) // 空串 = 显式清空（落库置 NULL）
|
||||
assertEquals(listOf(mapOf("GATE" to "G1")), next.collections["GTDT"]) // 缺失集合保留
|
||||
assertEquals(6, next.stateVersion) // 每次成功写入 +1
|
||||
assertEquals(FlightState.ACTIVE, next.state)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `snapshot keeps DELETED state and flags revive conflict by caller`() {
|
||||
fun `day plan replaces collections that are present in the record`() {
|
||||
val current = FlightSnapshot(
|
||||
"121", day, FlightState.ACTIVE, 3,
|
||||
scalars = mapOf("FLNO" to "CA001"),
|
||||
collections = mapOf("GTDT" to listOf(mapOf("GATE" to "G1"))),
|
||||
)
|
||||
val next = FlightStateEngine.snapshotState(
|
||||
current,
|
||||
ScheduleRecord(
|
||||
"121",
|
||||
scalars = mapOf("FLNO" to "CA002"),
|
||||
collections = mapOf("GTDT" to listOf(mapOf("GATE" to "G2"), mapOf("GATE" to "G3"))),
|
||||
),
|
||||
operationDay = day,
|
||||
keepDeleted = false,
|
||||
)
|
||||
|
||||
assertEquals(listOf(mapOf("GATE" to "G2"), mapOf("GATE" to "G3")), next.collections["GTDT"])
|
||||
assertEquals(4, next.stateVersion)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `new flight snapshot carries only its record scalars and full collection key set`() {
|
||||
val next = FlightStateEngine.snapshotState(
|
||||
null,
|
||||
ScheduleRecord("121", mapOf("FLNO" to "CA001"), mapOf("GTDT" to listOf(mapOf("GATE" to "G1")))),
|
||||
operationDay = day,
|
||||
keepDeleted = false,
|
||||
)
|
||||
|
||||
assertEquals(FlightState.ACTIVE, next.state)
|
||||
assertEquals(1L, next.stateVersion)
|
||||
assertEquals("CA001", next.scalars["FLNO"])
|
||||
assertFalse(next.scalars.containsKey("REMC")) // 无历史可保留
|
||||
assertEquals(10, next.collections.size) // 全键输出（§2.2）
|
||||
assertEquals(listOf(mapOf("GATE" to "G1")), next.collections["GTDT"])
|
||||
assertEquals(emptyList<Map<String, String>>(), next.collections["DELY"])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `day plan keeps DELETED state for revive conflict handling`() {
|
||||
val current = FlightSnapshot("121", day, FlightState.DELETED, 3, mapOf(), emptyMap())
|
||||
val next = FlightStateEngine.snapshotState(
|
||||
current, ScheduleRecord("121", mapOf("FLNO" to "CA001")), day, keepDeleted = true,
|
||||
)
|
||||
assertEquals(FlightState.DELETED, next.state) // §5.1 步骤 6：普通 SCHD 不恢复
|
||||
assertEquals(FlightState.DELETED, next.state) // §3.3：日计划不恢复 DELETED
|
||||
assertEquals(4, next.stateVersion)
|
||||
}
|
||||
|
||||
@@ -63,51 +105,54 @@ class FlightStateEngineTest {
|
||||
)
|
||||
|
||||
assertEquals("15DEC261900", next.scalars["ESTT"])
|
||||
assertEquals("CA001", next.scalars["FLNO"]) // 缺失 = 保留（§6.1）
|
||||
assertEquals("CA001", next.scalars["FLNO"]) // 缺失 = 保留（§3.2）
|
||||
assertEquals(listOf(mapOf("GATE" to "G1")), next.collections["GTDT"])
|
||||
assertEquals(2, next.stateVersion)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `validation rejects RECS mismatch and duplicate flid and day outside scope`() {
|
||||
fun `validation accepts valid snapshot and rejects RECS mismatch`() {
|
||||
val ok = FlightStateEngine.validateMessage(
|
||||
1, listOf(ScheduleRecord("121", mapOf("SODT" to "15DEC261723"))), day, day, opDay,
|
||||
1, listOf(ScheduleRecord("121", mapOf("SODT" to "15DEC261723"))), opDay,
|
||||
)
|
||||
assertTrue(ok is SnapshotValidation.Ok)
|
||||
|
||||
val recsMismatch = FlightStateEngine.validateMessage(
|
||||
2, listOf(ScheduleRecord("121", mapOf("SODT" to "15DEC261723"))), day, day, opDay,
|
||||
2, listOf(ScheduleRecord("121", mapOf("SODT" to "15DEC261723"))), opDay,
|
||||
) as SnapshotValidation.Invalid
|
||||
assertTrue(recsMismatch.flags.contains(SnapshotFlag.RECS_DROP))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `validation rejects duplicate flid and non computable operation day`() {
|
||||
val duplicate = FlightStateEngine.validateMessage(
|
||||
2,
|
||||
listOf(
|
||||
ScheduleRecord("121", mapOf("SODT" to "15DEC261723")),
|
||||
ScheduleRecord("121", mapOf("SODT" to "15DEC261723")),
|
||||
),
|
||||
day, day, opDay,
|
||||
opDay,
|
||||
)
|
||||
assertTrue(duplicate is SnapshotValidation.Invalid)
|
||||
|
||||
val outOfScope = FlightStateEngine.validateMessage(
|
||||
1, listOf(ScheduleRecord("121", mapOf("SODT" to "20DEC261723"))), day, day, opDay,
|
||||
val noSodt = FlightStateEngine.validateMessage(
|
||||
1, listOf(ScheduleRecord("121", emptyMap())), opDay,
|
||||
) as SnapshotValidation.Invalid
|
||||
assertTrue(outOfScope.flags.contains(SnapshotFlag.DAY_MISMATCH)) // §5.2 覆盖范围校验
|
||||
assertTrue(noSodt.flags.contains(SnapshotFlag.DAY_MISMATCH)) // 运营日不可计算 → 整包拒绝
|
||||
|
||||
val empty = FlightStateEngine.validateMessage(0, emptyList(), day, day, opDay) as SnapshotValidation.Ok
|
||||
val empty = FlightStateEngine.validateMessage(0, emptyList(), opDay) as SnapshotValidation.Ok
|
||||
assertTrue(empty.flags.contains(SnapshotFlag.EMPTY))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `validation rejects non numeric or oversized flid`() {
|
||||
val bad = FlightStateEngine.validateMessage(
|
||||
1, listOf(ScheduleRecord("ABC", mapOf("SODT" to "15DEC261723"))), day, day, opDay,
|
||||
1, listOf(ScheduleRecord("ABC", mapOf("SODT" to "15DEC261723"))), opDay,
|
||||
)
|
||||
assertTrue(bad is SnapshotValidation.Invalid)
|
||||
val tooLong = FlightStateEngine.validateMessage(
|
||||
1, listOf(ScheduleRecord("1234567890123", mapOf("SODT" to "15DEC261723"))), day, day, opDay,
|
||||
1, listOf(ScheduleRecord("1234567890123", mapOf("SODT" to "15DEC261723"))), opDay,
|
||||
)
|
||||
assertTrue(tooLong is SnapshotValidation.Invalid) // SIS §3.16.2 Number(1-12)
|
||||
assertTrue(tooLong is SnapshotValidation.Invalid) // FLID 数字型 Number(1-12)
|
||||
}
|
||||
}
|
||||
|
||||
+1
-1
@@ -9,7 +9,7 @@ import org.junit.jupiter.api.Test
|
||||
import java.sql.DriverManager
|
||||
|
||||
/**
|
||||
* Flyway 迁移引擎端到端验证（docs/flight-state.md §3.2 表职责基线）：
|
||||
* Flyway 迁移引擎端到端验证（docs/flight-state.md §2 权威模型 + design.md §2.1）：
|
||||
* 1. V1__flight_state_baseline.sql 在真实 PostgreSQL 上自动迁移成功；
|
||||
* 2. flyway_schema_history 落库且 success = true；
|
||||
* 3. 决策层/管道层/留痕层全表就绪；PIPELINE_LOCK 单行种子就位。
|
||||
|
||||
@@ -12,7 +12,7 @@ import org.junit.jupiter.api.Test
|
||||
import java.time.Instant
|
||||
|
||||
/**
|
||||
* 历史归档与物理清除（docs/flight-state.md §8.2，顺序不可颠倒）：
|
||||
* 历史归档与物理清除（docs/flight-state.md §6，顺序不可颠倒；design.md §6.2）：
|
||||
* 历史存储未接通必须删 0 条；先归档确认再物理清除；
|
||||
* 未经 FDEL 的航班在清除前补发删除事件。
|
||||
*/
|
||||
@@ -53,7 +53,7 @@ class HistorySweepJobTest {
|
||||
|
||||
val outcome = job.run(now)
|
||||
|
||||
assertEquals(0, outcome.purged) // §8.2 红线：先删历史再补当前态绝不允许
|
||||
assertEquals(0, outcome.purged) // §6 红线：历史存储未接通必须删 0 条
|
||||
assertTrue(f.findMainRow("F1") != null)
|
||||
assertEquals(0, events.rows.size)
|
||||
}
|
||||
@@ -78,14 +78,14 @@ class HistorySweepJobTest {
|
||||
assertEquals(2, outcome.selected)
|
||||
assertEquals(1, outcome.archived)
|
||||
assertEquals(1, outcome.purged)
|
||||
assertEquals(null, f.findMainRow("F1")) // §8.2 步骤 3：确认成功 → 物理删除
|
||||
assertEquals(null, f.findMainRow("F1")) // §6：归档确认成功 → 物理删除
|
||||
assertTrue(f.findMainRow("F2") != null) // 失败或不明确的保留重试（步骤 5）
|
||||
assertEquals(0, events.rows.size) // DELETED 航班清除不再发业务删除事件
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `never-fdel lifecycle purge emits tombstone before deletion`() {
|
||||
val f = seededFlight("F3", deleted = false, idleDays = 30) // ACTIVE 且超兜底窗（§8.1）
|
||||
val f = seededFlight("F3", deleted = false, idleDays = 30) // ACTIVE 且超兜底窗（§6 静默判据）
|
||||
val events = StubMsgEvents()
|
||||
val store = RecordingHistoryStore()
|
||||
val job = HistorySweepJob(
|
||||
@@ -94,7 +94,7 @@ class HistorySweepJobTest {
|
||||
|
||||
job.run(now)
|
||||
|
||||
// §7.3/§8.2 步骤 4：未经 FDEL 的航班，清除前补发一次删除事件
|
||||
// §3.3/§5/§6：未经 FDEL、由生命周期直接清除的航班，清除前补发一次删除事件
|
||||
val tombstone = events.rows.values.single { it.eventType == EventType.TOMBSTONE }
|
||||
assertEquals("F3", tombstone.partitionKey)
|
||||
assertTrue(tombstone.payloadJson.contains("\"deleted\":true"))
|
||||
|
||||
@@ -24,9 +24,9 @@ import org.junit.jupiter.api.Test
|
||||
import java.time.LocalDate
|
||||
|
||||
/**
|
||||
* FDEL（§6.2）+ ADFT（§6.3/§2.1）不变量：
|
||||
* FDEL（docs/flight-state.md §3.3）+ ADFT（§3.3/§2.1）不变量：
|
||||
* ACTIVE→DELETED 发布一次 tombstone 且明细保留；重复/迟到 FDEL 幂等不推进版本；
|
||||
* ADFT 重激活恢复 ACTIVE；新实例建立；缺失标量不误删（§2.1 保守语义）。
|
||||
* ADFT 重激活恢复 ACTIVE；新实例建立；缺失标量不误删（§3.3 保守 Set-only）。
|
||||
*/
|
||||
class FdelAndAdftProcessorTest {
|
||||
|
||||
@@ -84,7 +84,7 @@ class FdelAndAdftProcessorTest {
|
||||
val version = f.findMainRow("121")!!.stateVersion
|
||||
val result = proc.apply(head(), msg(), FlopPayload("121"))
|
||||
|
||||
assertEquals(ApplyResult.Succeeded, result) // §6.2 步骤 3：幂等成功
|
||||
assertEquals(ApplyResult.Succeeded, result) // §3.3：重复 FDEL 幂等成功
|
||||
assertEquals(version, f.findMainRow("121")!!.stateVersion)
|
||||
assertEquals(1, events.rows.values.count { it.eventType == EventType.TOMBSTONE })
|
||||
}
|
||||
@@ -97,7 +97,7 @@ class FdelAndAdftProcessorTest {
|
||||
val result = FdelProcessor(StubPipelineTx(), StubPipelineLock(), f, events, null, ObjectMapper())
|
||||
.apply(head(), msg(), FlopPayload("999"))
|
||||
|
||||
assertEquals(ApplyResult.Succeeded, result) // §6.2 步骤 4：迟到不报错
|
||||
assertEquals(ApplyResult.Succeeded, result) // §3.3：迟到/不存在幂等成功
|
||||
assertEquals(0, events.rows.size)
|
||||
}
|
||||
|
||||
@@ -118,7 +118,7 @@ class FdelAndAdftProcessorTest {
|
||||
|
||||
assertEquals(ApplyResult.Succeeded, result)
|
||||
val main = f.findMainRow("121")!!
|
||||
assertEquals(FlightState.ACTIVE, main.state) // §6.3 生命周期重激活
|
||||
assertEquals(FlightState.ACTIVE, main.state) // §3.3 生命周期重激活
|
||||
assertEquals(LocalDate.of(2026, 12, 15), main.operationDay)
|
||||
assertEquals("CA002", f.loadFullSnapshot("121")!!.scalars["FLNO"])
|
||||
}
|
||||
|
||||
@@ -32,8 +32,9 @@ import org.junit.jupiter.api.Test
|
||||
import java.time.LocalDate
|
||||
|
||||
/**
|
||||
* §5.1 applyScheduleRecords 不变量：重放幂等、完整性校验整包拒绝 DEAD(PROTOCOL)、
|
||||
* 归属日冲突不落地、DELETED 不恢复且记冲突告警、成功路径版本推进+事件+留痕+待办预登记。
|
||||
* 日计划主链路（docs/flight-state.md §3.1/§4）不变量：重放幂等、整包校验失败
|
||||
* DEAD(PROTOCOL)、归属日冲突不落地、DELETED 不恢复且记冲突告警、
|
||||
* 成功路径版本推进 + 事件 + 留痕 + 待办预登记。
|
||||
*/
|
||||
class ScheduleProcessorTest {
|
||||
|
||||
@@ -107,7 +108,7 @@ class ScheduleProcessorTest {
|
||||
assertEquals(1, log.entries.size)
|
||||
assertEquals(SnapshotResult.COMMITTED, log.entries.single().result)
|
||||
assertEquals(1, log.entries.single().upserted)
|
||||
assertEquals(1, todo.count()) // §7.2 事务内预登记
|
||||
assertEquals(1, todo.count()) // design.md §3.3/§6.1 事务内预登记
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -120,7 +121,7 @@ class ScheduleProcessorTest {
|
||||
val result = processor(proc = proc, flights = flights, log = log)
|
||||
.applyScheduleRecords(head(), message(makeBody("121" to "15DEC261723")))
|
||||
|
||||
assertEquals(ApplyResult.ReplaySkipped, result) // §5.1 步骤 2
|
||||
assertEquals(ApplyResult.ReplaySkipped, result) // §4 重放判定：幂等成功
|
||||
assertNull(flights.findMainRow("121"))
|
||||
assertEquals(SnapshotResult.REPLAY_SKIPPED, log.entries.single().result)
|
||||
}
|
||||
@@ -132,7 +133,7 @@ class ScheduleProcessorTest {
|
||||
val schdBody = ScheduleBody(recsDeclared = 2, records = makeBody("121" to "15DEC261723").records)
|
||||
|
||||
val result = processor(flights = flights, log = log).applyScheduleRecords(head(), message(schdBody)) as ApplyResult.DeadProtocol
|
||||
assertTrue(result.flags.contains(SnapshotFlag.RECS_DROP)) // §5.2 RECS = 实收数
|
||||
assertTrue(result.flags.contains(SnapshotFlag.RECS_DROP)) // §4：声明数量不符整包拒绝
|
||||
assertNull(flights.findMainRow("121"))
|
||||
assertEquals(SnapshotResult.ROLLED_BACK, log.entries.single().result)
|
||||
}
|
||||
@@ -140,7 +141,7 @@ class ScheduleProcessorTest {
|
||||
@Test
|
||||
fun `same flid across operation days is rejected whole-batch without writes`() {
|
||||
val flights = StubFlightState()
|
||||
// 预置 121 已归属 12-15；报文声称 12-16 → §5.3 第四行
|
||||
// 预置 121 已归属 12-15；报文声称 12-16 → §2.1 归属冲突
|
||||
flights.persistFullState(
|
||||
com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot(
|
||||
"121", LocalDate.of(2026, 12, 15), FlightState.ACTIVE, 1, mapOf("SODT" to "15DEC261723"), emptyMap(),
|
||||
@@ -178,7 +179,7 @@ class ScheduleProcessorTest {
|
||||
)
|
||||
|
||||
assertEquals(ApplyResult.Succeeded, result)
|
||||
assertEquals(FlightState.DELETED, flights.findMainRow("121")!!.state) // §5.1 步骤 6 不恢复
|
||||
assertEquals(FlightState.DELETED, flights.findMainRow("121")!!.state) // §3.3：日计划不恢复
|
||||
assertEquals(8L, flights.findMainRow("121")!!.stateVersion)
|
||||
assertEquals(SnapshotResult.COMMITTED, log.entries.single().result)
|
||||
assertTrue(log.entries.single().flags.contains(SnapshotFlag.SCHD_REVIVE_CONFLICT))
|
||||
@@ -193,7 +194,7 @@ class ScheduleProcessorTest {
|
||||
proc.insert(msgId)
|
||||
val p = processor(proc = proc, flights = flights, log = log, todo = todo, tx = TxRunner { true })
|
||||
|
||||
// §9：数据库/内部故障 → 异常上抛，MessageProcessor 记 FAILED(INFRA) 退避；不写任何终态
|
||||
// design.md §2.3：数据库/内部故障 → 异常上抛，MessageProcessor 记 FAILED(INFRA) 退避；不写任何终态
|
||||
org.junit.jupiter.api.Assertions.assertThrows(IllegalStateException::class.java) {
|
||||
p.applyScheduleRecords(head(), message(makeBody("121" to "15DEC261723")))
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user