fix(ingress): preserve compat receipt semantics
接收结果边界改为 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、轮询对失败入队补建且仅一条）。
This commit is contained in:
windyboy
@@ -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
|
||||
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)
|
||||
}
|
||||
|
||||
@@ -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<MailboxRow> = emptyList()
|
||||
|
||||
override fun maxId(): Long? = null
|
||||
|
||||
override fun minId(): Long? = null
|
||||
|
||||
override fun existingIds(msgIds: Collection<Long>): Set<Long> = 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("<MSG/>")
|
||||
|
||||
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("<MSG/>")
|
||||
|
||||
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("<MSG/>") }
|
||||
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("<MSG/>") // 落信成功；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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user