From a6b2120836d1cf586b4a00f3c2a11cf33f02c515 Mon Sep 17 00:00:00 2001 From: windyboy Date: Sat, 12 Sep 2026 22:39:16 +0800 Subject: [PATCH] =?UTF-8?q?fix(infra):=20=E6=B6=88=E9=99=A4=20StubReposito?= =?UTF-8?q?ries=20=E7=BC=96=E8=AF=91=E8=AD=A6=E5=91=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - KDoc 移到注解之前（2 处） - upsertSchd 改显式 null 检查避免 !! 警告 - findBackfillDue 链式调用转 asSequence - 移除 persistFullState 中 always-null 的 fallback - markDeleted/revive 改 !! 断言（mains/snapshots 同步写入） --- .../infra/stub/StubRepositories.kt | 23 +++++++++++-------- 1 file changed, 13 insertions(+), 10 deletions(-) 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 b1b2db9..19c3517 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 @@ -54,9 +54,9 @@ class StubPipelineLock : PipelineLockRepository { override fun lock() = Unit } +/** 内存版 PROC_STATE：入队幂等，写终态时一并写下回填待办。 */ @Singleton @Requires(property = "msgx.stubs", value = "true") -/** 内存版 PROC_STATE：入队幂等，写终态时一并写下回填待办。 */ class StubProcState(private val clock: Clock = Clock.systemUTC()) : ProcStateRepository { val rows = linkedMapOf() val bound = linkedMapOf() @@ -186,6 +186,7 @@ class StubProcState(private val clock: Clock = Clock.systemUTC()) : ProcStateRep override fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List = rows.values + .asSequence() .filter { it.state.isTerminal() && it.backfillAt == null && it.backfillAbandonedAt == null } // 超期判据是本地入队时间，与 JDBC 实现保持一致（不用库方时钟的 receivedAt）。 .map { it to (it.enqueuedAt?.let { e -> e < overdueBefore } ?: false) } @@ -194,6 +195,7 @@ class StubProcState(private val clock: Clock = Clock.systemUTC()) : ProcStateRep .sortedWith(compareBy({ it.first.backfillAttempts }, { it.first.msgId })) .take(limit) .map { (row, overdue) -> BackfillDue(row.msgId, row.backfillAttempts, overdue) } + .toList() override fun hasAny(): Boolean = rows.isNotEmpty() @@ -251,11 +253,12 @@ class StubMsgEvents : MsgEventRepository { */ private fun upsertSchd(e: MsgEvent): Long { val existing = rows.values.firstOrNull { it.target == Targets.KAFKA_SCHD && it.partitionKey == e.partitionKey } - val accepted = existing == null || - e.stateVersion > existing.stateVersion || - (e.stateVersion == existing.stateVersion && e.eventType == EventType.TOMBSTONE && existing.eventType != EventType.TOMBSTONE) - if (!accepted) return existing!!.eventId!! - existing?.let { rows.remove(it.eventId!!) } + if (existing != null) { + val accepted = e.stateVersion > existing.stateVersion || + (e.stateVersion == existing.stateVersion && e.eventType == EventType.TOMBSTONE && existing.eventType != EventType.TOMBSTONE) + if (!accepted) return existing.eventId!! + rows.remove(existing.eventId!!) + } val id = ids.incrementAndGet() rows[id] = e.copy( eventId = id, state = EventStatus.PENDING, attempts = 0, nextAttemptAt = null, @@ -327,7 +330,7 @@ class StubFlightState : FlightStateRepository { } val main = FlightMainRow( flid = snapshot.flid, - operationDay = snapshot.operationDay ?: existing?.operationDay, + operationDay = snapshot.operationDay, state = snapshot.state, stateVersion = snapshot.stateVersion, lastMsgId = msgId, @@ -342,7 +345,7 @@ class StubFlightState : FlightStateRepository { val main = mains[flid] ?: return false if (main.state != FlightState.ACTIVE) return false mains[flid] = main.copy(state = FlightState.DELETED, stateVersion = main.stateVersion + 1, lastMsgId = msgId, updatedAt = now) - snapshots[flid] = (snapshots[flid] ?: return true).copy(state = FlightState.DELETED, stateVersion = main.stateVersion + 1) + snapshots[flid] = snapshots[flid]!!.copy(state = FlightState.DELETED, stateVersion = main.stateVersion + 1) return true } @@ -350,7 +353,7 @@ class StubFlightState : FlightStateRepository { val main = mains[flid] ?: return false if (main.state != FlightState.DELETED) return false mains[flid] = main.copy(state = FlightState.ACTIVE, stateVersion = main.stateVersion + 1, lastMsgId = msgId, updatedAt = now) - snapshots[flid] = (snapshots[flid] ?: return true).copy(state = FlightState.ACTIVE, stateVersion = main.stateVersion + 1) + snapshots[flid] = snapshots[flid]!!.copy(state = FlightState.ACTIVE, stateVersion = main.stateVersion + 1) return true } @@ -450,9 +453,9 @@ class StubReqTrack : ReqTrackRepository { } } +/** 内存版共享信箱：可以模拟上游写入、库方清除，并记录哪些行被打上了处理标记。 */ @Singleton @Requires(property = "msgx.stubs", value = "true") -/** 内存版共享信箱：可以模拟上游写入、库方清除，并记录哪些行被打上了处理标记。 */ class StubInbox(private val clock: Clock = Clock.systemUTC()) : CminmsgInboxRepository { val raws = linkedMapOf() private val received = linkedMapOf()