diff --git a/src/channels/message/ingress-drain.debounce-failure.test.ts b/src/channels/message/ingress-drain.debounce-failure.test.ts index 0537cbe32119..8a8396f31eba 100644 --- a/src/channels/message/ingress-drain.debounce-failure.test.ts +++ b/src/channels/message/ingress-drain.debounce-failure.test.ts @@ -84,4 +84,62 @@ describe("channel ingress drain debounce failures", () => { drain.dispose(); }); }); + + it("keeps watchdog recovery after retry settlement fails", async () => { + vi.useFakeTimers(); + await withTempState(async (stateDir) => { + let clock = 10_000; + const queue = createTestIngressQueue(stateDir, { now: () => clock }); + await queue.enqueue( + "debounced-settlement-failure", + { text: "retry me" }, + { laneKey: "shared", receivedAt: clock }, + ); + queue.release = async () => { + throw new Error("persistent release failure"); + }; + + const sessionError = new Error("Session changed while starting work. Retry."); + const debouncer = createInboundDebouncer<{ lifecycle: ChannelIngressDispatchLifecycle }>({ + debounceMs: 0, + buildKey: () => "shared", + onFlush: (entries, createFlush) => + createFlush({ + lifecycle: entries[0]?.lifecycle, + dispatch: async () => { + throw sessionError; + }, + }), + onError: () => undefined, + }); + const drain = createChannelIngressDrain({ + queue, + now: () => clock, + adoptionStallTimeoutMs: 200_000, + dispatchClaimedEvent: async (_event, lifecycle) => { + await debouncer.enqueue({ lifecycle }); + return { kind: "deferred" }; + }, + }); + + expect(await drain.drainOnce()).toEqual({ started: 1 }); + await vi.advanceTimersByTimeAsync(127_000); + clock += 127_000; + await drain.waitForIdle(); + + expect((await queue.listClaims()).map((claim) => claim.id)).toEqual([ + "debounced-settlement-failure", + ]); + expect(drain.activeLaneKeys().has("shared")).toBe(true); + + clock += 73_000; + await vi.advanceTimersByTimeAsync(73_000); + expect(await queue.listClaims()).toEqual([]); + expect(await queue.listFailed?.({ limit: "all" })).toMatchObject([ + { id: "debounced-settlement-failure", reason: "handler-timeout" }, + ]); + expect(drain.activeLaneKeys().has("shared")).toBe(false); + drain.dispose(); + }); + }); }); diff --git a/src/channels/message/ingress-drain.ts b/src/channels/message/ingress-drain.ts index 4cc9c0b250c5..b6cdd089f41f 100644 --- a/src/channels/message/ingress-drain.ts +++ b/src/channels/message/ingress-drain.ts @@ -523,7 +523,7 @@ export function createChannelIngressDrain< if (state.guillotined || state.superseded) { return; } - clearStallTimer(state); + // Keep recovery armed until disposition commits; removeActive clears it after success. await state.settleOnce(async () => { await applyFailureDisposition(state.claim, error); });