feat: support streaming WritableStream writes with lifecycle-managed RPC payloads

.changeset/cool-firm-full.md

@@ -0,0 +1,5 @@
+---
+"capnweb": minor
+---
+
+Support RpcTargets (and other RPC stubs) as ReadableStream/WritableStream chunks without disposing their capabilities when `write()` returns. Stream chunk payloads now keep lifecycle tied to the chunk (via `Symbol.dispose` when needed) so methods on streamed stubs remain usable after the write resolves.

__tests__/index.test.ts

@@ -3005,6 +3005,119 @@ describe("ReadableStream over RPC", () => {
 
     expect(cancelCalled).toBe(true);
   });
+
+  it("can send an RpcTarget as a stream chunk and call methods on it", async () => {
+    // The sender enqueues a pass-by-reference RpcTarget into a ReadableStream. The receiver
+    // reads it back as a stub and can call methods on it, which route back to the sender.
+    class ChunkTarget extends RpcTarget {
+      square(i: number) { return i * i; }
+      get label() { return "chunk-target"; }
+    }
+
+    let stream = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new ChunkTarget());
+        controller.close();
+      }
+    });
+
+    class StreamReceiver extends RpcTarget {
+      async receiveStream(stream: ReadableStream) {
+        let reader = stream.getReader();
+        let { value } = await reader.read();
+        using chunk = value as any;
+        let result = {
+          squared: await chunk.square(5),
+          label: await chunk.label,
+        };
+        // Drain the closing `done` chunk so the underlying pipe is released.
+        await reader.read();
+        return result;
+      }
+    }
+
+    await using harness = new TestHarness(new StreamReceiver());
+    let result: any = await harness.stub.receiveStream(stream);
+
+    expect(result.squared).toBe(25);
+    expect(result.label).toBe("chunk-target");
+  });
+
+  it("can return a ReadableStream containing RpcTargets and call methods on chunks", async () => {
+    // The server returns a ReadableStream whose chunks are RpcTargets. The client reads them
+    // as stubs and can call methods that route back to the server.
+    class ChunkTarget extends RpcTarget {
+      square(i: number) { return i * i; }
+      get label() { return "server-chunk"; }
+    }
+
+    class StreamProvider extends RpcTarget {
+      getStream(): ReadableStream {
+        return new ReadableStream({
+          start(controller) {
+            controller.enqueue(new ChunkTarget());
+            controller.close();
+          }
+        });
+      }
+    }
+
+    await using harness = new TestHarness(new StreamProvider());
+    let stream: any = await harness.stub.getStream();
+
+    let reader = stream.getReader();
+    let { value } = await reader.read();
+    using chunk = value as any;
+
+    expect(await chunk.square(6)).toBe(36);
+    expect(await chunk.label).toBe("server-chunk");
+
+    // Drain the closing `done` chunk so the underlying pipe is released.
+    let { done } = await reader.read();
+    expect(done).toBe(true);
+  });
+
+  it("can send multiple RpcTargets as stream chunks", async () => {
+    class ChunkTarget extends RpcTarget {
+      constructor(private id: number) { super(); }
+      getId() { return this.id; }
+    }
+
+    let stream = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new ChunkTarget(1));
+        controller.enqueue(new ChunkTarget(2));
+        controller.enqueue(new ChunkTarget(3));
+        controller.close();
+      }
+    });
+
+    class StreamReceiver extends RpcTarget {
+      async receiveStream(stream: ReadableStream) {
+        let reader = stream.getReader();
+        let ids: number[] = [];
+        let chunks: any[] = [];
+        try {
+          while (true) {
+            let { done, value } = await reader.read();
+            if (done) break;
+            chunks.push(value);
+            ids.push(await value.getId());
+          }
+          return ids;
+        } finally {
+          for (let chunk of chunks) {
+            chunk[Symbol.dispose]();
+          }
+        }
+      }
+    }
+
+    await using harness = new TestHarness(new StreamReceiver());
+    let result: any = await harness.stub.receiveStream(stream);
+
+    expect(result).toEqual([1, 2, 3]);
+  });
 });
 
 // =======================================================================================

src/core.ts

@@ -1200,6 +1200,58 @@ export class RpcPayload {
     }
   }
 
+  // Deliver this payload as the argument list to `WritableStreamDefaultWriter.write(chunk)`.
+  //
+  // Unlike deliverCall(), stream chunks are enqueued and read asynchronously. Disposing the
+  // payload when write() returns would free any stubs nested in the chunk before the reader
+  // dequeues them. For object chunks we therefore keep the payload alive and attach disposal
+  // to the chunk (same pattern as deliverResolve). Bare RpcStubs already implement
+  // Symbol.dispose, so their own disposer owns the capability. Primitive chunks cannot hold
+  // stubs, so the payload is disposed after write settles.
+  //
+  // Expects `this.value` to be a one-element argument array `[chunk]`. Returns a payload
+  // wrapping the (void) write result.
+  public async deliverStreamWrite(
+      writer: { write(chunk: unknown): Promise<void> }): Promise<RpcPayload> {
+    try {
+      let promises: Promise<void>[] = [];
+      this.deliverTo(this, "value", promises);
+
+      // Same e-order constraint as deliverCall: if there are no nested promises, call write
+      // synchronously so concurrent writes stay ordered.
+      if (promises.length > 0) {
+        await Promise.all(promises);
+      }
+
+      let chunk = (this.value as unknown[])[0];
+
+      if (chunk instanceof Object) {
+        // Object (including RpcStub): do not dispose the payload when write() returns — the
+        // reader still needs any nested capabilities. Attach a payload disposer when the
+        // chunk doesn't already have one (RpcStub already disposes its own hook).
+        if (!(Symbol.dispose in chunk)) {
+          Object.defineProperty(chunk, Symbol.dispose, {
+            value: () => this.dispose(),
+            writable: true,
+            enumerable: false,
+            configurable: true,
+          });
+        }
+        await writer.write(chunk);
+        return RpcPayload.fromAppReturn(undefined);
+      }
+
+      // Primitive chunk: nothing to keep alive beyond the write.
+      await writer.write(chunk);
+      this.dispose();
+      return RpcPayload.fromAppReturn(undefined);
+    } catch (err) {
+      // deliverTo / write failed: free any capabilities we still own.
+      this.dispose();
+      throw err;
+    }
+  }
+
   // Produce a promise for this payload for return to the application. Any RpcPromises in the
   // payload are awaited and substituted with their results first.
   //

src/streams.ts

@@ -62,8 +62,11 @@ class WritableStreamStubHook extends StubHook {
         state.closed = true;
       }
 
-      let func = state.writer[method] as Function;
-      let promise = args.deliverCall(func, state.writer);
+      // write(chunk) delivers asynchronously: stubs nested in the chunk must outlive
+      // writer.write() returning. See RpcPayload.deliverStreamWrite().
+      let promise = method === "write"
+          ? args.deliverStreamWrite(state.writer)
+          : args.deliverCall(state.writer[method] as Function, state.writer);
       return new PromiseStubHook(promise.then(payload => new PayloadStubHook(payload)));
     } catch (err) {
       return new ErrorStubHook(err);