Accept an AbortSignal in onRpcBroken and cleanup registrations when stubs are disposed

.changeset/onrpcbroken-abort-signal.md

@@ -0,0 +1,7 @@
+---
+"capnweb": minor
+---
+
+`onRpcBroken()` now takes an optional `AbortSignal`: `stub.onRpcBroken(cb, { signal })`. Aborting the signal drops the callback, and a signal that is already aborted registers nothing, even when the stub is already broken.
+
+Disposing a stub now drops the callbacks registered on it. This changes existing behavior. Previously, disposing a stub or promise left its `onRpcBroken()` callbacks registered, so they still fired when the connection was later lost. Disposal is per-stub, so callbacks registered on other stubs pointing at the same object still fire.

README.md

@@ -433,6 +433,23 @@ If anything happens to the stub that would cause all further method calls and pr
 * The stub's underlying connection is lost.
 * The stub is a promise, and the promise rejects.
 
+To stop listening, pass an `AbortSignal`:
+
+```ts
+let controller = new AbortController();
+
+stub.onRpcBroken((error: any) => {
+  console.error(error);
+}, { signal: controller.signal });
+
+// Later: stop listening.
+controller.abort();
+```
+
+Aborting the signal drops the callback. A signal that is already aborted registers nothing, even if the stub is already broken.
+
+Disposing a stub also drops the callbacks registered on it.
+
 ## Security Considerations
 
 * The WebSocket API in browsers always permits cross-site connections, and does not permit setting headers. Because of this, you generally cannot use cookies nor other headers for authentication. Instead, we highly recommend the pattern shown in the second example above, in which authentication happens in-band via an RPC method that returns the authenticated API.

__tests__/index.test.ts

@@ -2175,6 +2175,173 @@ describe("onRpcBroken", () => {
       {which: "hangingCall", error: new Error("test disconnect")},
     ]);
   });
+
+  it("never registers a callback whose signal is already aborted", async () => {
+    class TestBroken extends RpcTarget {
+      makeCounter() { return new Counter(0); }
+      throwError(): Promise<Counter> { throw new Error("test error"); }
+    }
+
+    let harness = new TestHarness(new TestBroken());
+    let stub = harness.stub;
+
+    let preAborted = new AbortController();
+    preAborted.abort();
+    let aborted = preAborted.signal;
+
+    let errors: string[] = [];
+
+    stub.onRpcBroken(() => { errors.push("stub"); }, {signal: aborted});
+
+    let counterPromise = stub.makeCounter();
+    counterPromise.onRpcBroken(() => { errors.push("counterPromise"); }, {signal: aborted});
+
+    // An already-broken stub normally reports synchronously; the aborted signal suppresses it.
+    let throwingPromise = stub.throwError();
+    await throwingPromise.catch(err => {});
+    throwingPromise.onRpcBroken(() => { errors.push("throwError"); }, {signal: aborted});
+    expect(errors).toStrictEqual([]);
+
+    harness.clientTransport.forceReceiveError(new Error("test disconnect"));
+    await pumpMicrotasks();
+
+    expect(errors).toStrictEqual([]);
+  });
+
+  it("deregisters a callback when its signal is aborted", async () => {
+    class TestBroken extends RpcTarget {
+      getValue() { return 42; }
+    }
+
+    let harness = new TestHarness(new TestBroken());
+    let stub = harness.stub;
+    expect(await stub.getValue()).toBe(42);
+
+    let errors: string[] = [];
+    let canceled = new AbortController();
+
+    stub.onRpcBroken(() => { errors.push("kept1"); });
+    stub.onRpcBroken(() => { errors.push("canceled"); }, {signal: canceled.signal});
+    stub.onRpcBroken(() => { errors.push("kept2"); });
+
+    canceled.abort();
+
+    harness.clientTransport.forceReceiveError(new Error("test disconnect"));
+    await pumpMicrotasks();
+
+    // The canceled callback is gone; the others still fire in registration order.
+    expect(errors).toStrictEqual(["kept1", "kept2"]);
+  });
+
+  it("honors the signal after the promise it was registered on has resolved", async () => {
+    class TestBroken extends RpcTarget {
+      makeCounter() { return new Counter(0); }
+    }
+
+    let harness = new TestHarness(new TestBroken());
+    let stub = harness.stub;
+
+    let errors: string[] = [];
+    let canceled = new AbortController();
+
+    // Register while the promise is still unresolved, so that the registration is later migrated
+    // onto the resolution.
+    let counterPromise = stub.makeCounter();
+    counterPromise.onRpcBroken(() => { errors.push("canceled"); }, {signal: canceled.signal});
+    counterPromise.onRpcBroken(() => { errors.push("kept"); });
+
+    await counterPromise;
+
+    // Abort only after the migration has happened.
+    canceled.abort();
+
+    harness.clientTransport.forceReceiveError(new Error("test disconnect"));
+    await pumpMicrotasks();
+
+    expect(errors).toStrictEqual(["kept"]);
+  });
+
+  it("removes the callback when the stub it was registered on is disposed", async () => {
+    class TestBroken extends RpcTarget {
+      getValue() { return 42; }
+    }
+
+    let harness = new TestHarness(new TestBroken());
+    let stub = harness.stub;
+    expect(await stub.getValue()).toBe(42);
+
+    let errors: string[] = [];
+
+    // Register on a dup, so that disposing it leaves the underlying import (and the registration
+    // made through `stub` below) alive.
+    let dup = stub.dup();
+    dup.onRpcBroken(() => { errors.push("dup"); });
+    stub.onRpcBroken(() => { errors.push("stub"); });
+
+    dup[Symbol.dispose]();
+
+    harness.clientTransport.forceReceiveError(new Error("test disconnect"));
+    await pumpMicrotasks();
+
+    // Disposing the dup drops its own callback even though no AbortSignal was involved. The
+    // callback registered on `stub` still fires.
+    expect(errors).toStrictEqual(["stub"]);
+  });
+
+  it("removes the callback when the promise it was registered on is disposed", async () => {
+    class TestBroken extends RpcTarget {
+      getValue() { return 42; }
+      hangingCall(): Promise<Counter> {
+        return new Promise(() => {});  // never resolves
+      }
+    }
+
+    let harness = new TestHarness(new TestBroken());
+    let stub = harness.stub;
+
+    let errors: string[] = [];
+
+    let hangingPromise = stub.hangingCall();
+    hangingPromise.onRpcBroken(() => { errors.push("hangingCall"); });
+    stub.onRpcBroken(() => { errors.push("stub"); });
+
+    hangingPromise[Symbol.dispose]();
+
+    harness.clientTransport.forceReceiveError(new Error("test disconnect"));
+    await pumpMicrotasks();
+
+    expect(errors).toStrictEqual(["stub"]);
+  });
+
+  it("honors independent signals on separate dups of the same import", async () => {
+    class TestBroken extends RpcTarget {
+      getValue() { return 42; }
+    }
+
+    let harness = new TestHarness(new TestBroken());
+    let stub = harness.stub;
+    expect(await stub.getValue()).toBe(42);
+
+    let errors: string[] = [];
+
+    // Two dups of the same underlying import, each registering with its own signal. Aborting one
+    // signal must drop only that dup's callback, since each hook composes its own signal with the
+    // shared entry.
+    let dup1 = stub.dup();
+    let dup2 = stub.dup();
+    let canceled = new AbortController();
+    let kept = new AbortController();
+
+    dup1.onRpcBroken(() => { errors.push("dup1"); }, {signal: canceled.signal});
+    dup2.onRpcBroken(() => { errors.push("dup2"); }, {signal: kept.signal});
+
+    canceled.abort();
+
+    harness.clientTransport.forceReceiveError(new Error("test disconnect"));
+    await pumpMicrotasks();
+
+    expect(errors).toStrictEqual(["dup2"]);
+  });
 });
 
 // =======================================================================================

__tests__/signal.test.ts

@@ -0,0 +1,110 @@
+// Copyright (c) 2026 Cloudflare, Inc.
+// Licensed under the MIT license found in the LICENSE.txt file or at:
+//     https://opensource.org/license/mit
+
+// Tests for `anySignal()`, the `AbortSignal.any` shim in src/signal.ts. Covers both the native
+// path (when `AbortSignal.any` exists) and the hand-rolled fallback used on runtimes that lack it.
+
+import { expect, it, describe, afterEach } from "vitest"
+import { anySignal } from "../src/signal.js"
+
+// Run the given callback with `AbortSignal.any` removed, forcing `anySignal()` down its fallback
+// path. The original is always restored, even if the callback throws.
+function withoutNativeAny(fn: () => void) {
+  let original = Object.getOwnPropertyDescriptor(AbortSignal, "any");
+  // @ts-expect-error - deleting an optional static for the duration of the test.
+  delete AbortSignal.any;
+  try {
+    expect(AbortSignal.any).toBeUndefined();
+    fn();
+  } finally {
+    if (original) Object.defineProperty(AbortSignal, "any", original);
+  }
+}
+
+describe("anySignal", () => {
+  afterEach(() => {
+    // Guard against a test leaving `AbortSignal.any` deleted if it somehow escaped the finally.
+    expect(typeof AbortSignal.any).toBe("function");
+  });
+
+  describe("native path", () => {
+    it("aborts when any source aborts", () => {
+      let a = new AbortController();
+      let b = new AbortController();
+      let composite = anySignal([a.signal, b.signal]);
+
+      expect(composite.aborted).toBe(false);
+      b.abort(new Error("boom"));
+      expect(composite.aborted).toBe(true);
+      expect((composite.reason as Error).message).toBe("boom");
+    });
+
+    it("is already aborted when a source is already aborted", () => {
+      let a = new AbortController();
+      a.abort(new Error("pre"));
+      let composite = anySignal([a.signal, new AbortController().signal]);
+
+      expect(composite.aborted).toBe(true);
+      expect((composite.reason as Error).message).toBe("pre");
+    });
+  });
+
+  describe("fallback path (AbortSignal.any unavailable)", () => {
+    it("aborts when any source aborts later, propagating the reason", () => {
+      withoutNativeAny(() => {
+        let a = new AbortController();
+        let b = new AbortController();
+        let composite = anySignal([a.signal, b.signal]);
+
+        expect(composite.aborted).toBe(false);
+        b.abort(new Error("boom"));
+        expect(composite.aborted).toBe(true);
+        expect((composite.reason as Error).message).toBe("boom");
+      });
+    });
+
+    it("is already aborted when a source is already aborted, propagating the reason", () => {
+      withoutNativeAny(() => {
+        let a = new AbortController();
+        a.abort(new Error("pre"));
+        let composite = anySignal([a.signal, new AbortController().signal]);
+
+        expect(composite.aborted).toBe(true);
+        expect((composite.reason as Error).message).toBe("pre");
+      });
+    });
+
+    it("only fires once even if multiple sources abort", () => {
+      withoutNativeAny(() => {
+        let a = new AbortController();
+        let b = new AbortController();
+        let composite = anySignal([a.signal, b.signal]);
+
+        let reasons: unknown[] = [];
+        composite.addEventListener("abort", () => { reasons.push(composite.reason); });
+
+        a.abort(new Error("first"));
+        b.abort(new Error("second"));
+
+        expect(reasons.length).toBe(1);
+        expect((reasons[0] as Error).message).toBe("first");
+      });
+    });
+
+    it("stops listening to a source once the composite has aborted", () => {
+      withoutNativeAny(() => {
+        let a = new AbortController();
+        let b = new AbortController();
+        let composite = anySignal([a.signal, b.signal]);
+
+        a.abort(new Error("first"));
+
+        // `b` outlives the composite; aborting it must not touch the already-settled reason, and
+        // its listener should have been dropped when the composite aborted.
+        b.abort(new Error("second"));
+        expect((composite.reason as Error).message).toBe("first");
+      });
+    });
+  });
+});

__type-tests__/rpc-promise-semantics.test.ts

@@ -67,6 +67,11 @@ userPromise.onRpcBroken((_error) => {})
 counterPromise.onRpcBroken((_error) => {})
 idPromise.onRpcBroken((_error) => {})
 
+// The options bag is optional and takes an AbortSignal.
+userPromise.onRpcBroken((_error) => {}, {})
+counterPromise.onRpcBroken((_error) => {}, { signal: new AbortController().signal })
+idPromise.onRpcBroken((_error) => {}, { signal: undefined })
+
 expectAssignable<Promise<number>>(counterPromise.increment(3))
 expectAssignable<Promise<number>>(counterPromise.value)
 expectAssignable<Promise<string>>(userPromise.getName())

packages/capnweb-validate/src/internal/core.ts

@@ -56,7 +56,10 @@ type WrapSide = "server" | "client";
 
 interface StubBase<T = unknown> extends Disposable {
   dup(): this;
-  onRpcBroken(callback: (error: unknown) => void): void;
+  onRpcBroken(
+    callback: (error: unknown) => void,
+    options?: { signal?: AbortSignal }
+  ): void;
   readonly __RPC_STUB_BRAND: T;
 }