SQC-871 Implements socket serialization for transferring socket over RPC

src/workerd/api/sockets.h

@@ -151,6 +151,22 @@ class Socket: public jsg::Object {
   void handleReadableEof(jsg::Lock& js, jsg::Promise<void> onEof);
   // Sets up relevant callbacks to handle the case when the readable stream reaches EOF.
 
+  // Resolves the `closed` promise when `disconnected` (from watchForDisconnect()) reports a
+  // disconnect. This attaches jsg `.then()` continuations, so it must run in a JS-executing context:
+  // directly in setupSocket(), or deferred to a microtask in deserialize() (which runs under a scope
+  // that forbids JS execution).
+  void wireClosedToDisconnect(jsg::Lock& js, kj::Promise<bool> disconnected);
+
+  // Observes the `opened` promise and records its settled state in `openedState`. Must be called
+  // after allocation (it uses JSG_THIS, which requires an initialized refcount). This lets
+  // serialize() reject transfers of sockets that haven't finished connecting.
+  void trackOpenedState(jsg::Lock& js);
+
+  // RPC serialization support
+  void serialize(jsg::Lock& js, jsg::Serializer& serializer);
+  static jsg::Ref<Socket> deserialize(
+      jsg::Lock& js, rpc::SerializationTag tag, jsg::Deserializer& deserializer);
+
   JSG_RESOURCE_TYPE(Socket) {
     JSG_READONLY_PROTOTYPE_PROPERTY(readable, getReadable);
     JSG_READONLY_PROTOTYPE_PROPERTY(writable, getWritable);
@@ -166,6 +182,8 @@ class Socket: public jsg::Object {
     });
   }
 
+  JSG_SERIALIZABLE(rpc::SerializationTag::SOCKET);
+
   void visitForMemoryInfo(jsg::MemoryTracker& tracker) const {
     tracker.trackFieldWithSize("connectionData", sizeof(IoOwn<ConnectionData>));
     tracker.trackField("readable", readable);
@@ -222,6 +240,12 @@ class Socket: public jsg::Object {
   // Used to keep track of a pending `close` operation on the socket.
   bool isClosing = false;
 
+  // Tracks the settled state of `openedPromise`. A Socket can only be serialized for RPC transfer
+  // once its connection has been established (OPENED); serializing while still PENDING or after a
+  // FAILED connection throws, since serialize() is synchronous and cannot await `opened`.
+  enum class OpenedState : uint8_t { PENDING, OPENED, FAILED };
+  OpenedState openedState = OpenedState::PENDING;
+
   kj::Promise<kj::Own<kj::AsyncIoStream>> processConnection();
   jsg::Promise<void> maybeCloseWriteSide(jsg::Lock& js);
   jsg::Promise<void> closeImplOld(jsg::Lock& js);

src/workerd/api/streams/readable.c++

@@ -678,6 +678,11 @@ class NoDeferredProxyReadableStream final: public ReadableStreamSource {
 
 }  // namespace
 
+kj::Own<ReadableStreamSource> newNoDeferredProxyReadableStream(
+    kj::Own<ReadableStreamSource> inner, IoContext& context) {
+  return kj::heap<NoDeferredProxyReadableStream>(kj::mv(inner), context);
+}
+
 void ReadableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) {
   // Serialize by effectively creating a `JsRpcStub` around this object and serializing that.
   // Except we don't actually want to do _exactly_ that, because we do not want to actually create
@@ -751,8 +756,8 @@ jsg::Ref<ReadableStream> ReadableStream::deserialize(
 
   kj::Own<kj::AsyncInputStream> in = ioctx.getExternalPusher()->unwrapStream(rs.getStream());
 
-  return js.alloc<ReadableStream>(ioctx,
-      kj::heap<NoDeferredProxyReadableStream>(newSystemStream(kj::mv(in), encoding, ioctx), ioctx));
+  return js.alloc<ReadableStream>(
+      ioctx, newNoDeferredProxyReadableStream(newSystemStream(kj::mv(in), encoding, ioctx), ioctx));
 }
 
 kj::StringPtr ReaderImpl::jsgGetMemoryName() const {

src/workerd/api/streams/readable.h

@@ -555,4 +555,11 @@ class CountQueuingStrategy: public jsg::Object {
   QueuingStrategyInit init;
 };
 
+// Wraps a ReadableStreamSource so that pumpTo() is never deferred past the IoContext's lifetime.
+// Needed for RPC/session-bound sources (e.g. a Socket transferred over RPC), whose backing stream
+// disconnects when the IoContext is destroyed (the JsRpcCustomEvent is canceled). See the
+// implementation in readable.c++ for details.
+kj::Own<ReadableStreamSource> newNoDeferredProxyReadableStream(
+    kj::Own<ReadableStreamSource> inner, IoContext& context);
+
 }  // namespace workerd::api

src/workerd/api/tests/BUILD.bazel

@@ -851,6 +851,12 @@ wd_test(
     tags = ["resources:socket:1"],
 )
 
+wd_test(
+    src = "outbound-interceptor-socket-test.wd-test",
+    args = ["--experimental"],
+    data = ["outbound-interceptor-socket-test.js"],
+)
+
 wd_test(
     src = "http-test-ts.ts-wd-test",
     args = ["--experimental"],

src/workerd/api/tests/js-rpc-socket-test.wd-test

@@ -32,6 +32,9 @@ const unitTests :Workerd.Config = (
           (name = "defaultExport", service = "default-loop"),
           (name = "twelve", json = "12"),
           (name = "GreeterFactory", service = "GreeterFactory-loop"),
+          # Only present in the socket (loopback) variant: gates the socket-transfer assertions in
+          # js-rpc-test.js, which require a real RPC boundary to keep the producer context alive.
+          (name = "socketTransfer", json = "true"),
         ],
 
         durableObjectNamespaces = [

src/workerd/api/tests/js-rpc-test.js

@@ -282,6 +282,25 @@ export class MyService extends WorkerEntrypoint {
     return func;
   }
 
+  async getSocket() {
+    // Connect to our own connect() handler (which writes "hello" and closes) and return the client
+    // socket. Returning it over RPC transfers the socket to the caller, which can then read the
+    // bytes back through the reconstructed (transferred) socket.
+    const socket = this.env.MyService.connect('localhost:1234');
+    // A Socket can only be serialized for RPC once its connection has been established; await
+    // `opened` before returning it (Socket::serialize() is synchronous and cannot await this).
+    await socket.opened;
+    return socket;
+  }
+
+  async getEchoSocket() {
+    // Connect to the default entrypoint's echo connect() handler and return the client socket, so
+    // the caller can transfer it over RPC and exercise a write→peer→read round-trip.
+    const socket = this.env.defaultExport.connect('localhost:1234');
+    await socket.opened;
+    return socket;
+  }
+
   getRpcPromise(callback) {
     return callback();
   }
@@ -607,6 +626,12 @@ export default class DefaultService extends WorkerEntrypoint {
     // Test this.env here just to prove omitting the constructor entirely works.
     return new Response('default service ' + this.env.twelve);
   }
+
+  // Echoes everything written to the socket back to the reader. Used by getEchoSocket() so the
+  // socket-transfer test can exercise a write→peer→read round-trip through a transferred socket.
+  async connect(socket) {
+    await socket.readable.pipeTo(socket.writable);
+  }
 }
 
 export let basicServiceBinding = {
@@ -878,6 +903,53 @@ export let namedServiceBinding = {
         'Could not serialize object of type "Object". This type does not support ' +
         'serialization.',
     });
+    // Socket transfer over RPC is only exercised in the socket (loopback) variant of this test.
+    // The in-process variant can't sustain it: once getSocket() returns, the producer's request
+    // context tears down (there is no persistent RPC session to hold it open), so the byte pump
+    // feeding the transferred readable stops. The `socketTransfer` marker binding is present only in
+    // js-rpc-socket-test.wd-test.
+    if (env.socketTransfer) {
+      // A Socket returned over RPC is transferred to the caller. Verify a real byte round-trip
+      // through the reconstructed (transferred) socket: getSocket() connects to our connect()
+      // handler which writes "hello" and closes, so reading to EOF must yield "hello".
+      {
+        const socket = await env.MyService.getSocket();
+
+        // The `opened` metadata must survive the transfer: getSocket() connected to
+        // 'localhost:1234', so the deserialized socket should report that as its remote address.
+        const info = await socket.opened;
+        assert.strictEqual(info.remoteAddress, 'localhost:1234');
+
+        const dec = new TextDecoder();
+        let result = '';
+        for await (const chunk of socket.readable) {
+          result += dec.decode(chunk, { stream: true });
+        }
+        result += dec.decode();
+        assert.strictEqual(result, 'hello');
+      }
+
+      // Exercise the write direction through a transferred socket: getEchoSocket() connects to an
+      // echo handler, so bytes we write must come back unchanged. This proves both socket.writable
+      // and socket.readable survive the RPC transfer.
+      {
+        const socket = await env.MyService.getEchoSocket();
+        await socket.opened;
+
+        const enc = new TextEncoder();
+        const dec = new TextDecoder();
+        const writer = socket.writable.getWriter();
+        await writer.write(enc.encode('ping'));
+        await writer.close();
+
+        let result = '';
+        for await (const chunk of socket.readable) {
+          result += dec.decode(chunk, { stream: true });
+        }
+        result += dec.decode();
+        assert.strictEqual(result, 'ping');
+      }
+    }
 
     // A stateless entryponit method that never returns should fail due to PendingEvent tracking.
     await assert.rejects(() => env.MyService.neverReturn(), {

src/workerd/api/tests/outbound-interceptor-socket-test.js

@@ -0,0 +1,167 @@
+// Copyright (c) 2026 Cloudflare, Inc.
+// Licensed under the Apache 2.0 license found in the LICENSE file or at:
+//     https://opensource.org/licenses/Apache-2.0
+
+// Generic "outbound interceptor" socket-transfer test.
+//
+// This mirrors a proxy/interceptor service that, rather than connecting to a backend itself, obtains
+// a socket from a downstream service over JS RPC and returns it to its own caller. The returned
+// Socket is therefore transferred across TWO RPC hops (backend connect() handler -> interceptor ->
+// client), exercising Socket::serialize()/deserialize() at each hop.
+import { WorkerEntrypoint } from 'cloudflare:workers';
+import assert from 'node:assert';
+
+// Downstream backend, reached via `env.BACKEND.connect()`. Echoes everything written back to the
+// reader so the client can prove both directions of the transferred socket survive.
+export class Backend extends WorkerEntrypoint {
+  async connect(socket) {
+    await socket.readable.pipeTo(socket.writable);
+  }
+}
+
+// Downstream backend that writes a marker and then half-closes: it closes only its write side (so
+// the client sees read EOF) while keeping its read side open by draining it. This lets the client
+// observe whether the transferred socket honors `allowHalfOpen` (i.e. whether the client auto-closes
+// its own write side once its read side hits EOF).
+export class HalfCloseBackend extends WorkerEntrypoint {
+  async connect(socket) {
+    const writer = socket.writable.getWriter();
+    await writer.write(new TextEncoder().encode('bye'));
+    await writer.close();
+    // Keep the read side open, draining anything the client sends until it closes its write side.
+    for await (const _chunk of socket.readable) {
+      // discard
+    }
+  }
+}
+
+// The interceptor. Mirrors the outbound-interceptor structure (a `host` check that routes "magic"
+// addresses to a specific backend), but returns the backend's socket over RPC rather than piping it
+// -- which is the pattern whose lifetime we want to validate.
+export class OutboundInterceptor extends WorkerEntrypoint {
+  async interceptConnect(host) {
+    // `this.env.BACKEND.connect(host)` does NOT call Backend's `connect(socket)` handler as an
+    // RPC method. `connect` is a reserved name on a service binding (Fetcher): this invokes the
+    // built-in client-side socket API `Fetcher::connect` (C++: `jsg::Ref<Socket> connect(...)` in
+    // api/http.{h,c++} -> connectImpl in api/sockets.c++), which RETURNS the *client* end of a new
+    // connection to the BACKEND service. workerd then dispatches the *server* end of that same
+    // connection to Backend's exported `connect(socket)` ingress handler (the echo above). So there
+    // are two ends of one pipe: what we return here is the client end.
+    const socket = this.env.BACKEND.connect(host);
+    // A Socket can only be serialized for RPC once its connection has been established, so we must
+    // await `opened` before returning it. Socket::serialize() is synchronous and cannot await this
+    // itself; transferring a still-connecting socket throws a DataCloneError.
+    await socket.opened;
+    return socket;
+  }
+
+  // Returns a socket WITHOUT awaiting `opened` first. Serializing it for the RPC response must throw
+  // a DataCloneError, since the connection state (and SocketInfo) isn't yet settled.
+  interceptConnectWithoutAwait(host) {
+    return this.env.BACKEND.connect(host);
+  }
+
+  // Connects to the half-close backend with the given `allowHalfOpen` option and transfers the
+  // socket to the caller. The option must survive serialization so the transferred socket behaves
+  // the same as a locally-created one.
+  async interceptConnectHalfOpen(host, allowHalfOpen) {
+    const socket = this.env.HALFCLOSE.connect(host, { allowHalfOpen });
+    await socket.opened;
+    return socket;
+  }
+}
+
+async function readToEnd(socket) {
+  const dec = new TextDecoder();
+  let result = '';
+  for await (const chunk of socket.readable) {
+    result += dec.decode(chunk, { stream: true });
+  }
+  result += dec.decode();
+  return result;
+}
+
+async function echoRoundTrip(socket, payload) {
+  await socket.opened;
+  const enc = new TextEncoder();
+  const dec = new TextDecoder();
+  const writer = socket.writable.getWriter();
+  await writer.write(enc.encode(payload));
+  await writer.close();
+  let result = '';
+  for await (const chunk of socket.readable) {
+    result += dec.decode(chunk, { stream: true });
+  }
+  result += dec.decode();
+  return result;
+}
+
+export let interceptorForwardsSocket = {
+  async test(ctrl, env) {
+    // interceptConnect() returns a socket obtained from the backend over RPC (backend -> interceptor);
+    // awaiting it here transfers it a second hop (interceptor -> client). If the producer context is
+    // not kept alive past interceptConnect()'s return, the echo round-trip below would hang.
+    const socket = await env.INTERCEPTOR.interceptConnect('backend:5432');
+    const echoed = await echoRoundTrip(socket, 'ping');
+    assert.strictEqual(echoed, 'ping');
+  },
+};
+
+export let transferringUnopenedSocketThrows = {
+  async test(ctrl, env) {
+    // A socket that hasn't finished connecting cannot be serialized for RPC. The interceptor returns
+    // one without awaiting `opened`, so serializing the RPC response must fail with a DataCloneError.
+    await assert.rejects(
+      env.INTERCEPTOR.interceptConnectWithoutAwait('backend:5432'),
+      {
+        name: 'DataCloneError',
+      }
+    );
+  },
+};
+
+export let transferredSocketHonorsAllowHalfOpenFalse = {
+  async test(ctrl, env) {
+    // With allowHalfOpen=false (the default), a transferred socket must auto-close its write side
+    // once the read side reaches EOF. The backend half-closes after sending "bye", so reading to EOF
+    // triggers the auto-close, which resolves `closed`.
+    const socket = await env.INTERCEPTOR.interceptConnectHalfOpen(
+      'halfclose:1',
+      false
+    );
+    assert.strictEqual(await readToEnd(socket), 'bye');
+    // If allowHalfOpen were not carried across transfer, the write side would stay open and `closed`
+    // would hang here.
+    await socket.closed;
+  },
+};
+
+export let transferredSocketHonorsAllowHalfOpenTrue = {
+  async test(ctrl, env) {
+    // With allowHalfOpen=true, the write side must stay open after the read side reaches EOF, so we
+    // can still write to the (still-reading) backend afterwards.
+    const socket = await env.INTERCEPTOR.interceptConnectHalfOpen(
+      'halfclose:1',
+      true
+    );
+    assert.strictEqual(await readToEnd(socket), 'bye');
+    // The write side must not have been auto-closed; writing after read EOF must succeed. If the
+    // transferred socket wrongly defaulted to allowHalfOpen=false, this write would throw.
+    const writer = socket.writable.getWriter();
+    await writer.write(new TextEncoder().encode('still-open'));
+    await writer.close();
+  },
+};
+
+export let transferredSocketClosedResolvesOnExplicitClose = {
+  async test(ctrl, env) {
+    // A transferred socket must still support explicit close(): calling it resolves `closed`.
+    // allowHalfOpen=true so nothing auto-closes the socket for us -- only the explicit close() does.
+    const socket = await env.INTERCEPTOR.interceptConnectHalfOpen(
+      'halfclose:1',
+      true
+    );
+    await socket.close();
+    await socket.closed;
+  },
+};