fix(processing): clear replay error state
人工重放把 FAILED/DEAD 拨回 PENDING 时,与 attempts/next_attempt_at 一起清空 error_class 与 last_error,使活动状态不携带上一轮失败结论;identity_key、白名单与 MessageLifecycleGate 不变。 同语句同时改 JDBC 与内存实现,避免两套行为漂移。验证:ReplayServiceTest 3/0、InboxLifecycleJdbcSqlTest 8/0(H2 真实 SQL)。
This commit is contained in:
+24
@@ -142,6 +142,30 @@ class InboxLifecycleJdbcSqlTest {
|
||||
assertEquals(InboxCursorRepository.Cursor(42L, t0), cursor.load())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `replay resets attempts, backoff and the previous error reason`() {
|
||||
seed(12L, t0)
|
||||
ds.update(
|
||||
"UPDATE proc_state SET state = 'FAILED', attempts = 4, next_attempt_at = ?, " +
|
||||
"error_class = 'INFRA', last_error = 'boom', identity_key = 'k12' WHERE msg_id = ?",
|
||||
{ ps ->
|
||||
ps.setTimestamp(1, Timestamp.from(t0))
|
||||
ps.setLong(2, 12L)
|
||||
},
|
||||
)
|
||||
|
||||
val n = proc.requeueByErrorClasses(listOf(ErrorClass.INFRA))
|
||||
|
||||
assertEquals(1, n)
|
||||
val row = proc.find(12L)!!
|
||||
assertEquals(ProcStatus.PENDING, row.state)
|
||||
assertEquals(0, row.attempts)
|
||||
assertNull(row.nextAttemptAt)
|
||||
assertNull(row.errorClass) // 重放后不携带上一轮失败结论(design「失败恢复与维护作业」)
|
||||
assertNull(row.lastError)
|
||||
assertEquals("k12", row.identityKey) // 身份保留,避免重放时把自己判成重复
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `pipeline transaction rolls back terminal and outbox writes after a late failure`() {
|
||||
seed(11L, t0)
|
||||
|
||||
@@ -11,7 +11,7 @@ import java.time.Instant
|
||||
|
||||
/**
|
||||
* 人工重放的规矩:只有"可以重放"的失败原因才能把消息从 FAILED / DEAD 拉回队列
|
||||
* (重试次数清零,马上再试一次);报文本身非法的记录永远不允许重放。
|
||||
* (重试次数清零、旧错误原因清空,马上再试一次);报文本身非法的记录永远不允许重放。
|
||||
*/
|
||||
class ReplayServiceTest {
|
||||
|
||||
@@ -67,7 +67,10 @@ class ReplayServiceTest {
|
||||
val s = rows[id]!!
|
||||
if (s.errorClass != null && s.errorClass in errorClasses &&
|
||||
(s.state == ProcStatus.FAILED || s.state == ProcStatus.DEAD)) {
|
||||
rows[id] = s.copy(state = ProcStatus.PENDING, attempts = 0, nextAttemptAt = null)
|
||||
rows[id] = s.copy(
|
||||
state = ProcStatus.PENDING, attempts = 0, nextAttemptAt = null,
|
||||
errorClass = null, lastError = null,
|
||||
)
|
||||
n++
|
||||
}
|
||||
}
|
||||
@@ -90,6 +93,9 @@ class ReplayServiceTest {
|
||||
assertEquals(ProcStatus.PENDING, repo.rows[1]!!.state)
|
||||
assertEquals(0, repo.rows[1]!!.attempts)
|
||||
assertNull(repo.rows[1]!!.nextAttemptAt)
|
||||
assertNull(repo.rows[1]!!.errorClass) // 重放后不携带上一轮失败结论
|
||||
assertNull(repo.rows[1]!!.lastError)
|
||||
assertEquals("k1", repo.rows[1]!!.identityKey) // 身份保留,避免重放时把自己判成重复
|
||||
assertEquals(ProcStatus.DEAD, repo.rows[2]!!.state) // MALFORMED 永不被重放
|
||||
assertEquals(ProcStatus.PENDING, repo.rows[3]!!.state)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user