From a89755c2008db705e236d537d202dd7abf67c0cd Mon Sep 17 00:00:00 2001 From: windyboy Date: Sat, 12 Sep 2026 20:32:27 +0800 Subject: [PATCH] fix(processing): clear replay error state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 人工重放把 FAILED/DEAD 拨回 PENDING 时，与 attempts/next_attempt_at 一起清空 error_class 与 last_error，使活动状态不携带上一轮失败结论；identity_key、白名单与 MessageLifecycleGate 不变。 同语句同时改 JDBC 与内存实现，避免两套行为漂移。验证：ReplayServiceTest 3/0、InboxLifecycleJdbcSqlTest 8/0（H2 真实 SQL）。 --- .../persistence/jdbc/JdbcPgRepositories.kt | 3 ++- .../infra/stub/StubRepositories.kt | 5 +++- .../jdbc/InboxLifecycleJdbcSqlTest.kt | 24 +++++++++++++++++++ .../infra/retry/ReplayServiceTest.kt | 10 ++++++-- 4 files changed, 38 insertions(+), 4 deletions(-) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt index 41d72ce..15ac7de 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt @@ -276,7 +276,8 @@ class JdbcProcStateRepository( if (errorClasses.isEmpty()) return 0 val placeholders = errorClasses.joinToString(",") { "?" } return ds.update( - "UPDATE proc_state SET state = 'PENDING', attempts = 0, next_attempt_at = NULL, updated_at = ? " + + "UPDATE proc_state SET state = 'PENDING', attempts = 0, next_attempt_at = NULL, " + + "error_class = NULL, last_error = NULL, updated_at = ? " + "WHERE state IN ('FAILED', 'DEAD') AND error_class IN ($placeholders)", { ps -> ps.setTimestamp(1, clock.instant().toSqlTimestamp()) 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 815118a..f190613 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 @@ -210,7 +210,10 @@ class StubProcState : ProcStateRepository { var n = 0 rows.forEach { (id, s) -> if (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++ } } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/InboxLifecycleJdbcSqlTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/InboxLifecycleJdbcSqlTest.kt index 26ae4c2..256211e 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/InboxLifecycleJdbcSqlTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/InboxLifecycleJdbcSqlTest.kt @@ -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) diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt index 9f0f42f..5c34989 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/retry/ReplayServiceTest.kt @@ -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) }