feat(acm2): 闭合 G-KAFKA-D3 / G-IGNORE / G-EVENT-RETENTION 三个缺口

- G-KAFKA-D3：max-in-flight 默认收敛到 1，KafkaD3Check 启动自检钉住三项联合满足 D3
- G-IGNORE：IgnoreRules 在身份绑定后精确匹配 TYPE 字段，命中写 SKIPPED + 回填意图
- G-EVENT-RETENTION：V10 迁移加 SENT_AT 列，投递原子写，EventCleanupJob 有界清理过期 SENT 行

.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

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`） | 航班完整态；无损字段保存 |

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**。日志出口故障不得阻塞业务线程。进程内计数类指标不经快照，重启归零。
 

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()
+    }
+}

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)

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)
             }

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,
 )

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()
 }

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) {

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<MsgEvent>
 
-    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<Long>)
+    fun markAllSent(eventIds: List<Long>, 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
 }
 
 /**

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<Long>) {
+    override fun markAllSent(eventIds: List<Long>, 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"),
     )
 }
 

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<Long>) {
-        eventIds.forEach { markSent(it) }
+    override fun markAllSent(eventIds: List<Long>, 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

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
+    }
+}

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

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<Pair<String, String>> = 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
+    }
+}

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 -> {

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:

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';

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")
+        }
+    }
+}

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)
     }
 

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)
+    }
+}

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

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 = "<MSG/>",
+    )
+
+    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 <T> 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] = "<RAW/>"
+        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] = "<RAW/>"
+        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] = "<RAW/>"
+            val msg = DecodedMessage(
+                meta = MetaFields("AODB", type, "X", 300L, 1L),
+                kind = MsgKind.Unsupported("$type-X"),
+                rawXml = "<MSG/>",
+            )
+            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] = "<RAW/>"
+        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] = "<RAW/>"
+        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] = "<RAW/>"
+        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] = "<RAW/>"
+        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())
+    }
+}