Add deferred WebSocket upgrade materialization (deferUpgradeMaterialization + materializeUpgrade)

.changeset/defer-upgrade-materialization.md

@@ -0,0 +1,7 @@
+---
+"@iterate-com/capnweb": minor
+---
+
+Tunneled WebSocket upgrade Responses now arrive in forwardable form by default on Cloudflare Workers: `Response.webSocket` deserializes to an opaque `DeferredWebSocketUpgrade` — a `{ readable, writable, init }` byte-stream pair — instead of an eagerly materialized `WebSocketPair` end. The pair can be carried across hops that serialize byte streams but not sockets (in particular native Workers RPC between isolates, whose serializer refuses a live WebSocket), and the new `materializeUpgrade(pair, init?)` export rebuilds a real upgrade `Response` at the hop that actually serves it; `pair.init` carries the provider's upgrade headers (e.g. a negotiated `Sec-WebSocket-Protocol`) so they survive to the served 101. On other runtimes the previous behavior (a usable `TunneledWebSocket`) remains the default. The new `deferUpgradeMaterialization` session option overrides the default in either direction — set it to `false` on Workers to restore eager materialization when the session endpoint itself serves the upgrade. Receive-side only: the wire format is unchanged.
+
+Behavior change on Workers only: code that received a tunneled upgrade over a capnweb session on workerd and used `response.webSocket` as a native socket at the session endpoint must either call `materializeUpgrade(response.webSocket)` or set `deferUpgradeMaterialization: false` on the session.

README.md

@@ -7,6 +7,12 @@
 > Delta vs upstream:
 > - **WebSocket-over-RPC** — a `Response` with a Workers-style `webSocket`
 >   upgrade can be passed over RPC (tunneled as a stream pair).
+> - **Deferred upgrade materialization** — on Cloudflare Workers a tunneled
+>   upgrade arrives by default as an opaque byte-stream pair that can cross
+>   hops which serialize byte streams but not sockets (e.g. native Workers
+>   RPC between isolates); `materializeUpgrade()` rebuilds the real socket at
+>   the hop that serves it. The `deferUpgradeMaterialization` session option
+>   overrides the default in either direction.
 > - **`onCall` session option** — server-side per-call hook for observability
 >   (used by Iterate OS ITX tracing); propagates through promise pipelining.
 > - Small `Provider` type tweak for better go-to-definition through stubs.

__tests__/test-server-workerd.js

@@ -10,7 +10,7 @@
 // build step for it. Instead, we're getting by configuring the worker in vitest.config.ts by
 // just specifying the raw JS modules.
 
-import { newWorkersRpcResponse } from "../dist/index-workers.js";
+import { newWorkersRpcResponse, newWebSocketRpcSession, materializeUpgrade } from "../dist/index-workers.js";
 import { RpcTarget, DurableObject } from "cloudflare:workers";
 
 // TODO(cleanup): At present we clone the implementation of Counter and TestTarget because
@@ -81,14 +81,104 @@ export class TestTarget extends RpcTarget {
   }
 }
 
+// See openDeferredEchoRemote below.
+let deferredEchoRetainer = null;
+
 export default {
   async fetch(req, env, ctx) {
+    // The design doc's "fetch-lane exit", end to end: defer a tunneled upgrade from an
+    // in-isolate capnweb session, materialize it, and serve it as a REAL HTTP upgrade from
+    // this fetch handler -- headers carried on pair.init and all.
+    if (new URL(req.url).pathname === "/deferred-upgrade") {
+      let origin = new WebSocketPair();
+      origin[1].accept();
+      origin[1].addEventListener("message", event => origin[1].send(event.data));
+
+      class Target extends RpcTarget {
+        openEcho() {
+          return new Response(null, {
+            status: 101,
+            webSocket: origin[0],
+            headers: { "sec-websocket-protocol": "itx-v1", "x-provider-custom": "survives" },
+          });
+        }
+      }
+
+      // (Holding a capnweb session in this request-scoped context triggers workerd
+      // hang-detector warnings after the test finishes; production holds sessions in a
+      // long-lived relay WebSocket context. Accepted as test-only noise.)
+      let pair = new WebSocketPair();
+      pair[0].accept();
+      pair[1].accept();
+      let api = newWebSocketRpcSession(pair[0]);  // deferred by default on Workers
+      newWebSocketRpcSession(pair[1], new Target());
+      let response = await api.openEcho();
+      return materializeUpgrade(response.webSocket);
+    }
+
     return newWorkersRpcResponse(req, new TestTarget(env), {
       onSendError(err) { return err; }
     });
   },
 
   async greet(name, env, ctx) {
     return `Hello, ${name}!`;
+  },
+
+  // The relay shape: THIS isolate holds a deferring capnweb session and returns the raw pair
+  // in a native RPC return payload; the caller materializes after the call settles. The session
+  // and the delivering Response must be retained past the call (module state) -- disposing them
+  // would release the payload-owned tunnel -- which is exactly the retention contract a real
+  // relay must follow.
+  async openDeferredEchoRemote(unused, env, ctx) {
+    let origin = new WebSocketPair();
+    origin[1].accept();
+    origin[1].addEventListener("message", event => origin[1].send(event.data));
+    let closeEvent = null;
+    origin[1].addEventListener("close",
+        event => { closeEvent = { code: event.code, reason: event.reason }; });
+
+    class Target extends RpcTarget {
+      openEcho() {
+        return new Response(null, {
+          status: 101,
+          webSocket: origin[0],
+          headers: { "sec-websocket-protocol": "itx-v1" },
+        });
+      }
+    }
+
+    // (Session held in an RPC-call context: same accepted hang-detector noise as the
+    // /deferred-upgrade route above.)
+    let pair = new WebSocketPair();
+    pair[0].accept();
+    pair[1].accept();
+    let api = newWebSocketRpcSession(pair[0]);  // deferred by default on Workers
+    newWebSocketRpcSession(pair[1], new Target());
+    let response = await api.openEcho();
+    deferredEchoRetainer = { api, response, getClose: () => closeEvent };
+    return response.webSocket;
+  },
+
+  async deferredEchoCloseEvent(unused, env, ctx) {
+    return deferredEchoRetainer ? deferredEchoRetainer.getClose() : null;
+  },
+
+  // Native workers-RPC leg of deferred upgrade materialization: receives a tunneled socket as a
+  // raw { readable, writable } pair (which workerd RPC serializes; a live WebSocket it refuses),
+  // materializes the socket in THIS isolate, and runs one echo round trip through it.
+  async materializeEcho(webSocket, env, ctx) {
+    let response = materializeUpgrade(webSocket);
+    let socket = response.webSocket;
+    socket.accept();
+    let reply = new Promise(resolve => {
+      socket.addEventListener("message", event => resolve(event.data), { once: true });
+    });
+    socket.send("ping across isolates");
+    try {
+      return await reply;
+    } finally {
+      socket.close(1000, "");
+    }
   }
 }