From 5beb67dadbc856066dd121ac75536818fffc1b24 Mon Sep 17 00:00:00 2001 From: windyboy Date: Sat, 12 Sep 2026 20:43:11 +0800 Subject: [PATCH] fix(ingress): preserve compat receipt semantics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 接收结果边界改为 insertRaw 是否返回 ID：接收时间读取失败回退注入 Clock 并留痕，PG insertIfAbsent 失败只记日志与 pgEnqueueFailures 计数，不再冒泡成 HTTP 失败（避免客户端重试多写信箱行）。缺的 PROC_STATE 由收报轮询按 US-01 补建且只建一条（INV-1），入口高 ID 仍等水位追平（INV-4）。 验证：./gradlew test 135 tests / 0 fail / 0 skipped（新增 InboxServiceTest 4 例：读时间失败+入队失败仍返回同一 ID、接收时间 NULL 回退时钟、真实插入失败仍失败且无伪造 ID、轮询对失败入队补建且仅一条）。 --- .../omms/msgexchange/ingress/InboxService.kt | 28 +++- .../msgexchange/ingress/InboxServiceTest.kt | 123 ++++++++++++++++++ 2 files changed, 144 insertions(+), 7 deletions(-) create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxServiceTest.kt diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxService.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxService.kt index e132113..54a5a98 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxService.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxService.kt @@ -5,6 +5,7 @@ import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository import jakarta.inject.Singleton import java.time.Clock import java.time.Instant +import java.util.concurrent.atomic.AtomicLong /** * 兼容 HTTP 入口：把一条报文写进共享信箱，再在自有 PG 里入队，供联调和影子对拍使用。 @@ -24,21 +25,34 @@ class InboxService( ) { private val log = org.slf4j.LoggerFactory.getLogger(InboxService::class.java) + /** 落信成功但 PG 入队失败的次数：由收报轮询按 `US-01` 补建，这里只做可观测计数。 */ + val pgEnqueueFailures = AtomicLong(0) + data class Receipt(val msgId: Long, val receivedAt: Instant) fun accept(rawXml: String): Receipt { + // 接收结果边界 = insertRaw 是否成功返回 ID（C-28 只承诺到落信）。只有这一步失败才返回失败； + // 之后的读取/入队失败若冒泡成 HTTP 失败，客户端重试会在信箱里多写一行。 val id = inbox.insertRaw(rawXml) - // 信箱接收时间缺失不能让"已经落信"的写入对外报失败（那会与"确认落信才返回 ID"的 - // 契约矛盾，并让客户端重试产生额外重复行）。回退到入队时间并留痕； - // 时间源用注入的 Clock，保证与收报/回填一致、测试可确定。 - val mailboxReceivedAt = inbox.receivedAtOf(id) + val now = clock.instant() + val mailboxReceivedAt = try { + inbox.receivedAtOf(id) + } catch (e: Exception) { + log.warn("mailbox receive time read failed msgId={}: {}", id, e.message) + null + } if (mailboxReceivedAt == null) { log.warn("mailbox receive time missing msgId={}, falling back to enqueue time", id) } - val now = clock.instant() val receivedAt = mailboxReceivedAt ?: now - // 入队时间单独传本地时钟：它是超期判据 R 的比较对象，不能借用库方时间（PRE-4）。 - procState.insertIfAbsent(id, receivedAt, enqueuedAt = now) + try { + // 入队时间单独传本地时钟：它是超期判据 R 的比较对象，不能借用库方时间（PRE-4）。 + procState.insertIfAbsent(id, receivedAt, enqueuedAt = now) + } catch (e: Exception) { + // PG 入队失败不回退接收结果：原文已在信箱，轮询会按 ID 补建，且只建一条（INV-1、US-01）。 + pgEnqueueFailures.incrementAndGet() + log.error("compat accepted but PG enqueue failed msgId={}; poller will re-create", id, e) + } log.info("compat-accepted msgId={}", id) return Receipt(id, receivedAt) } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxServiceTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxServiceTest.kt new file mode 100644 index 0000000..0107571 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxServiceTest.kt @@ -0,0 +1,123 @@ +package com.gzzn.omms.msgexchange.ingress + +import com.gzzn.omms.msgexchange.MutableClock +import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters +import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository +import com.gzzn.omms.msgexchange.infra.persistence.MailboxMarkResult +import com.gzzn.omms.msgexchange.infra.persistence.MailboxRow +import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository +import com.gzzn.omms.msgexchange.infra.stub.StubInbox +import com.gzzn.omms.msgexchange.infra.stub.StubInboxCursor +import com.gzzn.omms.msgexchange.infra.stub.StubPipelineTx +import com.gzzn.omms.msgexchange.infra.stub.StubProcState +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Assertions.assertThrows +import org.junit.jupiter.api.Test +import java.time.Instant + +/** + * 兼容入口的接收边界（`C-28`、`INV-1`）：`insertRaw` 返回 ID 即"已落信"。 + * 此后的接收时间读取失败或 PG 入队失败都只留痕/计数，不改接收结果——否则客户端重试会在 + * 信箱里多写一行；缺的 `PROC_STATE` 由收报轮询按 `US-01` 补建，且只建一条。 + */ +class InboxServiceTest { + + private val t0: Instant = Instant.parse("2026-09-06T02:00:00Z") + + /** 可注入失败的信箱：覆盖"插入失败 / 接收时间读失败 / 接收时间为 NULL"三种情形。 */ + private class FakeInbox( + private val rawFailure: Boolean = false, + private val receivedFailure: Boolean = false, + private val receivedNull: Boolean = false, + ) : CminmsgInboxRepository { + override fun insertRaw(rawXml: String): Long = + if (rawFailure) throw IllegalStateException("mailbox-down") else 42L + + override fun rawOf(msgId: Long): String? = null + + override fun receivedAtOf(msgId: Long): Instant? = when { + receivedFailure -> throw IllegalStateException("read-time-down") + receivedNull -> null + else -> Instant.parse("2026-09-06T01:59:55Z") + } + + override fun readRange(fromExclusive: Long, limit: Int): List = emptyList() + + override fun maxId(): Long? = null + + override fun minId(): Long? = null + + override fun existingIds(msgIds: Collection): Set = emptySet() + + override fun markProcessedIfUnmarked(msgId: Long, value: String): MailboxMarkResult = MailboxMarkResult.MISSING + } + + /** PG 入队第一次失败，其余委托给内存仓储（模拟轮询随后补建）。 */ + private class FlakyProcState(private val delegate: StubProcState) : ProcStateRepository by delegate { + private var failNextInsert = true + + override fun insertIfAbsent(msgId: Long, receivedAt: Instant?, enqueuedAt: Instant?): Boolean { + if (failNextInsert) { + failNextInsert = false + throw IllegalStateException("pg-down") + } + return delegate.insertIfAbsent(msgId, receivedAt, enqueuedAt) + } + } + + @Test + fun `receipt survives receive-time read failure and pg enqueue failure`() { + val proc = StubProcState() + val service = InboxService(FakeInbox(receivedFailure = true), FlakyProcState(proc), MutableClock(t0)) + + val receipt = service.accept("") + + assertEquals(42L, receipt.msgId) // 已落信：ID 照常返回 + assertEquals(t0, receipt.receivedAt) // 读失败 → 回退到注入时钟 + assertEquals(1L, service.pgEnqueueFailures.get()) // 入队失败只计数，不冒泡 + assertNull(proc.find(42L)) + } + + @Test + fun `missing receive time falls back to the injected clock`() { + val service = InboxService(FakeInbox(receivedNull = true), StubProcState(), MutableClock(t0)) + + val receipt = service.accept("") + + assertEquals(t0, receipt.receivedAt) + assertEquals(0L, service.pgEnqueueFailures.get()) + } + + @Test + fun `a real mailbox insert failure still fails without a fake id`() { + val service = InboxService(FakeInbox(rawFailure = true), StubProcState(), MutableClock(t0)) + + assertThrows(IllegalStateException::class.java) { service.accept("") } + assertEquals(0L, service.pgEnqueueFailures.get()) + } + + @Test + fun `the poller backfills a compat receipt whose pg enqueue failed, exactly once`() { + val inbox = StubInbox().apply { clear() } + val proc = StubProcState().apply { clear() } + val cursor = StubInboxCursor().apply { clear() } + val clock = MutableClock(t0) + val service = InboxService(inbox, FlakyProcState(proc), clock) + val poller = InboxPoller(inbox, proc, cursor, StubPipelineTx(), PipelineProps(), clock, PipelineCounters()) + + val receipt = service.accept("") // 落信成功；PG 入队失败但不抛 + assertEquals(1L, service.pgEnqueueFailures.get()) + assertNull(proc.find(receipt.msgId)) + + assertEquals(1, poller.pollOnce(t0)) // 轮询补建 + assertNotNull(proc.find(receipt.msgId)) + assertEquals(1, proc.rows.size) + assertEquals(receipt.msgId, cursor.cursor.committedUpTo) + + assertEquals(0, poller.pollOnce(t0)) // 已存在：不重复建行，也不重复计数 + assertEquals(1, proc.rows.size) + } +}