test(signal): cover merged ingress failures

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