diff --git a/src/auto-reply/inbound-debounce.ts b/src/auto-reply/inbound-debounce.ts index b5eb48e9fed9..63f93fe1c3f0 100644 --- a/src/auto-reply/inbound-debounce.ts +++ b/src/auto-reply/inbound-debounce.ts @@ -127,13 +127,13 @@ function createInboundDebounceFlush(params: { } // A failed dispatch must settle its source claim before releasing the keyed // lane; an already-admitted turn owns its later completion failure. - void completion - .then(markAdmitted, async (error: unknown) => { - if (!admitted) { - await lifecycle.onFailed?.(error); - } - }) - .then(markAdmitted, markAdmitted); + completion = completion.then(markAdmitted).catch(async (error: unknown) => { + if (!admitted && lifecycle.onFailed) { + await Promise.allSettled([lifecycle.onFailed(error)]); + } + markAdmitted(); + throw error; + }); return { admission, completion }; } @@ -197,7 +197,7 @@ export function createInboundDebouncer(params: InboundDebounceCreateParams activeCompletions.add(completion); const cleanup = () => activeCompletions.delete(completion); void completion.then(cleanup, cleanup); - await admission; + await Promise.race([admission, completion]); }; const cancelItems = (items: T[]) => { diff --git a/src/auto-reply/inbound.test.ts b/src/auto-reply/inbound.test.ts index 9e8c612c5516..06d74ca05c26 100644 --- a/src/auto-reply/inbound.test.ts +++ b/src/auto-reply/inbound.test.ts @@ -1008,6 +1008,45 @@ describe("createInboundDebouncer", () => { expect(calls).toEqual(["1", "2"]); }); + it("releases serialized keys when custom completion rejects before admission", async () => { + const failure = new Error("custom flush failed"); + const calls: string[] = []; + const reported: unknown[] = []; + const pendingAdmission = new Promise(() => {}); + const debouncer = createInboundDebouncer<{ key: string; id: string }>({ + debounceMs: 0, + serializeImmediate: true, + buildKey: (item) => item.key, + onFlush: (items) => { + const id = items[0]?.id ?? ""; + calls.push(id); + if (id === "first") { + return { admission: pendingAdmission, completion: Promise.reject(failure) }; + } + return flushOnCompletion(() => {}); + }, + onError: (error) => { + reported.push(error); + throw new Error("observer failed"); + }, + }); + + const first = debouncer.enqueue({ key: "a", id: "first" }); + await vi.waitFor(() => expect(calls).toEqual(["first"])); + const second = debouncer.enqueue({ key: "a", id: "second" }); + const secondOutcome = await Promise.race([ + second.then(() => "completed" as const), + new Promise<"stalled">((resolve) => { + setTimeout(() => resolve("stalled"), 100); + }), + ]); + + expect(secondOutcome).toBe("completed"); + await Promise.all([first, second, debouncer.drain()]); + expect(calls).toEqual(["first", "second"]); + expect(reported).toEqual([failure]); + }); + it("does not leak unhandled rejections when a keyed flush failure is awaited", async () => { const debouncer = createInboundDebouncer<{ key: string; id: string }>({ debounceMs: 0,