fix: release call arguments once a forwarded or queued call is sent

.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`.

.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.

__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<Counter>(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<Listener>;
+      subscribe(listener: RpcStub<Listener>) { 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<Counter>(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<TestTarget>(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 {

src/rpc.ts

@@ -187,6 +187,10 @@ class ImportTableEntry {
   constructor(public session: RpcSessionImpl, public importId: number, pulling: boolean) {
     if (pulling) {
       this.activePull = Promise.withResolvers<void>();
+
+      // 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,21 +814,25 @@ 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;
       }
 
       // HACK: Since the args is an array, devaluator will wrap in a second array. Need to unwrap.
       // TODO: Clean this up somehow.
       value.push((<Array<unknown>>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);
     }