Add support for sending ReadableStream and WritableStream over RPC, with automatic flow control. (#132)

* Refactor: RpcPayload.stubs -> hooks

Instead of storing an array of RpcStub, we now store an array of the underlying StubHooks.

This will make it easier to add support for new types like streams, which aren't RpcStubs, but they will wrap / be wrapped in StubHooks.

* Add support for sending WritableStream over RPC.

(Flow control is left for a future commit.)

Written using Claude+Opencode: https://share.opencode.cloudflare.dev/share/gJU0pT8p

(I cleaned some stuff up manually.)

* Add support for sending a ReadableStream over RPC.

As described in protocol.md, the basic idea here is that whenever we want to send a ReadableStream, we first send a message to the other side creating a "pipe". We pump our ReadableStream to the pipe's WritableSteam end, and we deliver the pipe's ReadableStream end to the remote peer. This way, we can begin pushing bytes immediately upon sending a ReadableStream, without waiting for the remote end to call back asking for the bytes (which would be an unnecessary round trip).

Written using Claude+Opencode: https://share.opencode.cloudflare.dev/share/ctbbBnOu

* Implement backpressure for streams.

If more than 256kb of writes are in flight, we pause writes until past writes complete so that the number drops back below 256kb.

Future commits will expand the window size.

Written using Claude+Opencode: https://share.opencode.cloudflare.dev/share/D1bqkx2K

This was not the best Claude session. I could probably have done it faster by hand.

* Optimization: Elide "pull" and "resolve" for streaming calls.

We add a new "stream" message to the protocol which skips these. See protocol.md for explanation.

(Since this is only used for streams, which were just introduced in this PR, this is not a breaking change.)

Written using Claude+Opencode: https://share.opencode.cloudflare.dev/share/OfY838e7

* Add changeset for streaming.

* Adaptively resize streaming windows.

With this change, we'll automatically update a stream's window size based on the observed bandwidth-delay product, in order to fully saturate the stream with minimal additional buffer bloat.

The algorithm works by observing when each stream chunk is sent and acknowledged (via return from the RPC), allowing us to calculate:
1. Minimum round trip time.
2. Running average bandwidth over the last RTT.

From that we calculate bandwidth-delay product and adjust the window to match. We actually set the window a bit bigger than the calculated BDP during startup (2x) and steady-state (1.25x) so that we can observe if the actual bandwidth is greater than expected, and thus update the window accordingly.

I worked with Claude+Opencode to design the algorithm and implement, although I significantly refactored almost everything it wrote as the code was pretty meh: https://share.opencode.cloudflare.dev/share/rGV0SKLW

* Document streaming support in readme.

* Fix streaming error propagation.

This bug was caught and fixed by AI (prompted by @dmmulroy): #132 (comment)

This commit applies exactly the suggestion from the comment.

* Fix occasional hang in stream tests.

This was a pain to track down, when the code goes into a busy loop, vitest gets very confused. It fails to display console.log()s that happened before the hang and doesn't even correctly report which test is running -- often claiming it's still working on some previous test. Ugh!

Anyway, it turns out that the stream implementations on some platforms require macro tasks to make progress, so pumping only microtasks doesn't get there. Weirdly, it seems to be non-deterministic. I was seeing workerd hang maybe 1/4 of the time, and webkit also hang sometimes but less often. It's possible the other platforms were also affected but even more rarely and I just never saw it happen.

* Fix ReadableStreams in received payloads not being disposed.

Fix and test provided by @dmmulroy here: #132 (comment)

* Don't cancel ReadableStreams that are locked.

I noticed that the `state.stream.cancel()` call in `ReadableStreamStubHook.dispose()` was often failing, throwing an exception because the stream was already locked. But we were ignoring the exceptions.

This actually fixes the problem in two ways:
1. When we pipe from a ReadableStream (which locks it), we also take a reference on its StubHook, which we dispose then the pipe completes. This makes sense and solves the case seen in the tests.
2. I also just made it skip the cancel() call if the stream is locked. Throwing the exception and ignoring it is just a waste of cycles.

Author: kentonv

.changeset/brave-donkeys-dress.md

@@ -0,0 +1,5 @@
+---
+"capnweb": minor
+---
+
+Added support for sending ReadableStream and WritableStream over RPC, with automatic flow control.

README.md

@@ -201,12 +201,12 @@ The following types can be passed over RPC (in arguments or return values), and
 * `Date`
 * `Uint8Array`
 * `Error` and its well-known subclasses
+* `ReadableStream` and `WritableStream`, with automatic flow control.
 
 The following types are not supported as of this writing, but may be added in the future:
 * `Map` and `Set`
 * `ArrayBuffer` and typed arrays other than `Uint8Array`
 * `RegExp`
-* `ReadableStream` and `WritableStream`, with automatic flow control.
 * `Headers`, `Request`, and `Response`
 
 The following are intentionally NOT supported:
@@ -305,6 +305,10 @@ The trick here is record-replay: On the calling side, Cap'n Web will invoke your
 
 Since all of the not-yet-determined values seen by the callback are represented as `RpcPromise`s, the callback's behavior is deterministic. Any actual computation (arithmetic, branching, etc.) can't possibly use these promises as (meaningful) inputs, so would logically produce the same results for every invocation of the callback. Any such computation will actually end up being performed on the sending side, just once, with the results being imbued into the recording.
 
+### Streaming with flow control
+
+You may pass a `ReadableStream` or `WritableStream` over RPC. When doing so, the RPC system automatically creates an equivalent stream at the other end and pumps bytes (or arbitrarily-typed chunks) across. This is done in such a way as to ensure the available bandwidth is fully utilized while minimizing buffer bloat, by observing the bandwidth-delay product and applying backpressure when too much is written. Multiple streams can be sent across the same connection -- they will be multiplexed appropriately, similar to HTTP/2 stream multiplexing.
+
 ### Cloudflare Workers RPC interoperability
 
 Cap'n Web works on any JavaScript platform. But, on Cloudflare Workers specifically, it's designed to play nicely with the [the built-in RPC system](https://blog.cloudflare.com/javascript-native-rpc/). The two have basically the same semantics, the only difference being that Workers RPC is a built-in API provided by the Workers Runtime, whereas Cap'n Web is implemented in pure JavaScript.

__tests__/flow-control.test.ts

@@ -0,0 +1,254 @@
+// Copyright (c) 2026 Cloudflare, Inc.
+// Licensed under the MIT license found in the LICENSE.txt file or at:
+//     https://opensource.org/license/mit
+
+import { expect, it, describe } from "vitest"
+import { FlowController, type SendToken } from "../src/streams.js"
+
+// To emulate random-ish chunk sizes while keeping the test reproducible, we cycle through this
+// list of sizes.
+const CHUNK_SIZES = [32 * 1024, 4 * 1024, 16000, 12345, 16, 9999, 4321, 8];
+
+// Helper: simulates a stream with a fake clock and set RTT and bandwidth. Sends chunks, advances
+// time, and acks in order.
+class StreamSimulator {
+  // Default RTT of 100ms.
+  rtt = 100;
+
+  // Default bandwidth of 10kB/ms = 10MB/s = 1MB/RTT. This is larger than the initial window size
+  // of 256k, so the window should grow if staurated. Individual test cases may set something
+  // different, or modify the properties during the test.
+  bandwidth = 10 * 1024;
+
+  // Convenience getter to calculate BDP.
+  get bdp() { return this.rtt * this.bandwidth; }
+
+  // Current simulated time.
+  t = 0;
+
+  // Are we currently blocked, according to the flow controller's return values from onSend/onAck?
+  blocked = false;
+
+  // The outgoing link is sending bytes (for a previous send()) until this time. Subsequent sends
+  // cannot start until after this time. This is how we simulate bandwidth constraints.
+  linkOccupiedUntil = 0;
+
+  // The flow controller itself.
+  fc = new FlowController(() => this.t);
+
+  // In-flight writes, in send order. Each has its token and scheduled ack time.
+  inFlight: { token: SendToken, ackTime: number }[] = [];
+
+  // Send a chunk at the current time.
+  send(size: number) {
+    // The new message begins sending now, unless some other message is still sending, in which
+    // case it sends after that message. It occupies the link until all the bytes are sent, based
+    // on the bandwidth.
+    this.linkOccupiedUntil = Math.max(this.linkOccupiedUntil, this.t) + size / this.bandwidth;
+
+    let { token, shouldBlock } = this.fc.onSend(size);
+
+    // ackTime = time when the chunk finishes writing out to the pipe, plus 1 rtt
+    this.inFlight.push({ token, ackTime: this.linkOccupiedUntil + this.rtt });
+    this.blocked = shouldBlock;
+  }
+
+  // Fill the window by sending chunks of the given size. Returns the number of chunks sent
+  // (the last one caused blocking).
+  fillWindow(chunkSize: number): number {
+    let count = 0;
+    while (!this.blocked) {
+      count++;
+      this.send(chunkSize);
+    }
+    return count;
+  }
+
+  // Advance to the next message's ack time, and deliver the ack to the flow controller.
+  waitForNextAck() {
+    let entry = this.inFlight.shift();
+    if (entry) {
+      this.t = entry.ackTime;
+      if (this.fc.onAck(entry.token)) {
+        this.blocked = false;
+      }
+    }
+  }
+
+  // Simulate the application writing to the stream as fast as it can for the given duration.
+  saturateFor(duration: number): void {
+    let endTime = this.t + duration;
+    let i = 0;
+
+    // Send chunks until blocked, then wait for acks until unblocked. Repeat until the end time
+    // is reached.
+    while (this.t < endTime) {
+      if (this.blocked) {
+        this.waitForNextAck();
+      } else {
+        this.send(CHUNK_SIZES[i++ % CHUNK_SIZES.length]);
+      }
+    }
+
+    // Drain remaining acks.
+    while (this.inFlight.length > 0) {
+      this.waitForNextAck();
+    }
+  }
+}
+
+describe("FlowController", () => {
+  it("blocks when window is full", () => {
+    let sim = new StreamSimulator();
+    let initialWindow = sim.fc.window;  // 256KB
+
+    // Send chunks until blocked.
+    let count = sim.fillWindow(64 * 1024);
+    expect(count).toBe(4);  // 4 * 64KB = 256KB = window
+    expect(sim.fc.bytesInFlight).toBe(initialWindow);
+  });
+
+  it("unblocks after ack frees space", () => {
+    let sim = new StreamSimulator();
+
+    // Send 4 chunks at staggered times so acks arrive at different times.
+    sim.t = 0;
+    sim.send(64 * 1024);
+    sim.t = 1;
+    sim.send(64 * 1024);
+    sim.t = 2;
+    sim.send(64 * 1024);
+    sim.t = 3;
+    expect(sim.blocked).toBe(false);
+    sim.send(64 * 1024);
+    expect(sim.blocked).toBe(true);
+
+    // Ack only the first one.
+    sim.waitForNextAck();
+    expect(sim.fc.bytesInFlight).toBe(192 * 1024);  // 3 chunks still in flight
+  });
+
+  it("window grows during startup", () => {
+    let sim = new StreamSimulator();
+
+    let initialWindow = sim.fc.window;  // 256KB
+    // Simulate for a few RTTs — window should grow toward the BDP.
+    sim.saturateFor(sim.rtt * 5);
+
+    // During startup, window should quickly grow past the 1MB BDP.
+    expect(sim.fc.window).toBeGreaterThan(initialWindow);
+    expect(sim.fc.window).toBeGreaterThan(sim.bdp);
+  });
+
+  it("exits startup after window growth plateaus", () => {
+    let sim = new StreamSimulator();
+
+    expect(sim.fc.inStartupPhase).toBe(true);
+
+    // Simulate for long enough that startup exits.
+    sim.saturateFor(sim.rtt * 50);
+
+    expect(sim.fc.inStartupPhase).toBe(false);
+  });
+
+  it("steady-state window converges near BDP", () => {
+    let sim = new StreamSimulator();
+
+    // Run through startup.
+    sim.saturateFor(sim.rtt * 50);
+    expect(sim.fc.inStartupPhase).toBe(false);
+
+    // Window should have settled around 25% above BDP.
+    expect(sim.fc.window).toBeGreaterThan(sim.bdp * 1.2);
+    expect(sim.fc.window).toBeLessThan(sim.bdp * 1.3);
+
+    // Keep running.
+    sim.saturateFor(sim.rtt * 20);
+
+    // Window should be pretty stable.
+    expect(sim.fc.window).toBeGreaterThan(sim.bdp * 1.2);
+    expect(sim.fc.window).toBeLessThan(sim.bdp * 1.3);
+  });
+
+  it("window does not shrink when app-limited", () => {
+    let sim = new StreamSimulator();
+
+    // Run through startup to establish a window.
+    sim.saturateFor(sim.rtt * 50);
+    let windowAfterStartup = sim.fc.window;
+
+    // Now repeatedly send small chunks that don't fill the window, waiting for each one.
+    for (let i = 0; i < 50; i++) {
+      sim.send(1024);
+      sim.waitForNextAck();
+    }
+
+    // Window should not have shrunk.
+    expect(sim.fc.window).toStrictEqual(windowAfterStartup);
+  });
+
+  it("window shrinks when pipe bandwidth decreases", () => {
+    let sim = new StreamSimulator();
+
+    // Run through startup at full speed — window should grow well above INITIAL_WINDOW.
+    sim.saturateFor(sim.rtt * 50);
+    expect(sim.fc.inStartupPhase).toBe(false);
+    expect(sim.fc.window).toBeGreaterThan(sim.bdp);
+
+    // Simulate pipe bandwidth drop to 1/4 of original. The sender still tries to fill the
+    // window, but acks arrive more slowly (bottleneck delivers chunks at 1/4 the rate).
+    sim.bandwidth /= 4;
+    sim.saturateFor(sim.rtt * 200);
+
+    // The window should have decayed, and is now closer to the new, lower bdp.
+    expect(sim.fc.window).toBeLessThan(sim.bdp * 2);
+  });
+
+  it("onError restores bytesInFlight without changing window", () => {
+    let sim = new StreamSimulator();
+
+    let { token } = sim.fc.onSend(64 * 1024);
+    expect(sim.fc.bytesInFlight).toBe(64 * 1024);
+
+    let windowBefore = sim.fc.window;
+    sim.fc.onError(token);
+    expect(sim.fc.bytesInFlight).toBe(0);
+    expect(sim.fc.window).toBe(windowBefore);
+  });
+
+  it("growth collar limits window increase to growthFactor per RTT", () => {
+    let sim = new StreamSimulator();
+    let chunkSize = 64 * 1024;
+
+    // Do a first round to get past the first-ack bootstrap.
+    sim.fillWindow(chunkSize);
+    sim.waitForNextAck();
+
+    // Record the initial window.
+    let windowBefore = sim.fc.window;
+
+    // Do another round.
+    sim.fillWindow(chunkSize);
+    // Ack all of them.
+    while (sim.inFlight.length > 0) {
+      sim.waitForNextAck();
+    }
+
+    // In startup, window can at most double per RTT (STARTUP_GROWTH_FACTOR = 2).
+    expect(sim.fc.window).toBeLessThanOrEqual(windowBefore * 2 + 1);
+  });
+
+  it("minimum window is enforced", () => {
+    let sim = new StreamSimulator();
+
+    // Set RTT and bandwidth excessively low, to create a low BDP.
+    sim.rtt = 1;
+    sim.bandwidth = 1;
+
+    // We're going to have to run for quite a while since bandwidth is so low.
+    sim.saturateFor(10000000);
+
+    // Window should have been clamped at MIN_WINDOW.
+    expect(sim.fc.window).toStrictEqual(64 * 1024);
+  });
+});