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.

Ltadrian · 2026-07-09T19:02:14Z

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?

@jasnell

In the case of Hyperdrive needs, socket upgrade isn't needed as Hyperdrive will handle TLS to the origin. In the generalized socket case I am not sure. I imagine that we would need to somehow allow the transferred socket to upgrade tls and start tls handshake but I would probably prefer if the socket has already been upgraded before transfer instead?
Updated to add isDefaultFetchPort to worker-interface.capnp and serialize/deserialize ✅
I thought about this for a while and I think throwing an error is fair if the socket is still pending/closed. I can't think of a good case where receiving a socket and needing to await on socket.opened would be a fair user experience. I would prefer that the socket is assumed to be opened and connected. I've gone ahead and implemented some checks on serialize to prevent serializing the socket when that state is pending/closed but fully open to other implementation ideas here
Throwing error on serialize feels fair to me. There is still the case of where on serialize the socket is opened, but during transit the connection closes and we assume the serialize socket info is still opened on deserialize. This should naturally resolve to closed but might show some disconnect errors? Let me know if this behavior is not preferred

Ltadrian · 2026-07-09T19:13:23Z

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

Thank you for the initial review, I have spent some extra time addressing the feedback and have added a second commit to this merge request to handle the various states I initially put on TODO: halfOpen, opened, closed

I'll re-review my second commit today again but this seems to be functional now

(opened/closed/allowHalfOpen/secureTransport) Carry the origin socket's runtime state over JS RPC and reconstruct equivalent behavior on the receiving side. - Wire format: add secureTransport (new enum), isDefaultFetchPort, localAddress, and allowHalfOpen to External.socket; fix the socket external doc comment. - opened: track settled state (OpenedState enum); serialize() throws DataCloneError unless OPENED; deserialize() resolves it from the transferred SocketInfo and marks the socket OPENED. - allowHalfOpen: carried across transfer; when false, re-establish the read-EOF -> write-side auto-close on the receiving side. - closed: extract shared disconnect wiring (used by setupSocket) and resolve `closed` on disconnect for transferred sockets. - deserialize scope: the EOF and closed wiring attach jsg `.then()` continuations (illegal under the no-JS deserialize scope), so defer them to microtasks; kj-side signals are latched synchronously so nothing is missed. - Tests: allowHalfOpen honored across transfer, unopened transfer throws, `closed` resolves on close(); js-rpc-test awaits `opened` before returning.

codspeed-hq · 2026-07-10T18:01:04Z

Merging this PR will improve performance by 9.77%

⚠️

Different runtime environments detected

Some benchmarks with significant performance changes were compared across different runtime environments,
which may affect the accuracy of the results.

Open the report in CodSpeed to investigate

⚡ 1 improved benchmark
✅ 71 untouched benchmarks
⏩ 129 skipped benchmarks¹

Performance Changes

	Benchmark	`BASE`	`HEAD`	Efficiency
⚡	`bodyWithHeaders[Response]`	34.2 µs	31.2 µs	+9.77%

Tip

Curious why this is faster? Comment @codspeedbot explain why this is faster on this PR, or directly use the CodSpeed MCP with your agent.

_{Comparing Ltadrian:dlapid/jsrpc_socket (a44dee4) with main (a77b7aa)}

129 benchmarks were skipped, so the baseline results were used instead. If they were deleted from the codebase, click here and archive them to remove them from the performance reports. ↩

dom96

Needs a test which calls startTls on a socket, otherwise LGTM. Not approving as I'd like James to take a look too.

jasnell · 2026-07-21T23:06:35Z

+    // ExplicitEndOutputStream needs an async end() to signal a clean EOF over the RPC ByteStream,
+    // but shutdownWrite() is synchronous. Kick off end() as an IoContext task so the clean-EOF RPC
+    // completes; dropping without end() would look like an abort to the peer. If there's no active
+    // IoContext (e.g. during teardown), just drop, which the ByteStream treats as an abort.


I would move the IoContext:current() check to before the destructive kj::mv that consumes the output

jasnell · 2026-07-21T23:07:44Z

+      output = kj::none;
+      if (IoContext::hasCurrent()) {
+        IoContext::current().addTask(
+            owned->end().attach(kj::mv(owned)).catch_([](kj::Exception&&) {}));


Not really a fan of swallowing the rejection case here. Should this at least log periodically?

jasnell · 2026-07-21T23:10:19Z

+  // whenResolved() creates a fresh branch off the `opened` promise without consuming
+  // `openedPromiseCopy` (which `close()` relies on). The branch settles in lockstep with `opened`.
+  openedPromiseCopy.whenResolved(js)
+      .then(js, [this, self = JSG_THIS](jsg::Lock&) { openedState = OpenedState::OPENED; },


Just capture self = JSG_THIS and use self->openedState = ...

jasnell · 2026-07-21T23:12:02Z


+void Socket::trackOpenedState(jsg::Lock& js) {
+  // whenResolved() creates a fresh branch off the `opened` promise without consuming
+  // `openedPromiseCopy` (which `close()` relies on). The branch settles in lockstep with `opened`.


"lockstep" is not entirely acurate. Since continuations run in registration order, if something else is following the openedPromiseCopy promise and is attached first, then that will run before the openedState = ... is set. We just need to verify there are no re-entrancy issues.

jasnell · 2026-07-21T23:12:59Z

+      .then(js, [this, self = JSG_THIS](jsg::Lock&) { openedState = OpenedState::OPENED; },
+          [this, self = JSG_THIS](jsg::Lock&, jsg::Value&&) {
+    openedState = OpenedState::FAILED;
+  }).markAsHandled(js);


You're attaching a catch handler and neither the success or failure cases can reject. The markAsHandled(js) here is likely unnecessary/redundant.

jasnell · 2026-07-21T23:13:23Z

+void Socket::wireClosedToDisconnect(jsg::Lock& js, kj::Promise<bool> disconnected) {
+  auto& context = IoContext::current();
+  context.awaitIo(js, kj::mv(disconnected))
+      .then(js, [this, self = JSG_THIS](jsg::Lock& js, bool canceled) {


Only capture self = JSG_THIS, not this

jasnell · 2026-07-21T23:16:15Z

+      break;
+  }
+
+  // Serialize the socket metadata, referencing the stream externals


Suggested change

// Serialize the socket metadata, referencing the stream externals

// Serialize the socket metadata, referencing the stream externals

// The call to write is synchronous, so capturing this is safe.

jasnell · 2026-07-21T23:19:45Z

+  js.v8Isolate->EnqueueMicrotask(js.wrapSimpleFunction(js.v8Context(),
+      JSG_VISITABLE_LAMBDA((self = socket.addRef(), disconnected = kj::mv(disconnectedOwn)), (self),
+          (jsg::Lock & js, const v8::FunctionCallbackInfo<v8::Value>& args) mutable {
+            self.get()->wireClosedToDisconnect(js, kj::mv(*disconnected));


Worth noting that if these throw JS exceptions, the error will be swallowed entirely by the EnqueueMicrotask

That can happen, for instance, if the microtask queue happens to be drained outside of an active IoContext, for instance. Using EnqueueMicrotask might not be the safest here as opposed to creating a resolved promise and attaching a continuation to it (which utilizes the cross-request promise resolve mechanism)

jasnell · 2026-07-21T23:23:35Z

+// the peer disconnects, or rejects on error.
+kj::Promise<void> handleDisconnected(
+    kj::AsyncIoStream& connection, kj::PromiseFulfiller<bool>& fulfiller) {
+  try {


Use KJ_TRY/KJ_CATCH

jasnell · 2026-07-21T23:25:17Z

+  auto deferredCancel = kj::defer([fulfiller = kj::mv(paf.fulfiller)]() mutable {
+    // The watch task was canceled without fulfilling the fulfiller (the Socket was GC'd); silently
+    // fulfill with true so wireClosedToDisconnect() knows not to resolve `closed`.
+    fulfiller->fulfill(true);


Should likely check to see if the fulfiller is still waiting first.

jasnell · 2026-07-21T23:26:44Z

+    // fulfill with true so wireClosedToDisconnect() knows not to resolve `closed`.
+    fulfiller->fulfill(true);
+  });
+  auto watchTask = handleDisconnected(connection, fulfiller).attach(kj::mv(deferredCancel));


Is this the only callsite for handleDisconnected? If it is, it's likely better to kj::mv the deferredCancel into the handleDisconnected(...) call rather than passing a bare reference and attaching the deferral.

jasnell · 2026-07-21T23:32:33Z


+kj::Own<ReadableStreamSource> newNoDeferredProxyReadableStream(
+    kj::Own<ReadableStreamSource> inner, IoContext& context) {
+  return kj::heap<NoDeferredProxyReadableStream>(kj::mv(inner), context);


Calling convention typically would put IoContext& context as the first argument.

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

Ltadrian force-pushed the dlapid/jsrpc_socket branch from b7082f5 to a44dee4 Compare July 9, 2026 19:25

Ltadrian marked this pull request as ready for review July 10, 2026 17:34

Ltadrian requested review from a team as code owners July 10, 2026 17:34

Ltadrian requested review from dom96 and jasnell July 20, 2026 15:39

dom96 reviewed Jul 21, 2026

View reviewed changes

jasnell reviewed Jul 21, 2026

View reviewed changes

	// Serialize the socket metadata, referencing the stream externals
	// Serialize the socket metadata, referencing the stream externals
	// The call to write is synchronous, so capturing this is safe.

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!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Ltadrian commented Jul 9, 2026

Uh oh!

Ltadrian commented Jul 9, 2026

Uh oh!

codspeed-hq Bot commented Jul 10, 2026

Merging this PR will improve performance by 9.77%

Performance Changes

Footnotes

Uh oh!

dom96 left a comment • 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!

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 left a comment •

edited

Loading