From c615b65e99400b582116ed0f7d9fff21dc7d4cfd Mon Sep 17 00:00:00 2001 From: Nathan Disidore Date: Mon, 9 Nov 2026 08:33:02 -0600 Subject: [PATCH] fix: release call arguments once a forwarded or queued call is sent When a call's arguments are an owned payload, the serializer only dups the payload's stubs or cites their import IDs, and sendCall() never disposed the payload on success. That is the case for a call forwarded to a remote stub (three parties, or a .map() callback) and for a call made through a pending RpcPromise. An RpcTarget passed this way never had its disposer run, and its table entries lived as long as the session. #241 covered the failure paths; this is the success path. Dispose the payload after the message is sent. Disposing can release an import the message cites, so doing it earlier makes the peer abort with "no such entry on exports table". App params don't own their stubs, so for them this does nothing. sendStream() gets the same line to honor the stream() ownership contract, though no public path delivers an owned payload there today. Disposing forwarded arguments can dispose an imported promise nobody has pulled yet, which rejected an unobserved promise (#293). Give activePull a no-op catch; real waiters still see the rejection. --- .changeset/release-forwarded-args.md | 5 ++ .changeset/unpulled-promise-rejection.md | 5 ++ __tests__/index.test.ts | 85 ++++++++++++++++++++++++ src/rpc.ts | 19 ++++-- 4 files changed, 109 insertions(+), 5 deletions(-) create mode 100644 .changeset/release-forwarded-args.md create mode 100644 .changeset/unpulled-promise-rejection.md diff --git a/.changeset/release-forwarded-args.md b/.changeset/release-forwarded-args.md new file mode 100644 index 00000000..bc128477 --- /dev/null +++ b/.changeset/release-forwarded-args.md @@ -0,0 +1,5 @@ +--- +"capnweb": patch +--- + +Fix stubs passed as call arguments never being released when the call is forwarded to a remote stub (including from a `.map()` callback) or made through a pending `RpcPromise`. diff --git a/.changeset/unpulled-promise-rejection.md b/.changeset/unpulled-promise-rejection.md new file mode 100644 index 00000000..58ff1367 --- /dev/null +++ b/.changeset/unpulled-promise-rejection.md @@ -0,0 +1,5 @@ +--- +"capnweb": patch +--- + +Fix an unhandled rejection when a call that was passed a promise fails, or is forwarded, before the promise is pulled. diff --git a/__tests__/index.test.ts b/__tests__/index.test.ts index 48008efd..c5868261 100644 --- a/__tests__/index.test.ts +++ b/__tests__/index.test.ts @@ -1580,6 +1580,19 @@ describe("promise pipelining", () => { await expect(() => promise).rejects.toThrow("test error"); await expect(() => promise2).rejects.toThrow("test error"); }); + + it("doesn't create an unhandled rejection when a call taking a promise fails", async () => { + await using harness = new TestHarness(new TestTarget()); + let stub: any = harness.stub; + + // The call fails before the server ever pulls the promise it received, so the server disposes + // an import that is still pending. + using counter = new RpcPromise(Promise.resolve(new Counter(4))); + await expect(() => stub.noSuchMethod(counter)).rejects.toThrow("is not a function"); + + // Give an unhandled rejection the chance to be reported while this test is still running. + await new Promise(resolve => setTimeout(resolve, 0)); + }); }); describe("promises and Blobs inside a Set", () => { @@ -2595,6 +2608,65 @@ describe("RpcImportHook argument disposal", () => { // ======================================================================================= +describe("forwarding calls to a stub from another session", () => { + // The near session's main object is a stub pointing into the far session. A call on it is not + // delivered to a local function: its arguments, as parsed off the wire, are serialized again + // onto the far session. + + it("releases a forwarded stub once the callee releases it", async () => { + let disposed = false; + class Listener extends RpcTarget { + notify() { return "notified"; } + [Symbol.dispose]() { disposed = true; } + } + + class Publisher extends RpcTarget { + private listener?: RpcStub; + subscribe(listener: RpcStub) { this.listener = listener.dup(); } + publish() { return this.listener!.notify(); } + unsubscribe() { this.listener![Symbol.dispose](); } + } + + await using far = new TestHarness(new Publisher()); + await using near = new TestHarness(far.stub.dup()); + let stub = near.stub; + + await stub.subscribe(new Listener()); + expect(await stub.publish()).toBe("notified"); + expect(disposed).toBe(false); + + await stub.unsubscribe(); + expect(disposed).toBe(true); + }); + + it("forwards a pipelined argument", async () => { + await using far = new TestHarness(new TestTarget()); + await using near = new TestHarness(far.stub.dup()); + let stub = near.stub; + + // On the far session, only the forwarded arguments refer to the counter promise. It must not + // be released before the call that cites it has been sent. + using counter = stub.makeCounter(4); + expect(await stub.incrementCounter(counter, 2)).toBe(6); + }); + + it("forwards a promise exported by the caller", async () => { + await using far = new TestHarness(new TestTarget()); + await using near = new TestHarness(far.stub.dup()); + let stub = near.stub; + + // The near session imports this promise, forwards it, and then releases its own import of + // it while still unresolved. + using counter = new RpcPromise(Promise.resolve(new Counter(4))); + expect(await stub.incrementCounter(counter, 2)).toBe(6); + + // Give an unhandled rejection the chance to be reported while this test is still running. + await new Promise(resolve => setTimeout(resolve, 0)); + }); +}); + +// ======================================================================================= + describe("constructing RpcPromise from a promise", () => { it("pipelines through a pending promise without pulling the resolution", async () => { await using harness = new TestHarness(new TestTarget()); @@ -2683,6 +2755,19 @@ describe("constructing RpcPromise from a promise", () => { await pumpMicrotasks(); }); + it("releases a stub passed in a call made through the promise", async () => { + let disposed = false; + class DisposableCounter extends Counter { + [Symbol.dispose]() { disposed = true; } + } + + await using harness = new TestHarness(new TestTarget()); + using stub = new RpcPromise(Promise.resolve(harness.stub.dup())); + + expect(await stub.incrementCounter(new DisposableCounter(4), 2)).toBe(6); + expect(disposed).toBe(true); + }); + it("disposes the eventual target when disposed before resolution", async () => { let disposed = false; class Disposable extends RpcTarget { diff --git a/src/rpc.ts b/src/rpc.ts index d3fc8267..ed6cebe6 100644 --- a/src/rpc.ts +++ b/src/rpc.ts @@ -187,6 +187,10 @@ class ImportTableEntry { constructor(public session: RpcSessionImpl, public importId: number, pulling: boolean) { if (pulling) { this.activePull = Promise.withResolvers(); + + // Nothing awaits this promise until pull() is called, and disposing the import first + // rejects it. That must not be reported as an unhandled rejection. + this.activePull.promise.catch(() => {}); } } @@ -810,6 +814,9 @@ class RpcSessionImpl implements Importer, Exporter { try { devalue = Devaluator.devaluate(args.value, undefined, this, args, this.encodingLevel); } catch (err) { + // Disposing the partially-serialized payload is safe: any hooks already exported were + // dups of the payload's hooks, and the Devaluator rolls back its own export-table entries + // before rethrowing. args.dispose(); throw err; } @@ -817,14 +824,15 @@ class RpcSessionImpl implements Importer, Exporter { // HACK: Since the args is an array, devaluator will wrap in a second array. Need to unwrap. // TODO: Clean this up somehow. value.push((>devalue)[0]); - - // Serializing the payload takes ownership of all stubs within, so the payload itself does - // not need to be disposed on success. If serialization threw partway through, disposing - // the payload above is safe: any hooks already exported were dups of the payload's hooks, - // and the Devaluator rolls back its own export-table entries before rethrowing. } this.send(["push", value]); + // A payload parsed off the wire, or deep-copied to wait on a promise, still owns its hooks: + // the message only carries dups of them or cites their import IDs. Dispose them once the + // message is sent, since that can release an import the message cites. For params straight + // from the app, which don't own their stubs, this does nothing. + args?.dispose(); + let entry = new ImportTableEntry(this, this.imports.length, false); this.imports.push(entry); return new RpcImportHook(/*isPromise=*/true, entry); @@ -853,6 +861,7 @@ class RpcSessionImpl implements Importer, Exporter { let msg = ["stream", value]; let size = this.send(msg); + args.dispose(); // only after sending, as in sendCall() if (size === undefined) { size = estimateEncodedSize(msg); }