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 d17c5ee..8a4d89b 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/MsgEvent.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/MsgEvent.kt @@ -28,5 +28,5 @@ data class MsgEvent( val nextAttemptAt: java.time.Instant? = null, val errorClass: ErrorClass? = null, val lastError: String? = null, - val createdAt: java.time.Instant = java.time.Instant.now(), + val createdAt: java.time.Instant, ) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/ProcState.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/ProcState.kt index 5d5f32d..27e48bb 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/ProcState.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/ProcState.kt @@ -86,5 +86,5 @@ data class ProcState( val backfillAbandonedAt: Instant? = null, /** 放弃原因（`MISSING_ROW` / `TRANSIENT_DEADLINE`），供人工对账与恢复判断。 */ val backfillAbandonedReason: String? = null, - val updatedAt: Instant = Instant.now(), + val updatedAt: Instant, ) 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 c0e713c..b1b2db9 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 @@ -30,6 +30,7 @@ import com.gzzn.omms.msgexchange.domain.SnapshotLogEntry import com.gzzn.omms.msgexchange.infra.persistence.SnapshotLogRepository import io.micronaut.context.annotation.Requires import jakarta.inject.Singleton +import java.time.Clock import java.time.Instant import java.time.LocalDate import java.time.ZoneId @@ -56,7 +57,7 @@ class StubPipelineLock : PipelineLockRepository { @Singleton @Requires(property = "msgx.stubs", value = "true") /** 内存版 PROC_STATE：入队幂等，写终态时一并写下回填待办。 */ -class StubProcState : ProcStateRepository { +class StubProcState(private val clock: Clock = Clock.systemUTC()) : ProcStateRepository { val rows = linkedMapOf() val bound = linkedMapOf() @@ -74,6 +75,7 @@ class StubProcState : ProcStateRepository { msgId, ProcStatus.PENDING, receivedAt = receivedAt, enqueuedAt = enqueuedAt ?: receivedAt, + updatedAt = clock.instant(), ) return true } @@ -89,7 +91,7 @@ class StubProcState : ProcStateRepository { val owner = bound[identityKey] if (owner != null && owner != msgId) return false bound[identityKey] = msgId - rows[msgId] = (rows[msgId] ?: ProcState(msgId, ProcStatus.PENDING)).copy(identityKey = identityKey) + rows[msgId] = (rows[msgId] ?: ProcState(msgId, ProcStatus.PENDING, updatedAt = clock.instant())).copy(identityKey = identityKey) return true } @@ -103,14 +105,14 @@ class StubProcState : ProcStateRepository { errorClass: ErrorClass?, lastError: String?, ) { - val old = rows[msgId] ?: ProcState(msgId, state) + val old = rows[msgId] ?: ProcState(msgId, state, updatedAt = clock.instant()) rows[msgId] = old.copy( state = state, nextAttemptAt = nextAttemptAt ?: old.nextAttemptAt, attempts = attempts ?: old.attempts, errorClass = errorClass ?: old.errorClass, lastError = lastError ?: old.lastError, - updatedAt = Instant.now(), + updatedAt = clock.instant(), ) } @@ -123,7 +125,7 @@ class StubProcState : ProcStateRepository { attempts: Int?, now: Instant, ) { - val old = rows[msgId] ?: ProcState(msgId, state) + val old = rows[msgId] ?: ProcState(msgId, state, updatedAt = clock.instant()) rows[msgId] = old.copy( state = state, errorClass = errorClass, @@ -451,7 +453,7 @@ class StubReqTrack : ReqTrackRepository { @Singleton @Requires(property = "msgx.stubs", value = "true") /** 内存版共享信箱：可以模拟上游写入、库方清除，并记录哪些行被打上了处理标记。 */ -class StubInbox : CminmsgInboxRepository { +class StubInbox(private val clock: Clock = Clock.systemUTC()) : CminmsgInboxRepository { val raws = linkedMapOf() private val received = linkedMapOf() private val marks = linkedMapOf() @@ -464,7 +466,7 @@ class StubInbox : CminmsgInboxRepository { override fun insertRaw(rawXml: String): Long { val id = ids.incrementAndGet() raws[id] = rawXml - received[id] = Instant.now() + received[id] = clock.instant() return id } @@ -497,7 +499,7 @@ class StubInbox : CminmsgInboxRepository { fun simulateExternalWrite(rawXml: String): Long = insertRaw(rawXml) /** 测试辅助：模拟先分配 ID、稍后才可见的迟提交行。 */ - fun restoreRow(msgId: Long, rawXml: String, receivedAt: Instant = Instant.now()) { + fun restoreRow(msgId: Long, rawXml: String, receivedAt: Instant = clock.instant()) { raws[msgId] = rawXml received[msgId] = receivedAt } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt index 4acb102..8511bc1 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt @@ -48,7 +48,7 @@ class InboxPoller( private var running = false /** @return 本轮新登记的消息条数（已登记过的行不计入，也不影响水位推进）。 */ - fun pollOnce(now: Instant = Instant.now()): Int { + fun pollOnce(now: Instant): Int { seedCutoverWatermarkIfConfigured(now) detectLateArrivalsIfDue(now) val batch = props.pipeline.claimBatch.coerceAtLeast(1) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJob.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJob.kt index 747ddd5..729cb70 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJob.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJob.kt @@ -49,7 +49,7 @@ class HistorySweepJob( data class SweepOutcome(val selected: Int, val archived: Int, val purged: Int, val snapLogPurged: Int = 0) - fun run(now: Instant = Instant.now()): SweepOutcome { + fun run(now: Instant): SweepOutcome { // 留痕是本地可重建记录，清理不需要外部归档确认，不受历史存储开关牵制。 val snapLogPurged = snapLogPurge?.purgeBefore(now.minus(Duration.ofDays(props.snapLogRetentionDays))) ?: 0 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Identity.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Identity.kt index 1ec0adb..693d14a 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Identity.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Identity.kt @@ -16,12 +16,13 @@ object Identity { fun of( msg: DecodedMessage, includeDayBoundary: Boolean, - day: LocalDate = LocalDate.now(), + day: LocalDate, ): String { val base = "${msg.meta.sndr}|${msg.meta.type}|${msg.meta.styp}|${msg.meta.seqn}" return if (includeDayBoundary) "$base|${day}" else base } - fun of(msg: DecodedMessage, props: PipelineProps.Identity): String = - of(msg, props.includeDayBoundary) + /** 日期边界由调用方按机场时区算出后传入；本算法不再自己读系统时间。 */ + fun of(msg: DecodedMessage, props: PipelineProps.Identity, day: LocalDate): String = + of(msg, props.includeDayBoundary, day) } 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 9f409ef..ad6a5b5 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/Pump.kt @@ -3,6 +3,7 @@ package com.gzzn.omms.msgexchange.processing import com.gzzn.omms.msgexchange.codec.FlopPayload import com.gzzn.omms.msgexchange.codec.ScheduleBody 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.ErrorClass @@ -18,6 +19,7 @@ import jakarta.inject.Singleton import java.time.Clock import java.time.Duration import java.time.Instant +import java.time.LocalDate import java.util.concurrent.atomic.AtomicLong /** @@ -147,6 +149,7 @@ class MessageProcessor( private val procFailure: ProcFailure, private val props: PipelineProps, private val clock: Clock, + private val operationDayProps: OperationDayProps, ) { private val log = org.slf4j.LoggerFactory.getLogger(MessageProcessor::class.java) @@ -186,7 +189,9 @@ class MessageProcessor( // I3：identity 仅首次绑定（head.identityKey == null）；FAILED 重试不重绑 if (head.identityKey == null) { - val identity = Identity.of(decoded, props.identity) + val identity = Identity.of( + decoded, props.identity, LocalDate.now(clock.withZone(operationDayProps.zoneId())), + ) if (!procState.tryBindIdentity(head.msgId, identity)) { val owner = procState.ownerOfIdentity(identity) ?: -1L log.info("duplicate-of:{} -> SKIPPED msgId={}", owner, head.msgId) diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt index 0ae5231..e49f592 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt @@ -92,7 +92,7 @@ class PipelineSmokeTest { assertNotNull(receipt.body()) // 受理 ID val id = receipt.body()!!.toLong() // 兼容入口只保证"已落信 + 已入队"，**不推进水位**；必须先被收报发现（W 追平）才可领取。 - ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce() + ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce(Instant.now()) pump.tick() // 解码成功但无 DELY Handler → FAILED(UNSUPPORTED) @@ -109,7 +109,7 @@ class PipelineSmokeTest { fun `replay reopens failed UNSUPPORTED row to PENDING`() { val receipt = controller.send(UNSUPPORTED_XML) val id = receipt.body()!!.toLong() - ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce() + ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce(Instant.now()) pump.tick() val stub = ctx.getBean(StubProcState::class.java) assertEquals(ProcStatus.FAILED, stub.snapshotOf(id)!!.state) @@ -135,7 +135,7 @@ class PipelineSmokeTest { val inbox = ctx.getBean(com.gzzn.omms.msgexchange.infra.stub.StubInbox::class.java) val proc = ctx.getBean(StubProcState::class.java) val dead = inbox.simulateExternalWrite("") - ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce() + ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce(Instant.now()) pump.tick() // 解码 MALFORMED → DEAD + 回填意图（同一条 UPDATE） @@ -152,7 +152,7 @@ class PipelineSmokeTest { assertTrue(inbox.isMarked(dead)) val fresh = inbox.simulateExternalWrite("") - ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce() + ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce(Instant.now()) assertNotNull(proc.find(fresh)) // 死信不阻断后续发现 } @@ -167,7 +167,7 @@ class PipelineSmokeTest { val flights = ctx.getBean(com.gzzn.omms.msgexchange.infra.stub.StubFlightState::class.java) val events = ctx.getBean(StubMsgEvents::class.java) val id = inbox.simulateExternalWrite("") // 非法报文（无 META）→ MALFORMED - ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce() + ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce(Instant.now()) pump.tick() @@ -260,8 +260,8 @@ class PipelineSmokeTest { val events = ctx.getBean(StubMsgEvents::class.java) val port = ctx.getBean(StubDeliveryPort::class.java) events.insertAll(listOf( - MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = "F1", stateVersion = 1, payloadJson = """{"FLID":"F1","v":"old"}"""), - MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = "F1", stateVersion = 2, payloadJson = """{"FLID":"F1","v":"new"}"""), + MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = "F1", stateVersion = 1, payloadJson = """{"FLID":"F1","v":"old"}""", createdAt = Instant.now()), + MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = "F1", stateVersion = 2, payloadJson = """{"FLID":"F1","v":"new"}""", createdAt = Instant.now()), )) dispatcher.flushSchd() diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/delivery/DispatcherTickTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/delivery/DispatcherTickTest.kt index 2cedc60..6bc3cb4 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/delivery/DispatcherTickTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/delivery/DispatcherTickTest.kt @@ -32,7 +32,7 @@ class DispatcherTickTest { Dispatcher(repo, port, p, FailureScheduler(p, clock)) private fun ev(id: Long, target: String, key: String, payload: String, version: Long = 0) = - MsgEvent(eventId = id, target = target, partitionKey = key, stateVersion = version, payloadJson = payload) + MsgEvent(eventId = id, target = target, partitionKey = key, stateVersion = version, payloadJson = payload, createdAt = MutableClock.BASE) @Test fun `first tick delivers msg and starts the independent schd flush`() { @@ -107,6 +107,7 @@ class DispatcherTickTest { MsgEvent( eventId = 1, target = Targets.KAFKA_SCHD, partitionKey = "F1", eventType = EventType.TOMBSTONE, stateVersion = 3, payloadJson = """{"flid":"F1","deleted":true}""", + createdAt = MutableClock.BASE, ), ), ) @@ -160,6 +161,7 @@ class DispatcherTickTest { MsgEvent( target = Targets.KAFKA_SCHD, partitionKey = "F1", eventType = EventType.TOMBSTONE, stateVersion = 3, payloadJson = """{"flid":"F1","deleted":true}""", + createdAt = MutableClock.BASE, ), ), ) @@ -234,7 +236,7 @@ private class ReentrantSchdPort(private val repo: StubMsgEvents) : DeliveryPort override fun sendKafkaSchd(topic: String, key: String, payloadJson: String) { repo.insertAll( listOf( - MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = key, stateVersion = 99, payloadJson = """{"v":"new"}"""), + MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = key, stateVersion = 99, payloadJson = """{"v":"new"}""", createdAt = MutableClock.BASE), ), ) } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt index 5c34989..3f576fa 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt @@ -22,13 +22,14 @@ class ReplayServiceTest { fun seed(id: Long, status: ProcStatus, ec: ErrorClass?) { rows[id] = ProcState( id, status, identityKey = "k$id", attempts = 3, errorClass = ec, lastError = "x", + updatedAt = Instant.EPOCH, ) } override fun insertIfAbsent(msgId: Long, receivedAt: Instant?, enqueuedAt: Instant?): Boolean = rows.putIfAbsent( msgId, - ProcState(msgId, ProcStatus.PENDING, receivedAt = receivedAt, enqueuedAt = enqueuedAt ?: receivedAt), + ProcState(msgId, ProcStatus.PENDING, receivedAt = receivedAt, enqueuedAt = enqueuedAt ?: receivedAt, updatedAt = Instant.EPOCH), ) == null override fun find(msgId: Long): ProcState? = rows[msgId] override fun findSuccessTerminal(msgId: Long): Boolean = rows[msgId]?.state == ProcStatus.SUCCEEDED diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FdelAndAdftProcessorTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FdelAndAdftProcessorTest.kt index 8c0875f..5cd4ab1 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FdelAndAdftProcessorTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FdelAndAdftProcessorTest.kt @@ -33,7 +33,7 @@ class FdelAndAdftProcessorTest { private val msgId = 7L - private fun head() = ProcState(msgId, ProcStatus.PENDING) + private fun head() = ProcState(msgId, ProcStatus.PENDING, updatedAt = java.time.Instant.EPOCH) private fun msg() = DecodedMessage( meta = MetaFields("AODB", "FDEL", "", 9L, 1L), @@ -172,6 +172,6 @@ class FdelAndAdftProcessorTest { @Suppress("unused") private fun unused() { - MsgEvent(target = "x", partitionKey = "y", payloadJson = "") + MsgEvent(target = "x", partitionKey = "y", payloadJson = "", createdAt = java.time.Instant.EPOCH) } } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IdentityTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IdentityTest.kt index 92712be..35f67d2 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IdentityTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IdentityTest.kt @@ -22,7 +22,7 @@ class IdentityTest { @Test fun `identity is pipe-joined SNDR TYPE STYP SEQN`() { - val identity = Identity.of(msg(42), includeDayBoundary = false) + val identity = Identity.of(msg(42), includeDayBoundary = false, day = LocalDate.of(2026, 9, 6)) assertEquals("AODB|FLOP|DELY|42", identity) } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt index 2de576e..0a7ffc5 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt @@ -39,7 +39,7 @@ class ScheduleProcessorTest { private val msgId = 42L - private fun head() = ProcState(msgId, ProcStatus.PENDING) + private fun head() = ProcState(msgId, ProcStatus.PENDING, updatedAt = java.time.Instant.EPOCH) private fun message(body: ScheduleBody) = DecodedMessage( meta = MetaFields("AODB", "SCHD", "DNLD", 100L, 1L),