feat: add Blob serialization support over RPC

Encode Blob as ["blob", contentType, ["readable", pipeId]] on the wire,
mirroring the existing Firefox Request body technique so devaluateImpl()
stays synchronous and e-order is preserved. On the receive side, bytes
are collected from the pipe stream via a new LocatedBlobPromise mechanism
in RpcPayload.deliverTo(), substituting the real Blob before user code
runs — analogous to how LocatedPromise handles RpcPromise substitution.

## Core changes

- src/core.ts: add "blob" TypeForRpc, BLOB_PROTOTYPE detection,
  LocatedBlobPromise type, RpcPayload blobPromises field, forEvaluate(),
  deliverTo(), dispose(), deepCopy(), disposeImpl(),
  ignoreUnhandledRejectionsImpl(), and followPath() coverage
- src/serialize.ts: streamToBlob() helper, Devaluator encode case,
  Evaluator decode case with sync fast-path for bytes/string content
  and async LocatedBlobPromise path for ReadableStream content
- src/types.d.ts: add Blob to BaseType
- .changeset/blob-rpc-support.md: minor bump changeset

## Tests

- blob serialization: decode-only unit tests via deserialize() covering
  bytes content, string content, empty blob, malformed wire values, and
  unsupported content expression types
- Blob over RPC: full round-trip tests including binary data, MIME type
  preservation, empty blob, compound return values, multiple blobs in
  one call, blob array returns, no MIME type, all 256 byte values,
  1 MB large blob, loopback (local) stub, blob alongside ReadableStream,
  dispose without reading, and an e-order regression test verifying that
  blob call messages are dispatched synchronously

Author: G4brym

.changeset/blob-rpc-support.md

@@ -0,0 +1,5 @@
+---
+"capnweb": minor
+---
+
+Add `Blob` as a serializable type over RPC. `Blob` objects can now be passed as call arguments and return values. The MIME type (`blob.type`) is preserved across the wire. `File` is not yet supported.

__tests__/index.test.ts

@@ -173,6 +173,56 @@ describe("simple serialization", () => {
 
 // =======================================================================================
 
+describe("blob serialization", () => {
+  it("can deserialize blob with bytes content", async () => {
+    // "hello!" base64-encoded without padding
+    let blob = deserialize('["blob","text/plain",["bytes","aGVsbG8h"]]') as Blob;
+    expect(blob).toBeInstanceOf(Blob);
+    expect(blob.type).toBe("text/plain");
+    expect(await blob.text()).toBe("hello!");
+  });
+
+  it("can deserialize blob with string content", async () => {
+    let blob = deserialize('["blob","text/html","<b>hi</b>"]') as Blob;
+    expect(blob).toBeInstanceOf(Blob);
+    expect(blob.type).toBe("text/html");
+    expect(await blob.text()).toBe("<b>hi</b>");
+  });
+
+  it("can deserialize empty blob", async () => {
+    let blob = deserialize('["blob","application/octet-stream",["bytes",""]]') as Blob;
+    expect(blob).toBeInstanceOf(Blob);
+    expect(blob.size).toBe(0);
+    expect(blob.type).toBe("application/octet-stream");
+  });
+
+  it("rejects malformed blob wire values", () => {
+    expect(() => deserialize('["blob"]')).toThrowError();
+    expect(() => deserialize('["blob",123,["bytes",""]]')).toThrowError();
+    // Missing content argument (length < 3)
+    expect(() => deserialize('["blob","text/plain"]')).toThrowError();
+  });
+
+  it("rejects blob with unsupported content expression type", () => {
+    // Content expression evaluates to a Date — not string, Uint8Array, or ReadableStream.
+    expect(() => deserialize('["blob","text/plain",["date",12345]]')).toThrowError(
+        /unknown special value/
+    );
+    // Content expression evaluates to a number.
+    expect(() => deserialize('["blob","text/plain",42]')).toThrowError(
+        /unknown special value/
+    );
+  });
+
+  it("throws when serializing Blob without an RPC session", () => {
+    if (typeof Blob === "undefined") return; // skip if Blob not available
+    let blob = new Blob(["hello"], {type: "text/plain"});
+    expect(() => serialize(blob)).toThrowError("Cannot create pipes without an RPC session");
+  });
+});
+
+// =======================================================================================
+
 class TestTransport implements RpcTransport {
   constructor(public name: string, private partner?: TestTransport) {
     if (partner) {
@@ -2444,3 +2494,238 @@ describe("Fetch API types over RPC", () => {
     expect(result.hasBody).toBe(false);
   });
 });
+
+// =======================================================================================
+
+describe("Blob over RPC", () => {
+  it("can send and receive a binary Blob", async () => {
+    await using harness = new TestHarness(new TestTarget());
+    let bytes = new TextEncoder().encode("hello from blob");
+    let blob = new Blob([bytes], {type: "application/octet-stream"});
+    using result = await harness.stub.echoBlob(blob);
+    expect(result).toBeInstanceOf(Blob);
+    expect(result.type).toBe("application/octet-stream");
+    expect(new Uint8Array(await result.arrayBuffer())).toStrictEqual(bytes);
+  });
+
+  it("preserves Blob MIME type", async () => {
+    await using harness = new TestHarness(new TestTarget());
+    let blob = new Blob(["<h1>hello</h1>"], {type: "text/html; charset=utf-8"});
+    using result = await harness.stub.echoBlob(blob);
+    expect(result.type).toBe("text/html; charset=utf-8");
+    expect(await result.text()).toBe("<h1>hello</h1>");
+  });
+
+  it("can send an empty Blob", async () => {
+    await using harness = new TestHarness(new TestTarget());
+    let blob = new Blob([], {type: "application/octet-stream"});
+    using result = await harness.stub.echoBlob(blob);
+    expect(result).toBeInstanceOf(Blob);
+    expect(result.size).toBe(0);
+    expect(result.type).toBe("application/octet-stream");
+  });
+
+  it("can send a Blob as part of a compound return value", async () => {
+    class BlobServer extends RpcTarget {
+      makePayload() {
+        return {
+          name: "test.txt",
+          blob: new Blob(["file content"], {type: "text/plain"}),
+          size: 12,
+        };
+      }
+    }
+
+    await using harness = new TestHarness(new BlobServer());
+    let stub = harness.stub as any;
+    let result = await stub.makePayload();
+    expect(result.name).toBe("test.txt");
+    expect(result.blob).toBeInstanceOf(Blob);
+    expect(result.blob.type).toBe("text/plain");
+    expect(await result.blob.text()).toBe("file content");
+    expect(result.size).toBe(12);
+  });
+
+  it("can send multiple Blobs in the same call", async () => {
+    // Exercises the LocatedBlobPromise array with more than one entry — the deliverTo()
+    // loop must resolve all pending blob assemblies before dispatching to user code.
+    class BlobCombiner extends RpcTarget {
+      async concatenate(a: Blob, b: Blob) {
+        let [textA, textB] = await Promise.all([a.text(), b.text()]);
+        return `${textA}|${textB}`;
+      }
+    }
+
+    await using harness = new TestHarness(new BlobCombiner());
+    let stub = harness.stub as any;
+    let result = await stub.concatenate(
+      new Blob(["hello"], {type: "text/plain"}),
+      new Blob(["world"], {type: "text/plain"}),
+    );
+    expect(result).toBe("hello|world");
+  });
+
+  it("can receive an array of Blobs in one return value", async () => {
+    // Multiple LocatedBlobPromise entries produced from a single return value.
+    class BlobFactory extends RpcTarget {
+      makeBlobs() {
+        return [
+          new Blob(["first"],  {type: "text/plain"}),
+          new Blob(["second"], {type: "text/plain"}),
+          new Blob(["third"],  {type: "text/plain"}),
+        ];
+      }
+    }
+
+    await using harness = new TestHarness(new BlobFactory());
+    let stub = harness.stub as any;
+    let [b1, b2, b3] = await stub.makeBlobs();
+    expect(await b1.text()).toBe("first");
+    expect(await b2.text()).toBe("second");
+    expect(await b3.text()).toBe("third");
+  });
+
+  it("round-trips a Blob with no MIME type", async () => {
+    // new Blob([bytes]) leaves .type as "" — the empty string must survive the round-trip
+    // and not become undefined or null.
+    await using harness = new TestHarness(new TestTarget());
+    let bytes = new TextEncoder().encode("untyped content");
+    let blob = new Blob([bytes]);
+    expect(blob.type).toBe("");
+    using result = await harness.stub.echoBlob(blob);
+    expect(result.type).toBe("");
+    expect(new Uint8Array(await result.arrayBuffer())).toStrictEqual(bytes);
+  });
+
+  it("preserves every possible byte value through the pipe", async () => {
+    // All 256 possible byte values in a single Blob — verifies the pipe mechanism
+    // neither corrupts nor truncates any byte.
+    await using harness = new TestHarness(new TestTarget());
+    let bytes = new Uint8Array(256);
+    for (let i = 0; i < 256; i++) bytes[i] = i;
+    let blob = new Blob([bytes], {type: "application/octet-stream"});
+    using result = await harness.stub.echoBlob(blob);
+    expect(result.size).toBe(256);
+    expect(new Uint8Array(await result.arrayBuffer())).toStrictEqual(bytes);
+  });
+
+  it("can send a large Blob over RPC", async () => {
+    // 1 MB blob — exercises multi-chunk stream collection in streamToBlob().
+    // Timeout is raised because CI machines can be slow to pump 1 MB through the
+    // fake in-process transport (default 5 s is too tight on some runners).
+    // Skipped in workerd: the isolate drops its connection when a large in-process
+    // stream is pumped through it (infrastructure limit, not a code bug).
+    if (navigator.userAgent === "Cloudflare-Workers") return;
+    await using harness = new TestHarness(new TestTarget());
+    let size = 1024 * 1024;
+    let bytes = new Uint8Array(size);
+    for (let i = 0; i < size; i++) bytes[i] = i & 0xff;
+    let blob = new Blob([bytes], {type: "application/octet-stream"});
+    using result = await harness.stub.echoBlob(blob);
+    expect(result.size).toBe(size);
+    expect(new Uint8Array(await result.arrayBuffer())).toStrictEqual(bytes);
+  }, 30_000);
+
+  it("can pass a Blob through a local (loopback) stub", async () => {
+    // No network — payload goes through deepCopy() rather than the Evaluator.
+    // Blobs are immutable so deepCopy() returns them as-is; no blobPromises involved.
+    using stub = new RpcStub(new TestTarget());
+    let bytes = new TextEncoder().encode("loopback content");
+    let blob = new Blob([bytes], {type: "text/plain"});
+    let result = await stub.echoBlob(blob);
+    expect(result).toBeInstanceOf(Blob);
+    expect(result.type).toBe("text/plain");
+    expect(await result.text()).toBe("loopback content");
+    result[Symbol.dispose]();
+  });
+
+  it("can send a Blob alongside a ReadableStream in the same call", async () => {
+    // Both types use the pipe mechanism. Tests that blobPromises processing in
+    // deliverTo() does not interfere with ReadableStream hooks.
+    class Server extends RpcTarget {
+      async receiveAll(blob: Blob, stream: ReadableStream<string>) {
+        let reader = stream.getReader();
+        let chunks: string[] = [];
+        for (;;) {
+          let {done, value} = await reader.read();
+          if (done) break;
+          chunks.push(value);
+        }
+        return `${await blob.text()}+${chunks.join(",")}`;
+      }
+    }
+
+    let stream = new ReadableStream<string>({
+      start(controller) {
+        controller.enqueue("a");
+        controller.enqueue("b");
+        controller.close();
+      }
+    });
+
+    await using harness = new TestHarness(new Server());
+    let stub = harness.stub as any;
+    let result = await stub.receiveAll(new Blob(["hello"], {type: "text/plain"}), stream);
+    expect(result).toBe("hello+a,b");
+  });
+
+  it("disposing a result containing a Blob does not throw", async () => {
+    // Blobs have no owned resources; disposeImpl() must be a silent no-op.
+    class BlobServer extends RpcTarget {
+      makeBlob() { return new Blob(["hello"], {type: "text/plain"}); }
+    }
+
+    await using harness = new TestHarness(new BlobServer());
+    let stub = harness.stub as any;
+    let result = await stub.makeBlob();
+    expect(result).toBeInstanceOf(Blob);
+    // Dispose without reading — should never throw.
+    expect(() => result[Symbol.dispose]()).not.toThrow();
+  });
+
+  it("blob call messages are dispatched synchronously (e-order)", async () => {
+    // Core regression test for the Approach-B design decision: the ["push"] wire message
+    // for a blob-containing call must be dispatched in the same microtask turn as surrounding
+    // calls, not deferred until blob.arrayBuffer() resolves.
+    //
+    // We fence the server transport so messages accumulate without being processed, then
+    // verify all three call messages are already in the queue before any await.
+    class OrderServer extends RpcTarget {
+      mark(n: number) { return n; }
+      acceptBlob(n: number, _blob: Blob) { return n; }
+    }
+
+    let clientTransport = new TestTransport("client");
+    let serverTransport = new TestTransport("server", clientTransport);
+
+    let client = new RpcSession<OrderServer>(clientTransport);
+    let server = new RpcSession(serverTransport, new OrderServer());
+
+    // Prevent the server from processing any messages until we release the fence.
+    serverTransport.fence();
+
+    let stub = client.getRemoteMain();
+    let blob = new Blob([new Uint8Array(64)], {type: "application/octet-stream"});
+
+    // Dispatch all three calls synchronously in the same microtask turn.
+    let p1 = stub.mark(1);
+    let p2 = stub.acceptBlob(2, blob);
+    let p3 = stub.mark(3);
+
+    // TestTransport.send() has no await so queue.push() is synchronous — no microtask
+    // yield needed. But we yield once to be safe.
+    await Promise.resolve();
+
+    // Queue must contain at least 4 messages: ["push" mark 1], ["pipe"], ["push" blob call],
+    // ["push" mark 3]. If blob dispatch were deferred (Approach A without sync plumbing),
+    // the blob's ["push"] would be missing here.
+    expect(serverTransport.pendingCount).toBeGreaterThanOrEqual(4);
+
+    serverTransport.releaseFence();
+    await Promise.all([p1, p2, p3]);
+
+    // Dispose the main stub to release the import table entry.
+    stub[Symbol.dispose]();
+    await pumpMicrotasks();
+  });
+});

__tests__/test-util.ts

@@ -66,4 +66,9 @@ export class TestTarget extends RpcTarget {
   returnNull() { return null; }
   returnUndefined() { return undefined; }
   returnNumber(i: number) { return i; }
+
+  async echoBlob(blob: Blob): Promise<Blob> {
+    let bytes = await blob.arrayBuffer();
+    return new Blob([bytes], {type: blob.type});
+  }
 }