SQC-871 Implements socket serialization for transferring socket over RPC by Ltadrian · Pull Request #6863 · cloudflare/workerd

Ltadrian · 2026-07-06T20:02:16Z

Adds TransferredSocketStream class that handles implementing underlying AsyncIoStream IO operations for deserializing socket
Socket deserialization unwraps raw kj streams and builds non-deferred stream to pass into TransferredSocketStream
Adds js-rpc-socket-test tests
Add outbound interceptor test to showcase full end to end use case of loopback binding

RPC - Adds TransferredSocketStream class that handles implementing underlying AsyncIoStream IO operations - Socket serialization/deserialzation unwraps raw kj streams and builds non-deferred sockets - Fix and assert existing test that returns socket over RPC and can read bytes - Add outbound interceptor test to showcase returning socket over RPC

jasnell · 2026-07-07T21:52:42Z

+  // Serialize the readable and writable streams as separate externals
+  readable->serialize(js, serializer);
+
+  writable->serialize(js, serializer);


Be aware the the JsReadableStream and JsWritableStream work necessarily will impact this. JsReadableStream has already landed. readable is no longer a jsg::Ref<ReadableStream> but is wrapped by a JsReadableStream. The serialization story is still a work in progress.

Thank you I wasn't aware of JsReadableStream before working on this. I've caught up on the implementation in the internal repo workerd and it looks like it hopefully be too much of a lift to refactor some of these parts. I'll keep my eye out for when this lands in public repo 😊 Do you think it would be a blocker to support both JsReadableStream/JsWritableStream before landing this socket serialization?

jasnell · 2026-07-07T21:53:31Z

+  // JsRpcCustomEvent is canceled), so their pumpTo() must not be deferred past the context's
+  // lifetime.
+  sysStreams.readable = newNoDeferredProxyReadableStream(kj::mv(sysStreams.readable), ioContext);
+  auto readable = js.alloc<ReadableStream>(ioContext, kj::mv(sysStreams.readable));


With JsReadableStream already landed, this would become JsReadableStream::create(...)

(JsReadableStream is not yet in the public repo... should arrive there soon)

jasnell · 2026-07-07T21:53:49Z

+  // No opened-gate on the writable: the connection is already established (opened resolves below),
+  // and passing a gate is unnecessary here.
+  auto writable = js.alloc<WritableStream>(ioContext, kj::mv(sysStreams.writable),
+      ioContext.getMetrics().tryCreateWritableByteStreamObserver());


this would change once JsWritableStream lands

jasnell · 2026-07-07T22:06:33Z

A few questions...

How does socket upgrade (startTLS) work with a serialized socket?
It looks like isDefaultFetchPort is not set?
There doesn't seem to be any handling of opened state. A socket uses the openedPromise to signal when the socket is actually opened/established. From the looks of it here, it would be possible to serialize a still not-yet-open socket but the receiving side would appear to be opened? It's not clear.
What if the socket is already being closed? Or is already closed?

dom96

Yeah, like James mentioned, I'd repeat that currently as things stand there is a need for more synchronisation with the various socket states (closed/open). But perhaps that's planned since this is still a draft :)

In general though I think this is a good start. I left a few comments about various things I'd improve as well, but nothing major.

dom96 · 2026-07-08T15:50:57Z

+      // echo handler, so bytes we write must come back unchanged. This proves both socket.writable
+      // and socket.readable survive the RPC transfer.


What about other socket operations like close() and startTls?

Also tests for allowHalfOpen would be good

dom96 · 2026-07-08T15:59:47Z

+    // NOTE: For an RPC-transferred stream this reliably fires only on hard RPC breakage (the
+    // underlying CapnpToKjStreamAdapter rejects). A clean peer-initiated close is instead surfaced
+    // via readable EOF, which setupSocket() turns into a close when allowHalfOpen is false.
+    if (output.get() == nullptr) return kj::NEVER_DONE;


Why would output be null here. Also should it be a kj::Maybe if it being null is a possibility?

dom96 · 2026-07-08T16:05:12Z

+      [this, remoteAddr = kj::str(remoteAddress)](rpc::JsValue::External::Builder builder) mutable {
+    auto socket = builder.initSocket();
+    socket.setRemoteAddress(remoteAddr);
+    socket.setSecureTransport(getSecureTransport());


Should this also include other metadata like the state of allowHalfOpen, opened, closed?

dom96 · 2026-07-08T16:06:17Z

+  // Read the externals in the same order Socket::serialize() wrote them: (1) socket metadata,
+  // (2) the readable stream, (3) the writable stream. The stream externals are consumed here
+  // directly (rather than via ReadableStream/WritableStream::deserialize) so that we recover the
+  // raw kj half-streams and can rebuild a real AsyncIoStream backing the Socket.


Nit: the order here probably doesn't matter, does it?

dom96 · 2026-07-08T16:07:51Z

+  auto remoteAddr = kj::str(socketData.getRemoteAddress());
+  auto transport = kj::str(socketData.getSecureTransport());
+
+  // (2) Readable side: recover the raw input stream from the pushed ByteStream (mirrors


Could ReadableStream::deserialize be reused here to avoid duplication? Same for WritableStream::deserialize below.

dom96 · 2026-07-08T16:09:26Z

+
+  // Parse the secure transport
+  SecureTransportKind secureTransport = SecureTransportKind::OFF;
+  if (transport == "on") {


Why is this a string? Can't we encode it as an enum?

dom96 · 2026-07-08T16:11:21Z

+  kj::Own<kj::AsyncIoStream> asyncIoStream =
+      kj::heap<TransferredSocketStream>(kj::mv(input), kj::mv(output));
+  auto refcountedConnection = kj::refcountedWrapper(kj::mv(asyncIoStream));
+  auto sysStreams = newSystemMultiStream(*refcountedConnection, ioContext);


This code and the code below shares a lot with setupSocket. Could you reuse that function to avoid as much duplication as possible?

dom96 · 2026-07-08T16:11:50Z

  blob @15;
  # A Blob, serialized as its MIME type followed by its raw bytes.
+
+  socket @16;


doc comment

Ltadrian changed the title ~~Implements socket serialization for transferring socket over RPC~~ SQC-871 Implements socket serialization for transferring socket over RPC Jul 6, 2026

jasnell reviewed Jul 7, 2026

View reviewed changes

Comment thread src/workerd/api/streams/readable.h

dom96 reviewed Jul 8, 2026

View reviewed changes

		// echo handler, so bytes we write must come back unchanged. This proves both socket.writable
		// and socket.readable survive the RPC transfer.

Uh oh!

Conversation

Ltadrian commented Jul 6, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jasnell Jul 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

jasnell commented Jul 7, 2026

Uh oh!

dom96 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dom96 Jul 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

jasnell Jul 7, 2026 •

edited

Loading

dom96 Jul 8, 2026 •

edited

Loading