diff --git a/.env.example b/.env.example index 6e2ee9a..5dc248b 100644 --- a/.env.example +++ b/.env.example @@ -19,9 +19,9 @@ MSGX_PG_PASSWORD=msgx_dev_pass # 4. Kafka（dev compose: kafka 单节点 KRaft 服务，或对接现网/测试网） MSGX_KAFKA_SERVERS=localhost:9092 -# --- 旧版 Kafka Broker 兼容配置（如现网 0.10.x / 1.x 无 INIT_PRODUCER_ID 协议时启用）--- -# 现代 Kafka（3.x/4.x，如本地 compose）：默认 acks=all, idempotence=true, max-in-flight=5 -# 旧版 Kafka（0.10.x/1.x）：取消注释以下三行以触发兼容降级，防止 UnsupportedVersionException +# --- Kafka Broker 兼容配置 --- +# D3 要求：acks=all + enable-idempotence=true + max-in-flight=1（默认值已满足） +# 旧版 Kafka（0.10.x/1.x，无 INIT_PRODUCER_ID 协议）：取消注释以下三行以触发兼容降级 # MSGX_KAFKA_ACKS=1 # MSGX_KAFKA_IDEMPOTENCE=false # MSGX_KAFKA_MAX_IN_FLIGHT=1 diff --git a/docs/invariants.md b/docs/invariants.md index 755fd34..6fa7187 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -107,14 +107,14 @@ | 缺口 | 含义 | 影响 | |---|---|---| -| `G-IGNORE` | 忽略规则（`LDM`/`REGN`/`RSTA`/`EROR`）未实现 | US-04；合法忽略报文当前按 `UNSUPPORTED` 处理 | +| ~~`G-IGNORE`~~ | ~~忽略规则（`LDM`/`REGN`/`RSTA`/`EROR`）未实现~~ | 已关闭：`IgnoreRules` 在身份绑定后精确匹配 `TYPE` 字段（codec 已大写），命中写 `SKIPPED` + 回填意图（`US-04`） | | `G-RESP-GUARD` | `RESP` 应答守卫未实现，当前与 `DNLD` 无差别进入快照写入 | 请求匹配闭环；`C-23` | | `G-REQ-TRACK` | `REQ_TRACK` 无运行时协调器：出站适配、请求编码、超时与应答匹配未实现 | US-08；`C-24` | | `G-PROC-HST` | `PROC_STATE_HST` 未建表，终态归档未落地 | US-11；归档能力 | | `G-FLOP-IDEMPOTENT` | 29 类 FLOP 幂等矩阵未补全 | `INV-20`、CLM-3 | -| `G-EVENT-RETENTION` | `MSG_EVENT` 已发送行的保留期与清理作业未实现 | outbox 有界性 | +| ~~`G-EVENT-RETENTION`~~ | ~~`MSG_EVENT` 已发送行的保留期与清理作业未实现~~ | 已关闭：`SENT_AT` 列 + 投递原子写 + `EventCleanupJob` 按 `eventRetention` 有界删除 | | `G-BACKFILL-BACKOFF` | 回填独立退避键（`backfill-backoff-ms` / `-cap-ms`）未实现，当前为代码内硬编码（取值见 reference） | 回填重试节奏 | -| `G-KAFKA-D3` | `kafka.producers.default.max-in-flight` 与 D3 要求的 1 不一致（取值见 reference） | 投递幂等前提 | +| `G-KAFKA-D3` | ~~已闭合~~：`max-in-flight` 默认收敛到 1，启动自检钉住三项联合满足 D3 | ~~投递幂等前提~~ | | `G-REPLAY-CHANNEL` | 「打标即清除」语义下的独立原文保留通道未设计 | CLM-5 | | `G-MAFL` | 主航班 `MAFL` 派生投影及主/共享原子级联未实现（规则见 `INV-21`/`INV-22`）；`MAFL` 不是 SIS/XML 入站字段 | 航班完整态；删除与重建 | | `G-SRVT-VIPF` | SIS/XML 的 `SRVT`、`VIPF` 无界集合尚未映射到持久化明细；wire/domain 只保留出现事实与原始内容，不参与合并与投递（清空语义见 `Q13`） | 航班完整态；无损字段保存 | diff --git a/docs/reference.md b/docs/reference.md index 5a62537..d429a98 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -28,7 +28,7 @@ | `msgx.pipeline.delivery-batch` | `200` | 条 | 假定 | `KAFKA:msg` 每轮每目标领取上限 | | `msgx.pipeline.delivery-drain-rounds` | `10` | 轮 | 假定 | 连取批数上限，让出循环跑 `schd` flush，防状态通知被积压饿死 | | `msgx.pipeline.autostart` | `false` | 布尔 | 安全默认 | 启动即拉起收报 / 主泵 / 投递循环；需真实仓储或 `msgx.stubs=true` | -| `msgx.pipeline.event-retention` | 目标参数（未实现） | Duration | 假定 | 已 `SENT` 事件行保留期，由维护作业清理；缺失则 outbox 无限增长 `[G-EVENT-RETENTION]` | +| `msgx.pipeline.event-retention` | `7d` | Duration | 假定 | 已 `SENT` 事件行保留期，由清理作业删除 `sent_at < now - retention` 的行；`[G-EVENT-RETENTION]` | ### 1.2 其它 `msgx.*` @@ -64,7 +64,7 @@ | `datasources.default.data-source-properties.connectTimeout` / `socketTimeout` | `3` / `30` | 假定 | 驱动级超时（秒），防网络黑洞 | | `kafka.producers.default.acks` | `all` | 契约（architecture D3） | 允许环境变量覆盖 | | `kafka.producers.default.enable-idempotence` | `true` | 契约（D3） | 允许环境变量覆盖 | -| `kafka.producers.default.max-in-flight-requests-per-connection` | `5` | **与 D3 不一致** | D3 要求 `1`；切流前必须收敛 `[G-KAFKA-D3]` | +| `kafka.producers.default.max-in-flight-requests-per-connection` | `1` | 契约（D3） | 启动自检钉住三项联合满足 D3；允许环境变量覆盖 | 环境变量清单以 `.env.example` 为准。 @@ -85,6 +85,7 @@ | `msgx.pipeline.job.last_sweep_selected` | 上一轮回填扫描选中的待办条数（扫描积压） | 持续顶到批次上限 → 扫描吃不消 | | `msgx.pipeline.codec.srvt_seen.total` | 入站记录中出现 `SRVT` 段的条数（尚未落明细表，`[G-SRVT-VIPF]`） | > 0 → 真实流量确有该段，按真实报文定案 `Q13` | | `msgx.pipeline.codec.vipf_seen.total` | 入站记录中出现 `VIPF` 段的条数（尚未落明细表，`[G-SRVT-VIPF]`） | 同上 | +| `msgx.pipeline.processing.ignored.total` | 命中 US-04 忽略清单的报文条数 | 增长是正常流量；归零反而需确认配置是否丢失 | 取数规则：统一走 `BacklogSnapshotProvider`（`PARAM:msgx.health.backlog-cache-ttl-ms`），`/health` 与 `/metrics` 共用同一快照——`backlog()` 是 `PROC_STATE` 的全表聚合，不能被高频抓取打穿；**无法取数上报 `NaN`，无可比记录的年龄/滞后类仪表上报 `-1`，都不伪造 0**。日志出口故障不得阻塞业务线程。进程内计数类指标不经快照，重启归零。 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/config/KafkaD3Check.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/config/KafkaD3Check.kt new file mode 100644 index 0000000..f4d500d --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/config/KafkaD3Check.kt @@ -0,0 +1,39 @@ +package com.gzzn.omms.msgexchange.config + +import io.micronaut.context.annotation.Context +import io.micronaut.context.annotation.Value +import jakarta.inject.Singleton + +/** + * D3 启动自检：Kafka 生产者的 acks / enable-idempotence / max-in-flight 三项 + * 必须联合满足幂等生产前提。不满足时拒绝启动——静默放行会让幂等保证在运行时失效。 + * + * 独立于 `PipelineLifecycle`（受 autostart 开关控制），无条件执行。 + */ +@Singleton +@Context +class KafkaD3Check( + @Value("\${kafka.producers.default.acks}") acks: String, + @Value("\${kafka.producers.default.enable-idempotence}") idempotence: String, + @Value("\${kafka.producers.default.max-in-flight-requests-per-connection}") maxInFlight: String, +) { + private val acks: String = acks + private val idempotence: Boolean = idempotence.toBoolean() + private val maxInFlight: Int = maxInFlight.toInt() + + fun validate() { + require(acks == "all" || acks == "-1") { + "D3 violation: kafka.producers.default.acks must be 'all', got '$acks'" + } + require(idempotence) { + "D3 violation: kafka.producers.default.enable-idempotence must be true, got $idempotence" + } + require(maxInFlight == 1) { + "D3 violation: kafka.producers.default.max-in-flight-requests-per-connection must be 1, got $maxInFlight" + } + } + + init { + validate() + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt index 2195049..d0022a3 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt @@ -82,6 +82,13 @@ class PipelineProps { */ var autostart: Boolean = false + /** + * MSG_EVENT 已确认投递行（`STATE='SENT'`）的保留期。 + * 清理作业删除 `SENT_AT < now - eventRetention` 的行； + * 取值应大于最长可能的投递重试周期，避免误删刚发出的事件。 + */ + var eventRetention: Duration = Duration.ofDays(7) + /** N28：attempt ≤ 0（如 FAILED 未递增 attempts 的行）不得抛异常，取下界=首档退避。 */ fun backoffFor(attempt: Int): Long { val index = (attempt - 1).coerceAtLeast(0) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt index e634f76..35e1001 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt @@ -130,7 +130,7 @@ class Dispatcher( // 未知目标不能在"未发送"的情况下被标记为已发；记账后停止本轮。 else -> error("unknown delivery target: $target") } - msgEvents.markSent(eventId) + msgEvents.markSent(eventId, scheduler.now()) true } catch (ex: Exception) { retryOrDead(e, ex.message ?: ex.javaClass.simpleName) @@ -158,7 +158,7 @@ class Dispatcher( EventType.UPSERT -> port.sendKafkaSchd("schd", e.partitionKey, e.payloadJson) } // 条件确认：读取时刻的代次（EVENT_ID + STATE_VERSION）被新写入覆盖时不标记，留待下一轮重发。 - e.eventId?.let { msgEvents.markSentIfVersion(it, e.stateVersion) } + e.eventId?.let { msgEvents.markSentIfVersion(it, e.stateVersion, scheduler.now()) } } catch (ex: Exception) { failures.add(e) } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/MsgEvent.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/MsgEvent.kt index 8a4d89b..871a0b8 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/MsgEvent.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/MsgEvent.kt @@ -29,4 +29,6 @@ data class MsgEvent( val errorClass: ErrorClass? = null, val lastError: String? = null, val createdAt: java.time.Instant, + /** 投递确认时刻（`STATE='SENT'` 时写入）；清理作业按此字段判断保留期。 */ + val sentAt: java.time.Instant? = null, ) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineCounters.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineCounters.kt index 6010f9c..d58873c 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineCounters.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineCounters.kt @@ -15,6 +15,7 @@ import java.util.concurrent.atomic.AtomicLong class PipelineCounters { private val srvtSeen = AtomicLong(0) private val vipfSeen = AtomicLong(0) + private val ignored = AtomicLong(0) /** * 入站记录里出现 `SRVT`/`VIPF` 段的条数（`[G-SRVT-VIPF]`）。 @@ -30,4 +31,9 @@ class PipelineCounters { fun srvtSeenCount(): Long = srvtSeen.get() fun vipfSeenCount(): Long = vipfSeen.get() + + /** US-04 忽略清单命中计数。 */ + fun ignoredAdd(count: Int = 1) { ignored.addAndGet(count.toLong()) } + + fun ignoredCount(): Long = ignored.get() } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetrics.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetrics.kt index 35d0261..c5b05c5 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetrics.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetrics.kt @@ -28,6 +28,7 @@ import java.time.Duration * - `msgx.pipeline.job.last_sweep_selected`：上一轮回填扫描选中的待办条数（扫描积压） * - `msgx.pipeline.codec.srvt_seen.total` / `msgx.pipeline.codec.vipf_seen.total`：入站记录里出现 * `SRVT`/`VIPF` 段的条数（尚未落明细表，`[G-SRVT-VIPF]`；> 0 表示真实流量确有该段） + * - `msgx.pipeline.processing.ignored.total`：命中 US-04 忽略清单的报文条数 * * 取数统一走 [BacklogSnapshotProvider]（30 秒 TTL），因此指标抓取不会打穿数据库。 * 无法取数时以 `NaN` 上报（Micrometer 的惯例表示"本次无值"），而不是伪造 0。 @@ -101,6 +102,10 @@ class PipelineMetrics( Gauge.builder("msgx.pipeline.codec.vipf_seen.total", counters) { it.vipfSeenCount().toDouble() } .strongReference(true) .register(registry) + + Gauge.builder("msgx.pipeline.processing.ignored.total", counters) { it.ignoredCount().toDouble() } + .strongReference(true) + .register(registry) } private fun backlogGauge(name: String, value: (com.gzzn.omms.msgexchange.infra.persistence.Backlog) -> Double) { diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt index 25420e2..fcc1531 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt @@ -170,7 +170,7 @@ interface MsgEventRepository { /** 领取待发的整态事件：写入侧已保证每个 `FLID` 只有一行，读端不再去重合并。 */ fun mergePendingSchd(now: Instant, limit: Int): List - fun markSent(eventId: Long) + fun markSent(eventId: Long, now: Instant) /** * 条件确认：只在该行仍是本批读取到的代次（`EVENT_ID` + `STATE_VERSION`）且尚未发出时标记 `SENT`。 @@ -178,13 +178,20 @@ interface MsgEventRepository { * * @return 受影响行数（0 或 1） */ - fun markSentIfVersion(eventId: Long, stateVersion: Long): Int + fun markSentIfVersion(eventId: Long, stateVersion: Long, now: Instant): Int - fun markAllSent(eventIds: List) + fun markAllSent(eventIds: List, now: Instant) fun scheduleRetry(eventId: Long, nextAttemptAt: Instant, attempts: Int) fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String, attempts: Int? = null) + + /** + * 删除已确认投递且超过保留期的事件行：`STATE='SENT' AND SENT_AT < cutoff`。 + * 带 `STATE='SENT'` 条件，避免与 upsertSchd 的并发写入冲突。 + * @return 实际删除的行数 + */ + fun deleteExpiredSent(cutoff: Instant, limit: Int): Int } /** diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt index aa92773..1d45a2e 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt @@ -473,6 +473,7 @@ class JdbcMsgEventRepository( next_attempt_at = NULL, error_class = NULL, last_error = NULL, + sent_at = NULL, created_at = EXCLUDED.created_at WHERE EXCLUDED.state_version > msg_event.state_version OR (EXCLUDED.state_version = msg_event.state_version AND EXCLUDED.event_type = 'TOMBSTONE') @@ -519,22 +520,26 @@ class JdbcMsgEventRepository( ::mapEvent, ) - override fun markSent(eventId: Long) { - ds.update("UPDATE msg_event SET state = 'SENT' WHERE event_id = ?", { ps -> ps.setLong(1, eventId) }) + override fun markSent(eventId: Long, now: Instant) { + ds.update( + "UPDATE msg_event SET state = 'SENT', sent_at = ? WHERE event_id = ?", + { ps -> ps.setTimestamp(1, now.toSqlTimestamp()); ps.setLong(2, eventId) }, + ) } /** 条件确认：仅当行仍是本批读到的代次且尚未发出时才标记，被新代次覆盖则影响 0 行。 */ - override fun markSentIfVersion(eventId: Long, stateVersion: Long): Int = + override fun markSentIfVersion(eventId: Long, stateVersion: Long, now: Instant): Int = ds.update( - "UPDATE msg_event SET state = 'SENT' WHERE event_id = ? AND state_version = ? AND state = 'PENDING'", - { ps -> ps.setLong(1, eventId); ps.setLong(2, stateVersion) }, + "UPDATE msg_event SET state = 'SENT', sent_at = ? WHERE event_id = ? AND state_version = ? AND state = 'PENDING'", + { ps -> ps.setTimestamp(1, now.toSqlTimestamp()); ps.setLong(2, eventId); ps.setLong(3, stateVersion) }, ) - override fun markAllSent(eventIds: List) { + override fun markAllSent(eventIds: List, now: Instant) { if (eventIds.isEmpty()) return val placeholders = eventIds.joinToString(",") { "?" } - ds.update("UPDATE msg_event SET state = 'SENT' WHERE event_id IN ($placeholders)") { ps -> - eventIds.forEachIndexed { i, id -> ps.setLong(i + 1, id) } + ds.update("UPDATE msg_event SET state = 'SENT', sent_at = ? WHERE event_id IN ($placeholders)") { ps -> + ps.setTimestamp(1, now.toSqlTimestamp()) + eventIds.forEachIndexed { i, id -> ps.setLong(i + 2, id) } } } @@ -557,6 +562,21 @@ class JdbcMsgEventRepository( ) } + override fun deleteExpiredSent(cutoff: Instant, limit: Int): Int = + ds.update( + """ + DELETE FROM msg_event WHERE event_id IN ( + SELECT event_id FROM msg_event + WHERE state = 'SENT' AND sent_at IS NOT NULL AND sent_at < ? + ORDER BY event_id LIMIT ? + ) + """.trimIndent(), + { ps -> + ps.setTimestamp(1, cutoff.toSqlTimestamp()) + ps.setInt(2, limit) + }, + ) + private fun mapEvent(rs: ResultSet) = MsgEvent( eventId = rs.getLong("event_id"), target = rs.getString("target"), @@ -570,6 +590,7 @@ class JdbcMsgEventRepository( errorClass = rs.getString("error_class")?.let(ErrorClass::valueOf), lastError = rs.getString("last_error"), createdAt = rs.getInstant("created_at") ?: clock.instant(), + sentAt = rs.getInstant("sent_at"), ) } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt index e6dbb8c..8f2b56c 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt @@ -262,7 +262,7 @@ class StubMsgEvents : MsgEventRepository { val id = ids.incrementAndGet() rows[id] = e.copy( eventId = id, state = EventStatus.PENDING, attempts = 0, nextAttemptAt = null, - errorClass = null, lastError = null, + errorClass = null, lastError = null, sentAt = null, ) return id } @@ -278,19 +278,19 @@ class StubMsgEvents : MsgEventRepository { .sortedBy { it.eventId!! } .take(limit) - override fun markSent(eventId: Long) { - rows[eventId] = (rows[eventId] ?: return).copy(state = EventStatus.SENT) + override fun markSent(eventId: Long, now: Instant) { + rows[eventId] = (rows[eventId] ?: return).copy(state = EventStatus.SENT, sentAt = now) } - override fun markSentIfVersion(eventId: Long, stateVersion: Long): Int { + override fun markSentIfVersion(eventId: Long, stateVersion: Long, now: Instant): Int { val row = rows[eventId] ?: return 0 if (row.state != EventStatus.PENDING || row.stateVersion != stateVersion) return 0 - rows[eventId] = row.copy(state = EventStatus.SENT) + rows[eventId] = row.copy(state = EventStatus.SENT, sentAt = now) return 1 } - override fun markAllSent(eventIds: List) { - eventIds.forEach { markSent(it) } + override fun markAllSent(eventIds: List, now: Instant) { + eventIds.forEach { markSent(it, now) } } override fun scheduleRetry(eventId: Long, nextAttemptAt: Instant, attempts: Int) { @@ -303,6 +303,16 @@ class StubMsgEvents : MsgEventRepository { attempts = attempts ?: rows[eventId]?.attempts ?: 0, ) } + + override fun deleteExpiredSent(cutoff: Instant, limit: Int): Int { + val ids = rows.values + .filter { it.state == EventStatus.SENT && it.sentAt != null && it.sentAt < cutoff } + .sortedBy { it.eventId!! } + .take(limit) + .map { it.eventId!! } + ids.forEach { rows.remove(it) } + return ids.size + } } @Singleton diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/EventCleanupJob.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/EventCleanupJob.kt new file mode 100644 index 0000000..5a3824f --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/EventCleanupJob.kt @@ -0,0 +1,30 @@ +package com.gzzn.omms.msgexchange.jobs + +import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository +import jakarta.inject.Singleton +import java.time.Instant + +/** + * MSG_EVENT 有界清理：删除已确认投递（`STATE='SENT'`）且 `SENT_AT` 超过保留期的行。 + * + * 每轮 JobRunner tick 跑一次，删除量有上限（`CLEANUP_BATCH`），避免单次长事务。 + * 只删 `STATE='SENT'` 的行——upsertSchd 写入的待发行状态是 `PENDING`，不会被误删。 + */ +@Singleton +class EventCleanupJob( + private val msgEvents: MsgEventRepository, + private val props: PipelineProps, +) { + data class CleanupOutcome(val deleted: Int) + + fun run(now: Instant): CleanupOutcome { + val cutoff = now.minus(props.pipeline.eventRetention) + val deleted = msgEvents.deleteExpiredSent(cutoff, CLEANUP_BATCH) + return CleanupOutcome(deleted) + } + + private companion object { + const val CLEANUP_BATCH = 500 + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt index 762cd62..3135be9 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt @@ -1,6 +1,7 @@ package com.gzzn.omms.msgexchange.jobs import com.gzzn.omms.msgexchange.config.OperationDayProps +import com.gzzn.omms.msgexchange.config.PipelineProps import com.gzzn.omms.msgexchange.infra.metrics.JobActivity import com.gzzn.omms.msgexchange.processing.BackfillService import jakarta.inject.Singleton @@ -23,8 +24,10 @@ import java.time.ZoneId class JobRunner( private val backfill: BackfillService, private val historySweep: HistorySweepJob, + private val eventCleanup: EventCleanupJob, private val clock: Clock, private val activity: JobActivity, + private val props: PipelineProps, operationDayProps: OperationDayProps, ) { private val log = org.slf4j.LoggerFactory.getLogger(JobRunner::class.java) @@ -43,7 +46,7 @@ class JobRunner( running = true activity.started() thread = Thread.ofPlatform().name("msgx-jobs").daemon(true).start { loop() } - log.info("job runner started (backfill {}s, history daily 03:30 {})", TICK_PERIOD.seconds, zone) + log.info("job runner started (backfill {}s, history daily 03:30 {}, event cleanup {})", TICK_PERIOD.seconds, zone, props.pipeline.eventRetention) } fun stop() { @@ -75,6 +78,10 @@ class JobRunner( try { val selected = backfill.sweep(startedAt) maybeHistorySweep() + val cleanupOutcome = eventCleanup.run(startedAt) + if (cleanupOutcome.deleted > 0) { + log.info("event cleanup: deleted {} expired SENT rows", cleanupOutcome.deleted) + } activity.tickFinished(startedAt, (System.nanoTime() - startedNanos) / 1_000_000, selected) } catch (e: InterruptedException) { throw e diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreRules.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreRules.kt new file mode 100644 index 0000000..45ebe5d --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreRules.kt @@ -0,0 +1,24 @@ +package com.gzzn.omms.msgexchange.processing + +/** + * US-04 忽略清单：基线规则为 `LDM-*`、`REGN-*`、`RSTA-*`、`EROR-*`。 + * + * 匹配按 `TYPE` 前缀做（`TYPE-*` 表示该 TYPE 下所有 STYP 均忽略）； + * `MetaFields.type` 已由 codec 统一大写，这里直接用大写常量比较。 + * 返回命中的规则标签（如 `LDM-*`），未命中返回 null。 + */ +object IgnoreRules { + private val rules: List> = listOf( + "LDM" to "LDM-*", + "REGN" to "REGN-*", + "RSTA" to "RSTA-*", + "EROR" to "EROR-*", + ) + + fun match(type: String): String? { + for ((prefix, label) in rules) { + if (type == prefix) return label + } + return null + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt index 7056af6..f4ff6fb 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt @@ -211,6 +211,15 @@ class MessageProcessor( } } + // US-04：忽略清单命中 → SKIPPED + 回填意图，不取 PIPELINE_LOCK、不写航班表、不创建 MSG_EVENT + val ignoreRule = IgnoreRules.match(decoded.meta.type) + if (ignoreRule != null) { + log.info("ignored:{} -> SKIPPED msgId={}", ignoreRule, head.msgId) + counters.ignoredAdd() + procState.markTerminal(head.msgId, ProcStatus.SKIPPED, lastError = "ignored:$ignoreRule", now = clock.instant()) + return + } + // 按报文类型分派：日计划走 SCHD，其余走 FLOP / FDEL / ADFT；报文缺载荷直接判为非法报文的死信 val result: ApplyResult = when (val kind = decoded.kind) { is MsgKind.Schd -> { diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index d03d200..41938c4 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -96,11 +96,12 @@ kafka: servers: ${MSGX_KAFKA_SERVERS} producers: default: # U03/R09：Micronaut Kafka 按具名 producer 解析，须有 default 层 - # 默认 acks=all + 幂等（现代 Kafka 3.x+）；若对接旧版 Broker（如现网 0.10.x/1.x，无 INIT_PRODUCER_ID 协议）， - # 经 MSGX_KAFKA_ACKS=1 与 MSGX_KAFKA_IDEMPOTENCE=false 降级兼容。 + # D3：acks=all + 幂等 + max-in-flight=1 联合保证幂等生产； + # 若对接旧版 Broker（如现网 0.10.x/1.x，无 INIT_PRODUCER_ID 协议）， + # 三项须同时降级为 MSGX_KAFKA_ACKS=1 / MSGX_KAFKA_IDEMPOTENCE=false / MSGX_KAFKA_MAX_IN_FLIGHT=1。 acks: ${MSGX_KAFKA_ACKS:all} enable-idempotence: ${MSGX_KAFKA_IDEMPOTENCE:true} - max-in-flight-requests-per-connection: ${MSGX_KAFKA_MAX_IN_FLIGHT:5} + max-in-flight-requests-per-connection: ${MSGX_KAFKA_MAX_IN_FLIGHT:1} eureka: client: registration: diff --git a/src/main/resources/db/migration/V10__event_sent_at.sql b/src/main/resources/db/migration/V10__event_sent_at.sql new file mode 100644 index 0000000..cf7cfbc --- /dev/null +++ b/src/main/resources/db/migration/V10__event_sent_at.sql @@ -0,0 +1,13 @@ +-- ===================================================================== +-- V10：MSG_EVENT 增加 SENT_AT，为有界清理提供时间基准（G-EVENT-RETENTION） +-- +-- 所有投递确认操作在设置 STATE='SENT' 的同一条 UPDATE 中写入 SENT_AT； +-- 清理作业只删除 STATE='SENT' AND SENT_AT IS NOT NULL AND SENT_AT < now - retention。 +-- +-- 存量 SENT 行的 SENT_AT 设为迁移时刻 CURRENT_TIMESTAMP， +-- 避免迁移完成后它们立即过期被清理掉。 +-- ===================================================================== + +ALTER TABLE msg_event ADD COLUMN sent_at TIMESTAMP(6) WITH TIME ZONE; + +UPDATE msg_event SET sent_at = CURRENT_TIMESTAMP WHERE state = 'SENT'; diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/config/KafkaD3CheckTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/config/KafkaD3CheckTest.kt new file mode 100644 index 0000000..8d6f54e --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/config/KafkaD3CheckTest.kt @@ -0,0 +1,39 @@ +package com.gzzn.omms.msgexchange.config + +import org.junit.jupiter.api.Assertions.assertDoesNotThrow +import org.junit.jupiter.api.Assertions.assertThrows +import org.junit.jupiter.api.Test + +class KafkaD3CheckTest { + + @Test + fun `default config satisfies D3`() { + assertDoesNotThrow { KafkaD3Check(acks = "all", idempotence = "true", maxInFlight = "1") } + } + + @Test + fun `acks=-1 also satisfies D3`() { + assertDoesNotThrow { KafkaD3Check(acks = "-1", idempotence = "true", maxInFlight = "1") } + } + + @Test + fun `max-in-flight not 1 rejects`() { + assertThrows(IllegalArgumentException::class.java) { + KafkaD3Check(acks = "all", idempotence = "true", maxInFlight = "5") + } + } + + @Test + fun `idempotence false rejects`() { + assertThrows(IllegalArgumentException::class.java) { + KafkaD3Check(acks = "all", idempotence = "false", maxInFlight = "1") + } + } + + @Test + fun `acks not all rejects`() { + assertThrows(IllegalArgumentException::class.java) { + KafkaD3Check(acks = "1", idempotence = "true", maxInFlight = "1") + } + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcMsgEventUpsertPgTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcMsgEventUpsertPgTest.kt index fdde8b5..bb7b6c3 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcMsgEventUpsertPgTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcMsgEventUpsertPgTest.kt @@ -45,8 +45,9 @@ class JdbcMsgEventUpsertPgTest { assertEquals(EventType.TOMBSTONE, pendingSchd(repo, key).single().eventType) // 条件确认：旧代次影响 0 行，当前代次标记成功 - assertEquals(0, repo.markSentIfVersion(firstId, 1)) - assertEquals(1, repo.markSentIfVersion(tombstoneId, 2)) + val now = java.time.Instant.parse("2025-01-01T00:00:00Z") + assertEquals(0, repo.markSentIfVersion(firstId, 1, now)) + assertEquals(1, repo.markSentIfVersion(tombstoneId, 2, now)) assertEquals(0, pendingSchd(repo, key).size) } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/EventCleanupJobTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/EventCleanupJobTest.kt new file mode 100644 index 0000000..1915f6e --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/EventCleanupJobTest.kt @@ -0,0 +1,78 @@ +package com.gzzn.omms.msgexchange.jobs + +import com.gzzn.omms.msgexchange.MutableClock +import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.domain.EventStatus +import com.gzzn.omms.msgexchange.domain.MsgEvent +import com.gzzn.omms.msgexchange.domain.Targets +import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import java.time.Duration +import java.time.Instant + +/** + * 清理作业只删 `STATE='SENT'` 且 `SENT_AT` 超过保留期的行； + * 未确认投递（`PENDING`）和未超期的 SENT 行不受影响。 + */ +class EventCleanupJobTest { + + private val clock = MutableClock(MutableClock.BASE) + private val props = PipelineProps().apply { pipeline.eventRetention = Duration.ofDays(7) } + + private fun sentEvent(sentAt: Instant): MsgEvent = MsgEvent( + target = Targets.KAFKA_MSG, + partitionKey = "pk", + payloadJson = "{}", + createdAt = MutableClock.BASE, + sentAt = sentAt, + ) + + @Test + fun `deletes SENT rows older than retention`() { + val repo = StubMsgEvents() + val old = repo.insertAll(listOf(sentEvent(clock.instant().minus(Duration.ofDays(8))))).single() + repo.markSent(old, clock.instant().minus(Duration.ofDays(8))) + val recent = repo.insertAll(listOf(sentEvent(clock.instant().minus(Duration.ofDays(2))))).single() + repo.markSent(recent, clock.instant().minus(Duration.ofDays(2))) + + val deleted = EventCleanupJob(repo, props).run(clock.instant()).deleted + + assertEquals(1, deleted) + assertEquals(EventStatus.SENT, repo.rows[recent]!!.state) + assertTrue(old !in repo.rows) + } + + @Test + fun `does not delete PENDING rows regardless of age`() { + val repo = StubMsgEvents() + val pending = MsgEvent( + target = Targets.KAFKA_MSG, + partitionKey = "pk", + payloadJson = "{}", + createdAt = MutableClock.BASE, + ) + val id = repo.insertAll(listOf(pending)).single() + + val deleted = EventCleanupJob(repo, props).run(clock.instant()).deleted + + assertEquals(0, deleted) + assertEquals(EventStatus.PENDING, repo.rows[id]!!.state) + } + + @Test + fun `cleans up multiple expired SENT rows in one run`() { + val repo = StubMsgEvents() + val oldSentAt = clock.instant().minus(Duration.ofDays(30)) + repeat(5) { + val id = repo.insertAll(listOf(sentEvent(oldSentAt))).single() + repo.markSent(id, oldSentAt) + } + val smallBatchProps = PipelineProps().apply { pipeline.eventRetention = Duration.ofDays(7) } + + val deleted = EventCleanupJob(repo, smallBatchProps).run(clock.instant()).deleted + + assertEquals(5, deleted) + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt index dda6a71..658ebd2 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt @@ -38,7 +38,7 @@ class JobRunnerTest { private fun runner(proc: ProcStateRepository, activity: JobActivity) = JobRunner( BackfillService(proc, StubInbox(), MailboxProps(), props, clock, MessageLifecycleGate()), - historySweep(), clock, activity, OperationDayProps(), + historySweep(), EventCleanupJob(StubMsgEvents(), props), clock, activity, props, OperationDayProps(), ) @Test diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt new file mode 100644 index 0000000..602f7e3 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt @@ -0,0 +1,205 @@ +package com.gzzn.omms.msgexchange.processing + +import com.gzzn.omms.msgexchange.codec.DecodeResult +import com.gzzn.omms.msgexchange.codec.XmlCodec +import com.gzzn.omms.msgexchange.config.OperationDayProps +import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.domain.DecodedMessage +import com.gzzn.omms.msgexchange.domain.MetaFields +import com.gzzn.omms.msgexchange.domain.MsgKind +import com.gzzn.omms.msgexchange.domain.ProcState +import com.gzzn.omms.msgexchange.domain.ProcStatus +import com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters +import com.gzzn.omms.msgexchange.infra.persistence.InboxCursorRepository +import com.gzzn.omms.msgexchange.infra.persistence.PipelineLockRepository +import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager +import com.gzzn.omms.msgexchange.infra.retry.ProcFailure +import com.gzzn.omms.msgexchange.infra.retry.FailureScheduler +import com.gzzn.omms.msgexchange.infra.stub.StubFlightState +import com.gzzn.omms.msgexchange.infra.stub.StubInbox +import com.gzzn.omms.msgexchange.infra.stub.StubInboxCursor +import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents +import com.gzzn.omms.msgexchange.infra.stub.StubProcState +import com.gzzn.omms.msgexchange.infra.stub.StubSnapshotLog +import com.fasterxml.jackson.databind.ObjectMapper +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Test +import java.time.Clock +import java.time.Instant + +/** + * 忽略类报文（US-04）：命中忽略清单的报文先绑定身份再写 SKIPPED， + * 不产生航班或 outbox 副作用；未命中的合法类型继续按 UNSUPPORTED 处理。 + */ +class IgnoreBranchTest { + + private val msgId = 99L + + private fun head() = ProcState(msgId, ProcStatus.PENDING, updatedAt = Instant.EPOCH) + + private fun decoded(type: String, styp: String) = DecodedMessage( + meta = MetaFields("AODB", type, styp, 200L, 1L), + kind = MsgKind.Unsupported("$type-$styp"), + rawXml = "", + ) + + private fun codecReturning(msg: DecodedMessage) = object : XmlCodec { + override fun decode(rawXml: String): DecodeResult = DecodeResult.Ok(msg) + override fun encodeRqrd(kind: String, rangeJson: String): String = "" + } + + private val noopTx = object : PipelineTransactionManager { + override fun inTransaction(block: () -> T): T = block() + } + private val noopLock = object : PipelineLockRepository { override fun lock() = Unit } + + private fun processor( + proc: StubProcState = StubProcState(), + inbox: StubInbox = StubInbox(), + counters: PipelineCounters = PipelineCounters(), + codec: XmlCodec, + ): MessageProcessor { + val cursor = StubInboxCursor() + cursor.cursor = InboxCursorRepository.Cursor(committedUpTo = Long.MAX_VALUE) + val clock = Clock.systemUTC() + val props = PipelineProps() + val opDay = OperationDayProps().apply { zone = "Asia/Shanghai"; cutoffHour = 0 } + val flights = StubFlightState() + val events = StubMsgEvents() + val log = StubSnapshotLog() + val procFailure = ProcFailure(proc, FailureScheduler(props, clock)) + return MessageProcessor( + inbox = inbox, + procState = proc, + codec = codec, + scheduleProcessor = ScheduleProcessor(noopTx, noopLock, proc, flights, events, log, opDay, ObjectMapper(), clock), + flopProcessor = FlopProcessor(noopTx, noopLock, flights, events, proc, ObjectMapper(), clock), + fdelProcessor = FdelProcessor(noopTx, noopLock, flights, events, proc, ObjectMapper(), clock), + adftProcessor = AdftProcessor(noopTx, noopLock, flights, events, proc, opDay, ObjectMapper(), clock), + procFailure = procFailure, + props = props, + clock = clock, + operationDayProps = opDay, + counters = counters, + ) + } + + @Test + fun `LDM subtype hits ignore rule and writes SKIPPED with reason`() { + val proc = StubProcState() + proc.insertIfAbsent(msgId, null) + val inbox = StubInbox() + inbox.raws[msgId] = "" + val msg = decoded("LDM", "ADQ") + val p = processor(proc = proc, inbox = inbox, codec = codecReturning(msg)) + + p.processOne(head()) + + val row = proc.find(msgId)!! + assertEquals(ProcStatus.SKIPPED, row.state) + assertEquals("ignored:LDM-*", row.lastError) + assertNotNull(row.backfillNextAt) + } + + @Test + fun `EROR subtype hits ignore rule`() { + val proc = StubProcState() + proc.insertIfAbsent(msgId, null) + val inbox = StubInbox() + inbox.raws[msgId] = "" + val msg = decoded("EROR", "GEN") + val p = processor(proc = proc, inbox = inbox, codec = codecReturning(msg)) + + p.processOne(head()) + + assertEquals(ProcStatus.SKIPPED, proc.find(msgId)!!.state) + assertEquals("ignored:EROR-*", proc.find(msgId)!!.lastError) + } + + @Test + fun `REGN and RSTA both hit ignore rules`() { + for ((type, rule) in listOf("REGN" to "REGN-*", "RSTA" to "RSTA-*")) { + val proc = StubProcState() + val id = msgId + type.hashCode().toLong() + proc.insertIfAbsent(id, null) + val inbox = StubInbox() + inbox.raws[id] = "" + val msg = DecodedMessage( + meta = MetaFields("AODB", type, "X", 300L, 1L), + kind = MsgKind.Unsupported("$type-X"), + rawXml = "", + ) + val p = processor(proc = proc, inbox = inbox, codec = codecReturning(msg)) + + p.processOne(ProcState(id, ProcStatus.PENDING, updatedAt = Instant.EPOCH)) + + assertEquals(ProcStatus.SKIPPED, proc.find(id)!!.state) + assertEquals("ignored:$rule", proc.find(id)!!.lastError) + } + } + + @Test + fun `ignore binds identity before skipping`() { + val proc = StubProcState() + proc.insertIfAbsent(msgId, null) + val inbox = StubInbox() + inbox.raws[msgId] = "" + val msg = decoded("LDM", "ADQ") + val p = processor(proc = proc, inbox = inbox, codec = codecReturning(msg)) + + p.processOne(head()) + + assertNotNull(proc.find(msgId)!!.identityKey) + } + + @Test + fun `duplicate identity takes precedence over ignore`() { + val ownerMsgId = msgId + 1 + val proc = StubProcState() + proc.insertIfAbsent(msgId, null) + proc.insertIfAbsent(ownerMsgId, null) + proc.tryBindIdentity(ownerMsgId, "AODB|LDM|ADQ|200") + val inbox = StubInbox() + inbox.raws[msgId] = "" + val msg = decoded("LDM", "ADQ") + val p = processor(proc = proc, inbox = inbox, codec = codecReturning(msg)) + + p.processOne(head()) + + val row = proc.find(msgId)!! + assertEquals(ProcStatus.SKIPPED, row.state) + assertEquals("duplicate-of:$ownerMsgId", row.lastError) + } + + @Test + fun `non-ignored unsupported type continues as UNSUPPORTED`() { + val proc = StubProcState() + proc.insertIfAbsent(msgId, null) + val inbox = StubInbox() + inbox.raws[msgId] = "" + val msg = decoded("XYZZ", "TEST") + val counters = PipelineCounters() + val p = processor(proc = proc, inbox = inbox, counters = counters, codec = codecReturning(msg)) + + p.processOne(head()) + + assertEquals(ProcStatus.FAILED, proc.find(msgId)!!.state) + assertEquals(0L, counters.ignoredCount()) + } + + @Test + fun `ignore increments counter`() { + val proc = StubProcState() + proc.insertIfAbsent(msgId, null) + val inbox = StubInbox() + inbox.raws[msgId] = "" + val msg = decoded("LDM", "ADQ") + val counters = PipelineCounters() + val p = processor(proc = proc, inbox = inbox, counters = counters, codec = codecReturning(msg)) + + p.processOne(head()) + + assertEquals(1L, counters.ignoredCount()) + } +}