From db28ac2d4818b28adfabbc4edf6081874695b8bd Mon Sep 17 00:00:00 2001 From: joshavant <830519+joshavant@users.noreply.github.com> Date: Tue, 11 Aug 2026 17:39:55 -0500 Subject: [PATCH] test(signal): cover merged ingress failures --- .../event-handler.ingress-lifecycle.test.ts | 45 ++++++++++++++++++- 1 file changed, 44 insertions(+), 1 deletion(-) diff --git a/extensions/signal/src/monitor/event-handler.ingress-lifecycle.test.ts b/extensions/signal/src/monitor/event-handler.ingress-lifecycle.test.ts index a6a4906f28c8..cb371972e467 100644 --- a/extensions/signal/src/monitor/event-handler.ingress-lifecycle.test.ts +++ b/extensions/signal/src/monitor/event-handler.ingress-lifecycle.test.ts @@ -16,13 +16,19 @@ type DispatchParams = { const { dispatchInboundMessageMock, recordInboundSessionMock, dispatchCapture, dispatchBehavior } = vi.hoisted(() => { const captured: DispatchParams[] = []; - const behavior = { adoptDuringDispatch: true }; + const behavior = { + adoptDuringDispatch: true, + failBeforeAdoption: undefined as Error | undefined, + }; return { dispatchCapture: captured, dispatchBehavior: behavior, recordInboundSessionMock: vi.fn(), dispatchInboundMessageMock: vi.fn(async (params: DispatchParams) => { captured.push(params); + if (behavior.failBeforeAdoption) { + throw behavior.failBeforeAdoption; + } if (behavior.adoptDuringDispatch) { // Mirror the real reply lane: adoption fires while dispatch runs. await params.replyOptions?.turnAdoptionLifecycle?.onAdopted(); @@ -102,9 +108,11 @@ beforeAll(async () => { function createTrackedLifecycle(): SignalIngressLifecycle & { adoptedCount: () => number; abandonedCount: () => number; + failedCount: () => number; } { let adopted = 0; let abandoned = 0; + let failed = 0; return { abortSignal: new AbortController().signal, onAdopted: async () => { @@ -112,11 +120,15 @@ function createTrackedLifecycle(): SignalIngressLifecycle & { }, onDeferred: () => {}, onAdoptionFinalizing: () => {}, + onFailed: () => { + failed += 1; + }, onAbandoned: () => { abandoned += 1; }, adoptedCount: () => adopted, abandonedCount: () => abandoned, + failedCount: () => failed, }; } @@ -133,6 +145,7 @@ describe("signal drain claim ownership", () => { beforeEach(() => { dispatchCapture.length = 0; dispatchBehavior.adoptDuringDispatch = true; + dispatchBehavior.failBeforeAdoption = undefined; dispatchInboundMessageMock.mockClear(); }); @@ -180,6 +193,36 @@ describe("signal drain claim ownership", () => { ); }); + it("fails every constituent claim when a merged dispatch rejects before adoption", async () => { + dispatchBehavior.failBeforeAdoption = new Error("merged dispatch failed"); + const handler = createSignalEventHandler( + createBaseSignalEventHandlerDeps({ + cfg: { messages: { inbound: { debounceMs: 40 } } }, + }), + ); + const first = createTrackedLifecycle(); + const second = createTrackedLifecycle(); + + const results = [ + await handler(createDataEvent({ timestamp: 1700000003500, message: "part one" }), first), + await handler(createDataEvent({ timestamp: 1700000003600, message: "part two" }), second), + ]; + expect(results).toEqual([{ kind: "deferred" }, { kind: "deferred" }]); + + await vi.waitFor( + () => { + expect(dispatchInboundMessageMock).toHaveBeenCalledOnce(); + expect(first.failedCount()).toBe(1); + expect(second.failedCount()).toBe(1); + }, + { timeout: 5_000 }, + ); + expect(first.adoptedCount()).toBe(0); + expect(second.adoptedCount()).toBe(0); + expect(first.abandonedCount()).toBe(0); + expect(second.abandonedCount()).toBe(0); + }); + it("settles the claim for a turn that finishes without adoption", async () => { // Dispatch resolves without the reply lane ever adopting (gated/no-reply): // the flush must complete the claim, or the stall watchdog would