feat(jobs): 提交后回填补偿落库 backfill_todo 并到期重试 (ACM2-29 P2-3)

- V1.3.2 迁移：backfill_todo 持久补偿表（attempts/next_attempt_at/last_error）
- BackfillTodoRepository 接口 + PG 实现与 stub；record 幂等 upsert
- BackfillSweepJob：到期重试，成功即删，失败指数退避（30s 起步封顶 15 分钟）；
  触发点 = InboxPoller 每轮心跳（重启即对账）+ JobExecutor BACKFILL_SWEEP 类型
- MessageProcessor/SnapshotFlow 回填失败落待办（成功终态不降级，业务不重放）；
  SKIPPED 重复报文同样补偿共享信箱回填，防止重复信件被反复轮询
- 新测试：扫描重试/退避/恢复对账 4 用例；快照回填失败落库断言；
  跳过路径补偿落库断言

验证：MSGX_PG_PORT=5433 真实 PG 全量 102 用例 0 失败 0 跳过 (ACM2-29)

src/test/kotlin/com/gzzn/omms/msgexchange/processing/MessageProcessorTest.kt

@@ -17,6 +17,7 @@ import com.gzzn.omms.msgexchange.domain.ProcState
 import com.gzzn.omms.msgexchange.domain.ProcStatus
 import com.gzzn.omms.msgexchange.domain.SchdPush
 import com.gzzn.omms.msgexchange.domain.Targets
+import com.gzzn.omms.msgexchange.infra.persistence.BackfillTodoRepository
 import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository
 import com.gzzn.omms.msgexchange.infra.persistence.FlightFields
 import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository
@@ -96,6 +97,7 @@ class MessageProcessorTest {
     private class FakeInbox : CminmsgInboxRepository {
         var raws = mutableMapOf<Long, String>()
         var throwOnRawOf: Throwable? = null
+        var throwOnBackfill: Throwable? = null
         val backfilled = mutableListOf<List<Any?>>()
 
         override fun insertRaw(rawXml: String): Long = 0
@@ -108,6 +110,7 @@ class MessageProcessorTest {
         }
 
         override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) {
+            throwOnBackfill?.let { throw it }
             backfilled += listOf(cminmsgsId, sndr, type, styp, seqn)
         }
     }
@@ -219,14 +222,15 @@ class MessageProcessorTest {
     private fun processor(
         procState: FakeProcState, inbox: FakeInbox, events: FakeEvents, codec: FakeCodec,
         registry: HandlerRegistry = HandlerRegistry(emptyList()),
+        backfillTodo: BackfillTodoRepository? = null,
     ): MessageProcessor {
         val props = PipelineProps()
         val scheduler = FailureScheduler(props, clock)
         val procFailure = ProcFailure(procState, scheduler)
         val flightSchd = FakeFlightSchd()
         val txManager = FakeTxManager()
-        val snapshot = SnapshotFlow(procState, flightSchd, events, FakeReqTrack(), procFailure, txManager, inbox)
-        return MessageProcessor(inbox, procState, events, CodecHolder(codec), HandlerHolder(registry), flightSchd, snapshot, procFailure, props, txManager)
+        val snapshot = SnapshotFlow(procState, flightSchd, events, FakeReqTrack(), procFailure, txManager, inbox, backfillTodo = backfillTodo)
+        return MessageProcessor(inbox, procState, events, CodecHolder(codec), HandlerHolder(registry), flightSchd, snapshot, procFailure, props, txManager, backfillTodo)
     }
     // ---------- tests ----------
     @Test
@@ -308,6 +312,24 @@ class MessageProcessorTest {
         assertTrue(ev.inserted.isEmpty())
     }
 
+    @Test
+    fun `skipped duplicate still compensates mailbox backfill and persists todo on failure`() {
+        val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents()
+        ps.record[1] = head()
+        inbox.raws[1] = "<MSG/>"
+        inbox.throwOnBackfill = IllegalStateException("mysql-down")
+        ps.bound["AODB|FLOP|DELY|1"] = 2L
+        val todo = com.gzzn.omms.msgexchange.infra.stub.StubBackfillTodo()
+
+        processor(ps, inbox, ev, FakeCodec(), backfillTodo = todo).processOne(ps.state(1))
+
+        // SKIPPED 不变，但共享信箱回填的补偿待办必须落库（否则重复报文被反复轮询）
+        assertEquals(ProcStatus.SKIPPED, ps.state(1).state)
+        assertEquals(1, todo.count())
+        assertEquals(1L, todo.tasks().single().cminmsgsId)
+        assertEquals("mysql-down", todo.lastErrorOf(1L))
+    }
+
     // ---------- U08 边界化（评审要求①③） ----------
     @Test
     fun `unexpected exception maps to FAILED INFRA with backoff at boundary`() {

src/test/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlowBackfillTest.kt

@@ -56,11 +56,19 @@ class SnapshotFlowBackfillTest {
             SnapshotFlow.StageResult.Ok(day, listOf("FL_X" to mapOf("FLID" to "FL_X")))
         }
 
-        val flow = SnapshotFlow(procState, flightSchd, msgEvents, reqTrack, procFailure, txManager, inbox)
+        // v2 §5（P2-3）：回填失败必须落持久补偿待办（否则重启后无从对账）
+        val todo = com.gzzn.omms.msgexchange.infra.stub.StubBackfillTodo()
+        val flow = SnapshotFlow(procState, flightSchd, msgEvents, reqTrack, procFailure, txManager, inbox, backfillTodo = todo)
         flow.publishSnapshot(ProcState(headId, ProcStatus.PENDING), decoded)
 
         assertEquals(ProcStatus.SUCCEEDED, procState.snapshotOf(headId)!!.state)
         assertTrue(flightSchd.findByFlid("FL_X") != null)
+        assertEquals(1, todo.count())
+        val recorded = todo.tasks().single()
+        assertEquals(901L, recorded.cminmsgsId)
+        assertEquals("SCHD", recorded.type)
+        assertEquals("DNLD", recorded.styp)
+        assertEquals("mysql-down", todo.lastErrorOf(901L))
     }
 
     @Test