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); }