fix(processing): keep exhausted failures terminal
ProcFailure.fail 在 markTerminal 后继续执行 update(FAILED),把 DEAD(EXHAUSTED) 覆写回非终态。改为耗尽分支写完终态即返回,保持 INV-6 终态不可逆与 INV-8 回填意图同语句。 回归:ProcFailureTest 覆盖临界次数保持 DEAD、不可回退 FAILED,以及非耗尽退避。
This commit is contained in:
@@ -30,6 +30,7 @@ class ProcFailure(
|
||||
lastError = "$reason; attempts=$attempts",
|
||||
now = scheduler.now(),
|
||||
)
|
||||
return // 终态已写:不再执行下面的 FAILED 更新,否则会把 DEAD 覆写回非终态
|
||||
}
|
||||
procState.update(
|
||||
head.msgId, ProcStatus.FAILED,
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
package com.gzzn.omms.msgexchange.infra.retry
|
||||
|
||||
import com.gzzn.omms.msgexchange.MutableClock
|
||||
import com.gzzn.omms.msgexchange.config.PipelineProps
|
||||
import com.gzzn.omms.msgexchange.domain.ErrorClass
|
||||
import com.gzzn.omms.msgexchange.domain.ProcStatus
|
||||
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.Test
|
||||
import java.time.Instant
|
||||
|
||||
/**
|
||||
* 处理失败的落账规矩:没到尝试上限才写 FAILED 退避重试;一到上限立刻写 DEAD(EXHAUSTED) 终态,
|
||||
* 不允许同一次失败调用再把它覆写回 FAILED(INV-6 终态不可逆、INV-8 终态与回填意图同一次写入)。
|
||||
*/
|
||||
class ProcFailureTest {
|
||||
|
||||
private val base: Instant = MutableClock.BASE
|
||||
|
||||
private fun scheduler(clock: MutableClock): FailureScheduler =
|
||||
FailureScheduler(PipelineProps(), clock)
|
||||
|
||||
private fun seed(repo: StubProcState, attempts: Int) {
|
||||
repo.insertIfAbsent(1L, base, base)
|
||||
repo.rows[1L] = repo.rows[1L]!!.copy(attempts = attempts)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `exhausted failure stays DEAD and cannot fall back to FAILED`() {
|
||||
val clock = MutableClock(base)
|
||||
val repo = StubProcState()
|
||||
seed(repo, attempts = 4) // 再失败一次即达到 max-attempts = 5
|
||||
|
||||
ProcFailure(repo, scheduler(clock)).fail(repo.find(1L)!!, ErrorClass.INFRA, "boom")
|
||||
|
||||
val row = repo.find(1L)!!
|
||||
assertEquals(ProcStatus.DEAD, row.state)
|
||||
assertEquals(ErrorClass.EXHAUSTED, row.errorClass)
|
||||
assertEquals(5, row.attempts)
|
||||
assertNull(row.nextAttemptAt)
|
||||
assertEquals("boom; attempts=5", row.lastError)
|
||||
assertNotNull(row.backfillNextAt) // 终态与回填意图同一次写入
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `non exhausted failure keeps FAILED with the backoff slot for this attempt`() {
|
||||
val clock = MutableClock(base)
|
||||
val repo = StubProcState()
|
||||
seed(repo, attempts = 0)
|
||||
|
||||
ProcFailure(repo, scheduler(clock)).fail(repo.find(1L)!!, ErrorClass.INFRA, "boom")
|
||||
|
||||
val row = repo.find(1L)!!
|
||||
assertEquals(ProcStatus.FAILED, row.state)
|
||||
assertEquals(ErrorClass.INFRA, row.errorClass)
|
||||
assertEquals(1, row.attempts)
|
||||
assertEquals(base.plusMillis(1000), row.nextAttemptAt)
|
||||
assertNull(row.backfillNextAt) // 非终态不登记回填意图
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user