feat(runtime): U07 生命周期装配 + stub 装配冒烟（ACM2-10 U07/U01 遗留）

- PipelineLifecycle：ServerStartupEvent 后在专用单线程（msgx-pump/msgx-dispatcher，daemon）拉起 Pump/Dispatcher 循环；停机 requestStop+interrupt+join；@Requires(msgx.pipeline.autostart=true) - PipelineProps.pipeline.autostart（默认 false，生产/测试默认不自动起循环，防泄漏线程） - stub 装配（infra/stub，@Requires(msgx.stubs=true)）：7 仓储内存实装 + StubXmlCodec（未实装 → CODEC_ERROR 走 FAILED 可重放路径）+ StubRedis + StubDeliveryPort + Holder Factory - 配置：application-dev.yml（stubs+autostart，/env、/beans 放开，无需 DB 即可 dev 跑通）； application-test.yml（stubs=true、autostart=false）；README 补 dev stub 运行说明 - PipelineSmokeTest（4）：等价 /beans 装配核验 + 收报→主泵领取→FAILED(CODEC_ERROR) + U11 重放回 PENDING + Dispatcher flush 聚合（stub 用例前清内存状态防上下文复用串扰） - 全量测试 34 个通过
2026-09-06 20:53:14 +08:00
parent 73113779d2
commit f6da0cb782
11 changed files with 518 additions and 7 deletions
@@ -0,0 +1,50 @@
+package com.gzzn.omms.msgexchange.nextgen
+
+import com.gzzn.omms.msgexchange.nextgen.delivery.Dispatcher
+import com.gzzn.omms.msgexchange.nextgen.processing.Pump
+import io.micronaut.context.annotation.Requires
+import io.micronaut.runtime.event.annotation.EventListener
+import io.micronaut.runtime.server.event.ServerStartupEvent
+import jakarta.annotation.PreDestroy
+import jakarta.inject.Singleton
+
+/**
+ * U07（T03+N29）：管道生命周期装配——服务启动（ServerStartupEvent）后，在各自专用单线程
+ * （msgx-pump / msgx-dispatcher，不占用 Netty event loop）上拉起 Pump 与 Dispatcher 循环；
+ * 停机时 requestStop + interrupt + join。仅当 `msgx.pipeline.autostart=true` 时装配
+ * （默认关：需要真实仓储或 msgx.stubs=true 才安全开启）。
+ */
+@Requires(property = "msgx.pipeline.autostart", value = "true")
+@Singleton
+class PipelineLifecycle(
+    private val pump: Pump,
+    private val dispatcher: Dispatcher,
+) {
+    private val threads = mutableListOf<Thread>()
+
+    @Volatile
+    private var started = false
+
+    @EventListener
+    fun start(event: ServerStartupEvent) {
+        startIfNeeded()
+    }
+
+    private fun startIfNeeded() {
+        if (started) return
+        started = true
+        threads += spawn("msgx-pump", pump::loop)
+        threads += spawn("msgx-dispatcher", dispatcher::loop)
+    }
+
+    private fun spawn(name: String, body: () -> Unit): Thread =
+        Thread.ofPlatform().name(name).daemon(true).start { body() }
+
+    @PreDestroy
+    fun stop() {
+        pump.stop()
+        dispatcher.stop()
+        threads.forEach { it.interrupt() }          // 解除 Thread.sleep 阻塞，加速退出
+        threads.forEach { runCatching { it.join(3000) } }
+    }
+}
@@ -29,6 +29,9 @@ class PipelineProps {
        var backoffCapMs: Long = 60_000
        var headDeadline: Duration = Duration.ofMinutes(10)     // 最坏 HOL 上界（毒丸升级）

+        /** U07：启动即拉起 Pump/Dispatcher 循环（默认关——需要真实仓储或 msgx.stubs=true 才可安全开启）。 */
+        var autostart: Boolean = false
+
        /** N28：attempt ≤ 0（如 FAILED 未递增 attempts 的行）不得抛异常，取下界=首档退避。 */
        fun backoffFor(attempt: Int): Long {
            val index = (attempt - 1).coerceAtLeast(0)
@@ -43,6 +43,11 @@ class Dispatcher(

    private var lastFlush: Instant = Instant.EPOCH

+    /** 优雅停机：loop 收尾后退出；线程中断由 Runner 负责。 */
+    fun stop() {
+        running = false
+    }
+
    fun loop() {
        while (running) {
            try {
@@ -0,0 +1,76 @@
+package com.gzzn.omms.msgexchange.nextgen.infra.stub
+
+import com.gzzn.omms.msgexchange.nextgen.codec.DecodeFailure
+import com.gzzn.omms.msgexchange.nextgen.codec.DecodeResult
+import com.gzzn.omms.msgexchange.nextgen.codec.XmlCodec
+import com.gzzn.omms.msgexchange.nextgen.delivery.DeliveryPort
+import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass
+import com.gzzn.omms.msgexchange.nextgen.infra.redis.FlightRedisClient
+import com.gzzn.omms.msgexchange.nextgen.infra.redis.RedisScript
+import com.gzzn.omms.msgexchange.nextgen.processing.CodecHolder
+import com.gzzn.omms.msgexchange.nextgen.processing.HandlerHolder
+import com.gzzn.omms.msgexchange.nextgen.processing.HandlerRegistry
+import io.micronaut.context.annotation.Bean
+import io.micronaut.context.annotation.Factory
+import io.micronaut.context.annotation.Requires
+import jakarta.inject.Singleton
+
+/**
+ * U07：stub 适配层——XmlCodec / Redis 客户端 / DeliveryPort / Handler 装配，仅在 msgx.stubs=true 时生效。
+ * stub codec 未实装 → 报文 decode 返回 CODEC_ERROR（走 FAILED 可重放路径，链路上可观测），
+ * 语义见 ACM2-10 U11：不把“未实装”写成报文非法/终态。
+ */
+@Requires(property = "msgx.stubs", value = "true")
+@Singleton
+class StubXmlCodec : XmlCodec {
+    override fun decode(rawXml: String): DecodeResult =
+        DecodeResult.Err(DecodeFailure(ErrorClass.CODEC_ERROR, "stub:codec-not-implemented"))
+
+    override fun encodeRqrd(kind: String, rangeJson: String): String = ""
+}
+
+@Requires(property = "msgx.stubs", value = "true")
+@Singleton
+class StubRedis : FlightRedisClient {
+    val hash = mutableMapOf<String, String>()
+    val evalCalls = mutableListOf<RedisScript>()
+
+    fun clear() { hash.clear(); evalCalls.clear() }
+
+    override fun eval(script: RedisScript, setPairs: List<Pair<String, String>>, delFields: List<String>) {
+        evalCalls += script
+        when (script) {
+            RedisScript.SNAPSHOT_REPLACE -> {
+                hash.putAll(setPairs)
+                delFields.forEach { hash.remove(it) }
+            }
+            RedisScript.BATCH_DELETE -> delFields.forEach { hash.remove(it) }
+        }
+    }
+
+    override fun hgetAllFlightInfo(): Map<String, String> = hash.toMap()
+}
+
+@Requires(property = "msgx.stubs", value = "true")
+@Singleton
+class StubDeliveryPort : DeliveryPort {
+    val sent = mutableListOf<Pair<String, String>>()
+
+    fun clear() { sent.clear() }
+
+    override fun sendKafka(topic: String, payloadJson: String) { sent += topic to payloadJson }
+    override fun indexFlightHts(payloadJson: String) = Unit
+    override fun projectRedis(payloadJson: String) = Unit
+}
+
+/** Holder 工厂：CodecHolder/HandlerHolder 由 Micronaut Bean 提供（取代直连构造占位）。 */
+@Factory
+@Requires(property = "msgx.stubs", value = "true")
+class StubHolderFactory {
+
+    @Bean
+    fun codecHolder(codec: StubXmlCodec): CodecHolder = CodecHolder(codec)
+
+    @Bean
+    fun handlerHolder(): HandlerHolder = HandlerHolder(HandlerRegistry(emptyList()))   // 阶段 2 前无 Handler 注册
+}
@@ -0,0 +1,234 @@
+package com.gzzn.omms.msgexchange.nextgen.infra.stub
+
+import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass
+import com.gzzn.omms.msgexchange.nextgen.domain.EventStatus
+import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent
+import com.gzzn.omms.msgexchange.nextgen.domain.ProcState
+import com.gzzn.omms.msgexchange.nextgen.domain.ProcStatus
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.CminmsgInboxRepository
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.FlightStateRepository
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.MsgEventRepository
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ProcStateRepository
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.PumpJobRepository
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.RefDataRepository
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ReqTrackRepository
+import io.micronaut.context.annotation.Requires
+import jakarta.inject.Singleton
+import java.time.Instant
+import java.util.concurrent.atomic.AtomicLong
+
+/**
+ * U07（U01 遗留 stub）：内存仓储装配——仅当 `msgx.stubs=true`（dev/影子冒烟）时生效
+ * （@Requires），让「启动 + 收报 + 主泵/投递循环 + /beans 端到端」在无 MySQL/Redis/Kafka 时可跑。
+ * 与真实 Micronaut Data 实装按模块替换；切换点条件显式，不误入生产。
+ */
+@Requires(property = "msgx.stubs", value = "true")
+@Singleton
+class StubProcState : ProcStateRepository {
+    private val rows = linkedMapOf<Long, ProcState>()
+    private val bound = mutableMapOf<String, Long>()
+
+    fun clear() { rows.clear(); bound.clear() }
+
+    override fun insert(cminmsgsId: Long, state: ProcStatus) {
+        rows[cminmsgsId] = ProcState(cminmsgsId, state)
+    }
+
+    override fun headUnfinished(): ProcState? =
+        rows.filterValues { it.state == ProcStatus.PENDING || it.state == ProcStatus.FAILED }
+            .minByOrNull { it.key }?.value
+
+    override fun tryBindIdentity(cminmsgsId: Long, identityKey: String): Boolean {
+        val owner = bound[identityKey]
+        if (owner != null && owner != cminmsgsId) return false
+        bound[identityKey] = cminmsgsId
+        rows[cminmsgsId] = (rows[cminmsgsId] ?: ProcState(cminmsgsId, ProcStatus.PENDING)).copy(identityKey = identityKey)
+        return true
+    }
+
+    override fun ownerOfIdentity(identityKey: String): Long? = bound[identityKey]
+
+    override fun update(
+        cminmsgsId: Long, state: ProcStatus, nextAttemptAt: Instant?, attempts: Int?,
+        errorClass: ErrorClass?, lastError: String?,
+    ) {
+        val old = rows[cminmsgsId] ?: ProcState(cminmsgsId, state)
+        rows[cminmsgsId] = old.copy(
+            state = state,
+            nextAttemptAt = nextAttemptAt ?: old.nextAttemptAt,
+            attempts = attempts ?: old.attempts,
+            errorClass = errorClass ?: old.errorClass,
+            lastError = lastError ?: old.lastError,
+        )
+    }
+
+    override fun requeueByErrorClasses(errorClasses: List<ErrorClass>): Int {
+        var n = 0
+        rows.keys.toList().forEach { id ->
+            val s = rows[id]!!
+            if (s.errorClass != null && s.errorClass in errorClasses &&
+                (s.state == ProcStatus.FAILED || s.state == ProcStatus.DEAD)) {
+                rows[id] = s.copy(state = ProcStatus.PENDING, attempts = 0, nextAttemptAt = null)
+                n++
+            }
+        }
+        return n
+    }
+
+    /** 测试/运维观测用：读取当前状态行。 */
+    fun snapshotOf(cminmsgsId: Long): ProcState? = rows[cminmsgsId]
+}
+
+@Requires(property = "msgx.stubs", value = "true")
+@Singleton
+class StubInbox : CminmsgInboxRepository {
+    private val raws = linkedMapOf<Long, String>()
+
+    fun clear() { raws.clear() }
+    private val ids = AtomicLong(0)
+
+    override fun insertRaw(rawXml: String): Long {
+        val id = ids.incrementAndGet()
+        raws[id] = rawXml
+        return id
+    }
+
+    override fun rawOf(cminmsgsId: Long): String? = raws[cminmsgsId]
+
+    override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) {
+        // 内存 stub：无列可回填，记录即可（后续由真实实装/审计消费）
+    }
+}
+
+@Requires(property = "msgx.stubs", value = "true")
+@Singleton
+class StubMsgEvents : MsgEventRepository {
+    private val rows = linkedMapOf<Long, MsgEvent>()
+
+    fun clear() { rows.clear() }
+    private val ids = AtomicLong(0)
+
+    override fun insertAll(events: List<MsgEvent>): List<Long> = events.map { e ->
+        val id = ids.incrementAndGet()
+        rows[id] = e.copy(eventId = id)
+        id
+    }
+
+    override fun headUnsent(target: String): MsgEvent? =
+        rows.values.filter { it.target == target && it.state == EventStatus.PENDING }
+            .minByOrNull { it.eventId ?: Long.MAX_VALUE }
+
+    override fun claimBatch(target: String, limit: Int): List<MsgEvent> =
+        rows.values.filter { it.target == target && it.state == EventStatus.PENDING }
+            .sortedBy { it.eventId ?: Long.MAX_VALUE }
+            .take(limit)
+
+    override fun markSent(eventId: Long) = mutate(eventId) { it.copy(state = EventStatus.SENT) }
+
+    override fun markAllSent(eventIds: List<Long>) = eventIds.forEach(::markSent)
+
+    override fun scheduleRetry(eventId: Long, nextAttemptAt: Instant, attempts: Int) =
+        mutate(eventId) { it.copy(state = EventStatus.PENDING, attempts = attempts, nextAttemptAt = nextAttemptAt) }
+
+    override fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String, attempts: Int?) =
+        mutate(eventId) { it.copy(state = EventStatus.DEAD, errorClass = errorClass, lastError = lastError, attempts = attempts ?: it.attempts) }
+
+    override fun insertSync(events: List<MsgEvent>) {
+        insertAll(events)
+    }
+
+    private fun mutate(eventId: Long, f: (MsgEvent) -> MsgEvent) {
+        val cur = rows[eventId] ?: return
+        rows[eventId] = f(cur)
+    }
+}
+
+@Requires(property = "msgx.stubs", value = "true")
+@Singleton
+class StubPumpJobs : PumpJobRepository {
+
+    fun clear() { rows.clear() }
+    data class JobRow(val job: PumpJobRepository.Job, var state: String, var lastError: String? = null)
+
+    private val rows = linkedMapOf<Long, JobRow>()
+    private val ids = AtomicLong(0)
+
+    override fun enqueue(kind: String) {
+        val id = ids.incrementAndGet()
+        rows[id] = JobRow(PumpJobRepository.Job(id, kind), "QUEUED")
+    }
+
+    override fun headQueued(): PumpJobRepository.Job? =
+        rows.values.firstOrNull { it.state == "QUEUED" }?.job
+
+    override fun markRunning(jobId: Long) { rows[jobId]?.state = "RUNNING" }
+    override fun markDone(jobId: Long) { rows[jobId]?.state = "DONE" }
+    override fun markFailed(jobId: Long, lastError: String) { rows[jobId]?.state = "FAILED"; rows[jobId]?.lastError = lastError }
+}
+
+@Requires(property = "msgx.stubs", value = "true")
+@Singleton
+class StubRefData : RefDataRepository {
+    private val gens = mutableMapOf<String, RefDataRepository.GenMeta>()
+    private val flat = mutableMapOf<Pair<String, String>, String>()
+
+    fun clear() { gens.clear(); flat.clear() }
+
+    override fun getGen(day: String): RefDataRepository.GenMeta? = gens[day]
+
+    override fun putGenIfVersion(day: String, expected: Long, new: RefDataRepository.GenMeta): Boolean {
+        val cur = gens[day]?.version ?: 0L
+        if (cur != expected) return false
+        gens[day] = new
+        return true
+    }
+
+    override fun upsertAll(rows: List<Triple<String, String, String>>) {
+        rows.forEach { flat[it.first to it.second] = it.third }
+    }
+}
+
+@Requires(property = "msgx.stubs", value = "true")
+@Singleton
+class StubReqTrack : ReqTrackRepository {
+    private val rows = mutableMapOf<Long, ReqTrackRepository.Req>()
+
+    fun clear() { rows.clear() }
+    private val ids = AtomicLong(0)
+
+    override fun findOpenByKind(kind: String): ReqTrackRepository.Req? =
+        rows.values.firstOrNull { it.reqType == kind && it.state in setOf("REGISTERED", "SENT", "WAITING") }
+
+    override fun forceExpireOpenOf(kind: String) {
+        rows.keys.toList().forEach { id ->
+            val r = rows[id]!!
+            if (r.reqType == kind && r.state in setOf("REGISTERED", "SENT", "WAITING"))
+                rows[id] = r.copy(state = "EXPIRED")
+        }
+    }
+
+    override fun insert(kind: String, paramsJson: String): Long {
+        val id = ids.incrementAndGet()
+        rows[id] = ReqTrackRepository.Req(id, kind, "REGISTERED")
+        return id
+    }
+
+    override fun linkCoutmsgs(reqId: Long, coutmsgsId: Long) = Unit
+    override fun markSent(reqId: Long, sentAt: Instant) { rows[reqId]?.let { rows[reqId] = it.copy(state = "SENT", sentAt = sentAt) } }
+    override fun expireIfWaiting(reqId: Long) { rows[reqId]?.let { rows[reqId] = it.copy(state = "EXPIRED") } }
+    override fun markDone(reqId: Long) { rows[reqId]?.let { rows[reqId] = it.copy(state = "DONE") } }
+}
+
+@Requires(property = "msgx.stubs", value = "true")
+@Singleton
+class StubFlightState : FlightStateRepository {
+    private val byDay = mutableMapOf<String, MutableList<Pair<String, String>>>()
+
+    fun clear() { byDay.clear() }
+
+    override fun replaceDay(day: String, flights: List<Pair<String, String>>) {
+        byDay[day] = flights.toMutableList()
+    }
+
+    override fun findByDay(day: String): List<Pair<String, String>> = byDay[day] ?: emptyList()
+}
@@ -34,6 +34,11 @@ class Pump(
    @Volatile
    private var running = true

+    /** 优雅停机：loop 在当前 tick 收尾后退出；线程中断由 Runner 负责。 */
+    fun stop() {
+        running = false
+    }
+
    fun loop() {
        while (running) {
            try {
@@ -1,6 +1,27 @@
-# 开发/影子对拍环境 profile（U03/N32）：放开管理端点可见性以便“改值—回读”与 /beans 装配核验。
-# 生产环境不加载本 profile（/env、/beans 保持默认 sensitive）。
+# 开发/影子对拍环境 profile（U03/N32 + U07）：
+#  - /env、/beans 放开以便“改值—回读”与装配核验（生产 profile 不加载本文件）
+#  - msgx.stubs=true：无 MySQL/Redis/Kafka 时以内存 stub 仓储跑通「启动 + 收报 + 主泵/投递循环」
+#  - msgx.pipeline.autostart=true：启动即拉起 Pump/Dispatcher 专用线程
 # 启动：MICRONAUT_ENVIRONMENTS=dev ./gradlew run
+msgx:
+  register-eureka: false
+  stubs: true
+  pipeline:
+    autostart: true
+
+# DB 仅在真实实装接入后需要；stub 模式不建连（Flyway 关闭；datasource 保留默认值以免占位符解析失败）
+flyway:
+  datasources:
+    default:
+      enabled: false
+
+datasources:
+  default:
+    url: ${MSGX_DB_URL:jdbc:mysql://127.0.0.1:3306/msgexchange_dev}
+    username: ${MSGX_DB_USER:root}
+    password: ${MSGX_DB_PASSWORD:}
+    driver-class-name: com.mysql.cj.jdbc.Driver
+
 micronaut:
  endpoints:
    env:
@@ -15,6 +15,7 @@ msgx:
    backoff-ms: [1000, 2000, 4000, 8000, 16000]   # 指数退避，单次封顶 60s
    backoff-cap-ms: 60000
    head-deadline: 10m              # 队头滞留上界 = 最坏 HOL 时长（毒丸升级）
+    autostart: false                # U07：启动即拉起 Pump/Dispatcher 循环；需真实仓储或 msgx.stubs=true 才开启（dev 见 application-dev.yml）
  schd:
    flush-period: 3s                # KEEP 现役推送节律
    flush-limit: 500                # 批上限，防积压尖峰