From 861ae70a8fe5c4ffcfd29794430f191c15e79d83 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 19 Aug 2026 14:01:58 -0700 Subject: [PATCH 01/14] Add ExternalHandler hook for serializing JS-implemented class instances Objects implemented in JavaScript (such as the TypeScript-implemented stream classes) have no JSG wrapper, so the serializer cannot recognize them the way it recognizes host objects. Give the ExternalHandler a trySerializeClassInstance() virtual, called from WriteHostObject() for class instances before falling through to DataCloneError, so a handler can recognize such objects by brand check and serialize them as externals. The default implementation declines, preserving existing behavior. Relatedly, let DeserializeInvoker accept deserialize() implementations whose return type wraps to a generic v8::Local (custom wrappers like JsReadableStream), enforcing at runtime that the wrapped result is an object. --- src/workerd/jsg/resource.h | 11 ++++++++++- src/workerd/jsg/ser.c++ | 3 +++ src/workerd/jsg/ser.h | 11 +++++++++++ 3 files changed, 24 insertions(+), 1 deletion(-) diff --git a/src/workerd/jsg/resource.h b/src/workerd/jsg/resource.h index 77e0788e06f..4e844ec4f05 100644 --- a/src/workerd/jsg/resource.h +++ b/src/workerd/jsg/resource.h @@ -1061,9 +1061,18 @@ struct DeserializeInvoker&...)> { static v8::Local call( TypeWrapper& wrapper, Lock& js, Tag tag, Deserializer& deserializer) { - return wrapper.wrap(js, js.v8Context(), kj::none, + auto wrapped = wrapper.wrap(js, js.v8Context(), kj::none, T::deserialize( js, tag, deserializer, TypeWrapper::template TYPE_HANDLER_INSTANCE...)); + if constexpr (kj::isSameType>()) { + return wrapped; + } else { + // deserialize() may return a type whose wrap produces a generic v8::Local + // (e.g. custom-wrapped types like JsReadableStream, whose jsgWrap can yield a plain JS + // object). Host-object deserialization must nonetheless produce an object. + KJ_ASSERT(wrapped->IsObject(), "deserialized host object did not wrap to an object"); + return wrapped.template As(); + } } }; diff --git a/src/workerd/jsg/ser.c++ b/src/workerd/jsg/ser.c++ index e7fb5cb6be0..184262c8e74 100644 --- a/src/workerd/jsg/ser.c++ +++ b/src/workerd/jsg/ser.c++ @@ -294,6 +294,9 @@ v8::Maybe Serializer::WriteHostObject(v8::Isolate* isolate, v8::LocalIsFunction()) { eh.serializeFunction(js, *this, object.As()); return v8::Just(true); + } else if (eh.trySerializeClassInstance(js, *this, object)) { + // The handler recognized this class instance (e.g. by brand check) and serialized it. + return v8::Just(true); } } diff --git a/src/workerd/jsg/ser.h b/src/workerd/jsg/ser.h index 9b9492b206a..8f3a024f396 100644 --- a/src/workerd/jsg/ser.h +++ b/src/workerd/jsg/ser.h @@ -89,6 +89,17 @@ class Serializer final: v8::ValueSerializer::Delegate { // they call for a different design. virtual void serializeProxy( jsg::Lock& js, jsg::Serializer& serializer, v8::Local proxy); + + // Offers the handler a class instance (an object whose prototype is not Object.prototype and + // which is not a JSG-wrapped host object) to serialize as an external. Returns true if the + // handler recognized the object and serialized it; false to fall through to the default + // behavior (DataCloneError). This is how objects implemented in JavaScript (e.g. the + // TypeScript-implemented stream classes, which have no JSG wrapper for the serializer to + // find) participate in external serialization: the handler recognizes them by brand check. + virtual bool trySerializeClassInstance( + jsg::Lock& js, jsg::Serializer& serializer, v8::Local object) { + return false; + } }; struct Options { From cdb0e63c1db3e8ee09a71d43118a6335d31101f2 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 19 Aug 2026 14:09:21 -0700 Subject: [PATCH 02/14] Add getPreferredEncoding to JsReadableStream Serializing a stream for RPC transfer preserves the underlying source's preferred encoding (e.g. gzip passthrough without a recompression round trip), which requires the bridge to expose the preference for both backends. The legacy arm forwards through the controller; the TypeScript arm reaches the underlying ReadableStreamNativeSource through the same non-detaching accessor as the encoding-aware tryGetLength arm, answering IDENTITY for JS-sourced (queued) streams. Like tryGetLength, the native source reports IDENTITY while identity bytes are stashed: the remainder is then not entirely in the source's preferred encoding. --- src/workerd/api/js-readable-stream-test.c++ | 100 +++++++++++++++++++- src/workerd/api/js-readable-stream.c++ | 41 ++++++++ src/workerd/api/js-readable-stream.h | 15 +++ 3 files changed, 154 insertions(+), 2 deletions(-) diff --git a/src/workerd/api/js-readable-stream-test.c++ b/src/workerd/api/js-readable-stream-test.c++ index 8d532c8c45b..358f0715055 100644 --- a/src/workerd/api/js-readable-stream-test.c++ +++ b/src/workerd/api/js-readable-stream-test.c++ @@ -1724,8 +1724,8 @@ KJ_TEST("JsReadableStream detach of a closed TypeScript-backed stream yields a c } // A ReadableStreamSource that reports a (pretend) pre-encoded gzip length in addition to -// its identity length, mimicking system streams whose bytes are stored encoded and can be -// passed through without a recompression round trip. +// its identity length, and prefers GZIP delivery, mimicking system streams whose bytes are +// stored encoded and can be passed through without a recompression round trip. class EncodedLengthSource final: public ReadableStreamSource { public: kj::Promise tryRead(void* buffer, size_t minBytes, size_t maxBytes) override { @@ -1738,6 +1738,10 @@ class EncodedLengthSource final: public ReadableStreamSource { return kj::none; } + StreamEncoding getPreferredEncoding() override { + return StreamEncoding::GZIP; + } + static constexpr uint64_t kIdentityLength = 100; static constexpr uint64_t kGzipLength = 42; }; @@ -1778,6 +1782,98 @@ KJ_TEST("JsReadableStream tryGetLength answers none for encoded queries on non-n }); } +KJ_TEST("JsReadableStream getPreferredEncoding forwards from the native source (both arms)") { + { + // TypeScript arm: reached through the non-detaching source accessor. + auto fixture = makeTsStreamsFixture(); + fixture.runInIoContext([&](const TestFixture::Environment& env) { + auto& js = env.js; + + auto native = JsReadableStream::create(js, env.context, kj::heap()); + KJ_EXPECT(native.getPreferredEncoding(js) == StreamEncoding::GZIP); + + // JS-sourced (queued) streams produce identity bytes. + auto queued = makeTsStream(js, jsg::JsValue(js.obj())); + KJ_EXPECT(queued.getPreferredEncoding(js) == StreamEncoding::IDENTITY); + + // Buffer-backed streams sit on the identity-only in-memory source. + JsReadableStream buffered(js, kj::str(kData)); + KJ_EXPECT(buffered.getPreferredEncoding(js) == StreamEncoding::IDENTITY); + }); + } + + { + // Legacy arm: forwards through the controller. + TestFixture fixture; + fixture.runInIoContext([&](const TestFixture::Environment& env) { + auto& js = env.js; + auto legacy = JsReadableStream::create(js, env.context, kj::heap()); + KJ_EXPECT(legacy.getPreferredEncoding(js) == StreamEncoding::GZIP); + }); + } +} + +// A source with actual content that prefers GZIP delivery, for exercising the interaction +// between the stash and the preferred-encoding report. +class EncodedContentSource final: public ReadableStreamSource { + public: + EncodedContentSource(kj::StringPtr data): data(data) {} + + kj::Promise tryRead(void* buffer, size_t minBytes, size_t maxBytes) override { + auto amount = kj::min(maxBytes, data.size() - offset); + kj::arrayPtr(static_cast(buffer), amount) + .copyFrom(data.slice(offset, offset + amount).asBytes()); + offset += amount; + return amount; + } + + StreamEncoding getPreferredEncoding() override { + return StreamEncoding::GZIP; + } + + private: + kj::StringPtr data; + size_t offset = 0; +}; + +KJ_TEST("ReadableStreamNativeSource stashed bytes force IDENTITY preferred encoding") { + TestFixture testFixture; + MockControllerState state; + testFixture.runInIoContext([&](const TestFixture::Environment& env) -> kj::Promise { + auto& js = env.js; + + auto source = + js.alloc(env.context, kj::heap(kData)); + auto controller = makeMockController(js, state, js.null()); + + // Untouched: the source's own preference passes through. + KJ_EXPECT(source->getPreferredEncoding() == StreamEncoding::GZIP); + + // Abandon a pull so its bytes land in the stash: the stashed bytes are identity bytes + // already drawn from the source, so the remainder is no longer entirely GZIP. + auto abortController = AbortController::constructor(js); + auto pullPromise = source->pull(js, controller, abortController->getSignal()); + abortController->abort(js, kj::none); + + auto promise = pullPromise + .then(js, + [&state, source = source.addRef(), controller = controller.addRef(js)]( + jsg::Lock& js) mutable { + KJ_EXPECT(state.enqueued.size() == 0); + KJ_EXPECT(source->getPreferredEncoding() == StreamEncoding::IDENTITY); + // Redelivery drains the stash, restoring the source's own preference. + return source->pull(js, controller.getHandle(js), freshSignal(js)) + .then(js, [source = kj::mv(source)](jsg::Lock& js) mutable { + KJ_EXPECT(source->getPreferredEncoding() == StreamEncoding::GZIP); + }); + }).then(js, [&state](jsg::Lock& js) { + KJ_EXPECT(state.enqueued.size() == 1); + KJ_EXPECT(state.enqueued[0].asPtr() == kData.asBytes()); + }); + return env.context.awaitJs(js, kj::mv(promise)); + }); +} + KJ_TEST("JsReadableStream cancel of a locked TypeScript-backed stream rejects") { auto fixture = makeTsStreamsFixture(); fixture.runInIoContext([&](const TestFixture::Environment& env) -> kj::Promise { diff --git a/src/workerd/api/js-readable-stream.c++ b/src/workerd/api/js-readable-stream.c++ index 57e248dae4e..d8558d16726 100644 --- a/src/workerd/api/js-readable-stream.c++ +++ b/src/workerd/api/js-readable-stream.c++ @@ -765,6 +765,32 @@ kj::Maybe JsReadableStream::tryGetLength(jsg::Lock& js, StreamEncoding return kj::none; } +StreamEncoding JsReadableStream::getPreferredEncoding(jsg::Lock& js) { + KJ_IF_SOME(i, impl) { + KJ_SWITCH_ONEOF(i.stream) { + KJ_CASE_ONEOF(stream, jsg::Ref) { + return stream->getController().getPreferredEncoding(); + } + KJ_CASE_ONEOF(obj, jsg::JsRef) { + // Only a native underlying source can prefer a non-identity encoding; queued + // (JS-sourced) streams produce identity bytes. + auto sourceValue = + webstreams::dispatchCall(js, "getReadableStreamNativeSource", obj.getHandle(js)); + if (sourceValue.isUndefined()) { + return StreamEncoding::IDENTITY; + } + auto& handler = + KJ_ASSERT_NONNULL(js.tryGetTypeHandler>()); + auto source = KJ_REQUIRE_NONNULL(handler.tryUnwrap(js, sourceValue), + "getReadableStreamNativeSource did not return a ReadableStreamNativeSource"); + return source->getPreferredEncoding(); + } + } + KJ_UNREACHABLE; + } + return StreamEncoding::IDENTITY; +} + jsg::Promise> JsReadableStream::arrayBuffer( jsg::Lock& js, uint64_t limit) { KJ_IF_SOME(i, impl) { @@ -1457,6 +1483,21 @@ kj::Maybe ReadableStreamNativeSource::tryGetLength(StreamEncoding enco return kj::none; } +StreamEncoding ReadableStreamNativeSource::getPreferredEncoding() { + KJ_IF_SOME(active, state) { + // Stashed bytes are identity bytes already drawn from the source: once any exist, the + // remaining content is not entirely in the source's preferred encoding, and only + // IDENTITY describes it. + if (!stash.empty()) { + return StreamEncoding::IDENTITY; + } + return active.source->getPreferredEncoding(); + } + // EOF'd, canceled, or consumed: nothing more will be produced; IDENTITY trivially + // describes the empty remainder. + return StreamEncoding::IDENTITY; +} + kj::Own ReadableStreamNativeSource::releaseForPump(jsg::Lock& js) { KJ_IF_SOME(active, state) { // Extraction requires an undisturbed stream, and any pull implies a read (which diff --git a/src/workerd/api/js-readable-stream.h b/src/workerd/api/js-readable-stream.h index d5bbab28f79..5d532a41410 100644 --- a/src/workerd/api/js-readable-stream.h +++ b/src/workerd/api/js-readable-stream.h @@ -158,6 +158,13 @@ class JsReadableStream final { kj::Maybe tryGetLength( jsg::Lock& js, StreamEncoding encoding = StreamEncoding::IDENTITY); + // The encoding the stream's remaining content would prefer to be transferred in: forwarded + // from the underlying native source when there is one (in a state where its preference + // still describes the remainder), IDENTITY otherwise (JS-sourced streams produce identity + // bytes; a null stream has no content). Used by serialize() to preserve encoding + // passthrough (e.g. gzip) when transferring a stream over RPC. + StreamEncoding getPreferredEncoding(jsg::Lock& js); + // Cancel the stream with the given reason, indicating a loss of interest in the data. The // stream is left disturbed and closed. Rejects if the stream is currently locked, matching // ReadableStream.prototype.cancel(). Canceling a null stream is a no-op (resolved promise). @@ -383,6 +390,14 @@ class ReadableStreamNativeSource final: public jsg::Object { // non-detaching source accessor. kj::Maybe tryGetLength(StreamEncoding encoding); + // The encoding the underlying source would prefer to deliver its remaining content in + // (e.g. GZIP for a passthrough-compressed response body). IDENTITY once the source is + // done, canceled, or consumed, and whenever identity bytes are stashed (stashed bytes + // make a mixed-encoding remainder, which only IDENTITY describes). C++-only, reached the + // same way as the encoding-aware tryGetLength arm; JsReadableStream::serialize() uses it + // to preserve encoding passthrough over RPC transfer. + StreamEncoding getPreferredEncoding(); + JSG_RESOURCE_TYPE(ReadableStreamNativeSource) { JSG_PRIVATE_SYMBOL(kNativeSource); JSG_METHOD(pull); From ea0ee5013b932ddb9d8f9656da0b7a2d0f6be8a6 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 19 Aug 2026 14:15:44 -0700 Subject: [PATCH 03/14] Implement the TypeScript arm of JsReadableStream::serialize Extract the RPC wire plumbing from ReadableStream::serialize into newReadableStreamSerializeSink -- requiring an RPC-backed serializer, pushing the ByteStream to the peer, and writing the external-table entry (encoding + expected length) -- and share it between the legacy serialize and the new TypeScript arm. The arms differ only in how the encoding and expected length are obtained (controller vs. the bridge's getPreferredEncoding/tryGetLength dispatch) and in which pumpTo drives the transfer; lock/disturb validation stays in pumpTo for both, and the TypeScript arm preserves encoding passthrough exactly like the legacy path. --- src/workerd/api/js-readable-stream-test.c++ | 16 +++++++++++ src/workerd/api/js-readable-stream.c++ | 17 ++++++++++- src/workerd/api/js-readable-stream.h | 9 +++--- src/workerd/api/streams/readable.c++ | 31 +++++++++++++-------- src/workerd/api/streams/readable.h | 13 +++++++++ 5 files changed, 70 insertions(+), 16 deletions(-) diff --git a/src/workerd/api/js-readable-stream-test.c++ b/src/workerd/api/js-readable-stream-test.c++ index 358f0715055..2e3501afef7 100644 --- a/src/workerd/api/js-readable-stream-test.c++ +++ b/src/workerd/api/js-readable-stream-test.c++ @@ -1836,6 +1836,22 @@ class EncodedContentSource final: public ReadableStreamSource { size_t offset = 0; }; +KJ_TEST("JsReadableStream serialize of a TypeScript-backed stream requires an RPC serializer") { + auto fixture = makeTsStreamsFixture(); + fixture.runInIoContext([&](const TestFixture::Environment& env) { + auto& js = env.js; + + // Parity with ReadableStream::serialize(): a serializer without an RPC external handler + // must be rejected with DOMDataCloneError before the stream is touched. + auto stream = makeTsStream(js, jsg::JsValue(js.obj())); + jsg::Serializer serializer(js); + KJ_EXPECT_THROW_MESSAGE( + "ReadableStream can only be serialized for RPC", stream.serialize(js, serializer)); + KJ_EXPECT(!stream.isDisturbed(js)); + KJ_EXPECT(!stream.isLocked(js)); + }); +} + KJ_TEST("ReadableStreamNativeSource stashed bytes force IDENTITY preferred encoding") { TestFixture testFixture; MockControllerState state; diff --git a/src/workerd/api/js-readable-stream.c++ b/src/workerd/api/js-readable-stream.c++ index d8558d16726..8bc5968d596 100644 --- a/src/workerd/api/js-readable-stream.c++ +++ b/src/workerd/api/js-readable-stream.c++ @@ -1142,7 +1142,22 @@ void JsReadableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) { stream->serialize(js, serializer); } KJ_CASE_ONEOF(obj, jsg::JsRef) { - KJ_UNIMPLEMENTED("TypeScript-backed ReadableStream is not yet supported"); + // Mirrors ReadableStream::serialize(): pumpTo() performs the lock/disturb validation, + // so the stream must not be modified before that call (the encoding/length queries are + // non-mutating reads). + IoContext& ioctx = IoContext::current(); + + auto encoding = getPreferredEncoding(js); + auto expectedLength = tryGetLength(js, encoding); + + auto sink = newReadableStreamSerializeSink(js, serializer, encoding, expectedLength); + + ioctx.addTask(ioctx.waitForDeferredProxy(pumpTo(js, kj::mv(sink), EndStream::YES)) + .catch_([](kj::Exception&& e) { + // Errors in pumpTo() are automatically propagated to the source and destination. We + // don't want to throw them from here since it'll cause an uncaught exception to be + // reported, even if the application actually does handle it! + })); } } } diff --git a/src/workerd/api/js-readable-stream.h b/src/workerd/api/js-readable-stream.h index 5d532a41410..663f683cf53 100644 --- a/src/workerd/api/js-readable-stream.h +++ b/src/workerd/api/js-readable-stream.h @@ -103,10 +103,11 @@ class JsReadableStream final { // ReadableStream is used. Buffer-backed construction (the data constructors above) // dispatches the same way; see bufferBackedImpl(). // - // TODO(streams-ts): serialize() is the one JsReadableStream operation still lacking a - // TypeScript arm (JS RPC transfer of TS-backed streams; planned as a later phase along - // with the deserialize receive path). Everything else -- pumpTo, unwrap, tee, detach, - // and the pipe dispatch cells -- has landed. + // TODO(streams-ts): every JsReadableStream operation (pumpTo, unwrap, tee, detach, + // serialize, and the pipe dispatch cells) now has a TypeScript arm; the remaining RPC + // gap is the receive path (ReadableStream::deserialize constructs legacy streams + // unconditionally) and serializer recognition of TS streams passed directly as RPC + // values. static JsReadableStream create( jsg::Lock& js, IoContext& ioContext, kj::Own source); diff --git a/src/workerd/api/streams/readable.c++ b/src/workerd/api/streams/readable.c++ index 05c5e8389f9..fb7c30f33b5 100644 --- a/src/workerd/api/streams/readable.c++ +++ b/src/workerd/api/streams/readable.c++ @@ -702,8 +702,11 @@ kj::Own newNoDeferredProxyReadableStream( return kj::heap(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. +kj::Own newReadableStreamSerializeSink(jsg::Lock& js, + jsg::Serializer& serializer, + StreamEncoding encoding, + kj::Maybe expectedLength) { + // Serialize by effectively creating a `JsRpcStub` around the stream and serializing that. // Except we don't actually want to do _exactly_ that, because we do not want to actually create // a `JsRpcStub` locally. So do the important parts of `JsRpcStub::constructor()` followed by // `JsRpcStub::serialize()`. @@ -714,16 +717,8 @@ void ReadableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) { JSG_REQUIRE(externalHandler != nullptr, DOMDataCloneError, "ReadableStream can only be serialized for RPC."); - // NOTE: We're counting on `pumpTo()`, below, to check that the stream is not locked or disturbed - // and other common checks. It's important that we don't modify the stream in any way before - // that call. - IoContext& ioctx = IoContext::current(); - auto& controller = getController(); - StreamEncoding encoding = controller.getPreferredEncoding(); - auto expectedLength = controller.tryGetLength(encoding); - capnp::ByteStream::Client streamCap = [&]() { auto req = externalHandler->getExternalPusher().pushByteStreamRequest(capnp::MessageSize{2, 0}); KJ_IF_SOME(el, expectedLength) { @@ -744,7 +739,21 @@ void ReadableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) { kj::Own kjStream = ioctx.getByteStreamFactory().capnpToKjExplicitEnd(kj::mv(streamCap)); - auto sink = newSystemStream(kj::mv(kjStream), encoding, ioctx); + return newSystemStream(kj::mv(kjStream), encoding, ioctx); +} + +void ReadableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) { + // NOTE: We're counting on `pumpTo()`, below, to check that the stream is not locked or disturbed + // and other common checks. It's important that we don't modify the stream in any way before + // that call. + + IoContext& ioctx = IoContext::current(); + + auto& controller = getController(); + StreamEncoding encoding = controller.getPreferredEncoding(); + auto expectedLength = controller.tryGetLength(encoding); + + auto sink = newReadableStreamSerializeSink(js, serializer, encoding, expectedLength); ioctx.addTask( ioctx.waitForDeferredProxy(pumpTo(js, kj::mv(sink), true)).catch_([](kj::Exception&& e) { diff --git a/src/workerd/api/streams/readable.h b/src/workerd/api/streams/readable.h index 4840601d7dc..7399cf1a325 100644 --- a/src/workerd/api/streams/readable.h +++ b/src/workerd/api/streams/readable.h @@ -567,4 +567,17 @@ class CountQueuingStrategy: public jsg::Object { kj::Own newNoDeferredProxyReadableStream( IoContext& context, kj::Own inner); +// Builds the wire plumbing for transferring a readable stream over RPC: requires `serializer` +// to be RPC-backed (throws DOMDataCloneError otherwise), pushes a ByteStream to the peer, +// writes the external-table entry describing it (encoding plus expected length, when known), +// and returns the local sink the stream's remaining content must be pumped into (ending the +// sink when the stream ends). Shared by ReadableStream::serialize() and JsReadableStream's +// TypeScript arm, which differ only in how the encoding/length are obtained and how the pump +// is driven. The encoding/length reads happen before the handler requirement is checked; both +// are non-mutating, so the reordering relative to the thrown error is unobservable. +kj::Own newReadableStreamSerializeSink(jsg::Lock& js, + jsg::Serializer& serializer, + StreamEncoding encoding, + kj::Maybe expectedLength); + } // namespace workerd::api From 966cd87a836a80190d62df1774843a1d1756198d Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 19 Aug 2026 14:28:33 -0700 Subject: [PATCH 04/14] Implement the TypeScript arm of JsWritableStream::serialize Expose the WritableStreamRpcAdapter revoke machinery from streams/writable.c++ through a newWritableStreamRpcAdapter factory (adopted by the legacy native-sink arm as well), and drive it from a new TsWriterSink: a WritableStreamSink view of the TypeScript stream's writer that enters the isolate per operation, the counterpart of the legacy WritableStreamJsRpcAdapter. The TypeScript arm acquires the writer (which performs the lock validation), wires the sink through the shared adapter with the same revoke-on-IoContext-end and abort-on-disconnect semantics, and transfers in IDENTITY encoding like the legacy JavaScript-backed arm. Extracting a wrapped native sink for encoding-aware transfer is left as a future optimization. --- src/workerd/api/js-writable-stream-test.c++ | 22 +++ src/workerd/api/js-writable-stream.c++ | 158 +++++++++++++++++++- src/workerd/api/streams/writable.c++ | 12 +- src/workerd/api/streams/writable.h | 20 +++ 4 files changed, 208 insertions(+), 4 deletions(-) diff --git a/src/workerd/api/js-writable-stream-test.c++ b/src/workerd/api/js-writable-stream-test.c++ index b4d2a87d9a9..b6f96de4877 100644 --- a/src/workerd/api/js-writable-stream-test.c++ +++ b/src/workerd/api/js-writable-stream-test.c++ @@ -1264,6 +1264,28 @@ KJ_TEST("JsWritableStream create TS arm: closure waitable rejection skips the si KJ_EXPECT(!state.aborted); } +KJ_TEST("JsWritableStream serialize of a TypeScript-backed stream requires an RPC serializer") { + auto fixture = makeTsStreamsFixture(); + fixture.runInIoContext([&](const TestFixture::Environment& env) { + auto& js = env.js; + + auto cppExports = KJ_ASSERT_NONNULL(tryGetBootstrapExport(js, "webstreams/cpp_exports")); + auto exportsObj = KJ_ASSERT_NONNULL(cppExports.tryCast()); + auto constructor = + KJ_ASSERT_NONNULL(exportsObj.get(js, "WritableStream"_kj).tryCast()); + auto streamObj = constructor.newInstance(js, jsg::JsValue(js.obj())); + auto stream = KJ_ASSERT_NONNULL(JsWritableStream::tryUnwrapTs(js, jsg::JsValue(streamObj))); + + // Parity with WritableStream::serialize(): a serializer without an RPC external handler + // must be rejected with DOMDataCloneError before the stream is touched -- in particular, + // before getWriter() locks it. + jsg::Serializer serializer(js); + KJ_EXPECT_THROW_MESSAGE( + "WritableStream can only be serialized for RPC", stream.serialize(js, serializer)); + KJ_EXPECT(!stream.isLocked(js)); + }); +} + KJ_TEST("JsWritableStream::tryUnwrapTs adopts TypeScript streams and rejects impostors") { auto fixture = makeTsStreamsFixture(); SinkState state; diff --git a/src/workerd/api/js-writable-stream.c++ b/src/workerd/api/js-writable-stream.c++ index ec0e0f2a09c..df89ff262cd 100644 --- a/src/workerd/api/js-writable-stream.c++ +++ b/src/workerd/api/js-writable-stream.c++ @@ -4,9 +4,12 @@ #include #include +#include #include +#include #include +#include #include #include @@ -126,6 +129,128 @@ kj::Promise pipeFromPump(kj::Rc state, EndStream end) { co_await kj::mv(proxy.proxyTask); } +// A WritableStreamSink view of a TypeScript-implemented WritableStream, backing +// JsWritableStream::serialize()'s TypeScript arm: it holds the stream's writer (acquired by +// serialize(), which locks the stream) and dispatches each KJ-side operation into the isolate +// to drive it. This is the counterpart of the legacy WritableStreamJsRpcAdapter +// (streams/writable.c++), expressed as a WritableStreamSink so the WritableStreamRpcAdapter +// revoke machinery is reused through newWritableStreamRpcAdapter(): the adapter's revoke path +// cancels any in-flight operation and drops this sink, and the destructor then schedules the +// writer's abort algorithm (matching the legacy adapter's disconnect semantics). Writes are +// delivered to the stream as ArrayBuffers, like the legacy adapter's. +class TsWriterSink final: public WritableStreamSink { + public: + TsWriterSink(IoContext& context, jsg::JsRef writer) + : context(context), + writer(kj::mv(writer)) {} + + ~TsWriterSink() noexcept(false) { + // If the stream was not explicitly ended and the writer is still held, being dropped + // (the peer released the stream, or the revoke path canceled it) aborts the writer, so + // the app's abort algorithm runs. The abort is scheduled as a task -- the destructor + // cannot enter the isolate synchronously -- and may not run at all if the IoContext is + // already tearing down; both behaviors match the legacy WritableStreamJsRpcAdapter, as + // does the inability to convey the peer's actual abort reason. + if (!ended) { + KJ_IF_SOME(w, writer) { + scheduleAbort(kj::mv(w), disconnectedException()); + } + writer = kj::none; + } + } + + kj::Promise write(kj::ArrayPtr buffer) override { + if (writer == kj::none) { + return KJ_EXCEPTION(FAILED, "Write after stream has been closed."); + } + if (buffer == nullptr) return kj::READY_NOW; + return context.run([this, buffer](Worker::Lock& lock) mutable { + jsg::Lock& js = lock; + auto ab = jsg::JsArrayBuffer::create(js, buffer); + return context.awaitJs(lock, invokeWriter(js, "write"_kj, jsg::JsValue(ab))); + }); + } + + kj::Promise write(kj::ArrayPtr> pieces) override { + if (writer == kj::none) { + return KJ_EXCEPTION(FAILED, "Write after stream has been closed."); + } + size_t amount = 0; + for (auto& piece: pieces) { + amount += piece.size(); + } + if (amount == 0) return kj::READY_NOW; + return context.run([this, amount, pieces](Worker::Lock& lock) mutable { + jsg::Lock& js = lock; + // The received buffers are only guaranteed to live until the returned promise + // resolves, but the application may hold the delivered ArrayBuffer longer, so the + // bytes must be copied into a fresh allocation. + auto ab = jsg::JsArrayBuffer::create(js, amount); + auto ptr = ab.asArrayPtr(); + for (auto& piece: pieces) { + if (piece.size() == 0) continue; + ptr.write(piece); + } + return context.awaitJs(lock, invokeWriter(js, "write"_kj, jsg::JsValue(ab))); + }); + } + + kj::Promise end() override { + if (writer == kj::none) { + return KJ_EXCEPTION(FAILED, "End after stream has been closed."); + } + ended = true; + return context.run([this](Worker::Lock& lock) mutable { + jsg::Lock& js = lock; + return context.awaitJs(lock, invokeWriter(js, "close"_kj)); + }); + } + + void abort(kj::Exception reason) override { + // Nothing calls this in the RPC wiring (the adapter's revoke path drops the sink and the + // destructor handles the abort), but the interface requires it: forward the reason to + // the writer's abort algorithm. + KJ_IF_SOME(w, writer) { + scheduleAbort(kj::mv(w), kj::mv(reason)); + } + writer = kj::none; + } + + private: + IoContext& context; + kj::Maybe> writer; + bool ended = false; + + // Invoke a writer method under the isolate lock, returning its (required) promise result. + jsg::Promise invokeWriter( + jsg::Lock& js, kj::StringPtr method, kj::Maybe arg = kj::none) { + auto& w = KJ_UNWRAP_OR(writer, { kj::throwFatalException(disconnectedException()); }); + auto result = webstreams::invokeMethod( + js, w.getHandle(js), method, arg.orDefault(jsg::JsValue(js.undefined()))); + return js.toVoidPromise(KJ_REQUIRE_NONNULL( + JSG_TRY_CAST_PROMISE(result), "writer method did not return a promise", method)); + } + + void scheduleAbort(jsg::JsRef writer, kj::Exception reason) { + context.addTask( + context.run([writer = kj::mv(writer), reason = kj::mv(reason)](Worker::Lock& lock) mutable { + jsg::Lock& js = lock; + auto ex = js.exceptionToJsValue(kj::mv(reason)); + auto result = + webstreams::invokeMethod(js, writer.getHandle(js), "abort"_kj, ex.getHandle(js)); + auto promise = js.toVoidPromise( + KJ_REQUIRE_NONNULL(JSG_TRY_CAST_PROMISE(result), "abort() did not return a promise")); + return IoContext::current().awaitJs(lock, kj::mv(promise)); + })); + } + + static kj::Exception disconnectedException() { + return JSG_KJ_EXCEPTION(DISCONNECTED, Error, + "WritableStream received over RPC was disconnected because the remote execution context " + "has endeded."); + } +}; + } // namespace JsWritableStream::JsWritableStream(jsg::Ref stream) @@ -359,7 +484,38 @@ void JsWritableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) { stream->serialize(js, serializer); } KJ_CASE_ONEOF(obj, jsg::JsRef) { - KJ_UNIMPLEMENTED("TypeScript-backed WritableStream is not yet supported"); + // Mirrors WritableStream::serialize()'s JavaScript-backed arm (streams/writable.c++): + // the peer's bytes are driven through the stream's writer, always in IDENTITY encoding. + // (Like that arm, this could someday learn to extract a wrapped native sink and + // transfer it encoding-aware; see the detach extension hook.) + auto& handler = JSG_REQUIRE_NONNULL(serializer.getExternalHandler(), DOMDataCloneError, + "WritableStream can only be serialized for RPC."); + auto externalHandler = dynamic_cast(&handler); + JSG_REQUIRE(externalHandler != nullptr, DOMDataCloneError, + "WritableStream can only be serialized for RPC."); + + IoContext& ioctx = IoContext::current(); + + // NOTE: We're counting on getWriter() to check that the stream is not locked and other + // common checks. It's important we don't modify the WritableStream before this call. + auto writerValue = webstreams::invokeMethod(js, obj.getHandle(js), "getWriter"_kj); + auto writerObj = KJ_REQUIRE_NONNULL( + JSG_TRY_CAST_OBJECT(writerValue), "getWriter() did not return an object"); + + auto wrapper = + newWritableStreamRpcAdapter(kj::heap(ioctx, jsg::JsRef(js, writerObj))); + + // Make sure this stream will be revoked if the IoContext ends. + ioctx.addTask(wrapper.completionOrRevoke.attach(ioctx.registerPendingEvent())); + + auto capnpStream = ioctx.getByteStreamFactory().kjToCapnp(kj::mv(wrapper.stream)); + + externalHandler->write( + [capnpStream = kj::mv(capnpStream)](rpc::JsValue::External::Builder builder) mutable { + auto ws = builder.initWritableStream(); + ws.setByteStream(kj::mv(capnpStream)); + ws.setEncoding(StreamEncoding::IDENTITY); + }); } } } diff --git a/src/workerd/api/streams/writable.c++ b/src/workerd/api/streams/writable.c++ index 523c215127a..95a1d18cbde 100644 --- a/src/workerd/api/streams/writable.c++ +++ b/src/workerd/api/streams/writable.c++ @@ -499,6 +499,12 @@ class WritableStreamJsRpcAdapter final: public capnp::ExplicitEndOutputStream { } // namespace +WritableStreamRpcWrapper newWritableStreamRpcAdapter(kj::Own inner) { + auto wrapper = kj::heap(kj::mv(inner)); + auto completionOrRevoke = wrapper->waitForCompletionOrRevoke(); + return WritableStreamRpcWrapper{kj::mv(wrapper), kj::mv(completionOrRevoke)}; +} + void WritableStream::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 @@ -520,12 +526,12 @@ void WritableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) { // NOTE: We're counting on `removeSink()`, to check that the stream is not locked and other // common checks. It's important we don't modify the WritableStream before this call. auto encoding = sink->disownEncodingResponsibility(); - auto wrapper = kj::heap(kj::mv(sink)); + auto wrapper = newWritableStreamRpcAdapter(kj::mv(sink)); // Make sure this stream will be revoked if the IoContext ends. - ioctx.addTask(wrapper->waitForCompletionOrRevoke().attach(ioctx.registerPendingEvent())); + ioctx.addTask(wrapper.completionOrRevoke.attach(ioctx.registerPendingEvent())); - auto capnpStream = ioctx.getByteStreamFactory().kjToCapnp(kj::mv(wrapper)); + auto capnpStream = ioctx.getByteStreamFactory().kjToCapnp(kj::mv(wrapper.stream)); externalHandler->write([capnpStream = kj::mv(capnpStream), encoding]( rpc::JsValue::External::Builder builder) mutable { diff --git a/src/workerd/api/streams/writable.h b/src/workerd/api/streams/writable.h index 46444760995..f45c9103d32 100644 --- a/src/workerd/api/streams/writable.h +++ b/src/workerd/api/streams/writable.h @@ -9,6 +9,10 @@ #include #include +namespace capnp { +class ExplicitEndOutputStream; +} + namespace workerd::api { class WritableStreamDefaultWriter: public jsg::Object, public WritableStreamController::Writer { @@ -216,4 +220,20 @@ class WritableStream: public jsg::Object, public kj::PtrTarget { friend class WritableImpl; }; +// The pieces of a WritableStreamSink wrapped for transfer over capnp RPC (the sending side of +// a WritableStream serialization): `stream` is the capnp-compatible wrapper to hand to the +// ByteStreamFactory, and `completionOrRevoke` must be added as a task on the IoContext (with a +// pending event registered). The promise resolves when the peer drops the stream; if it is +// canceled first (the IoContext ends), the wrapped stream is revoked: pending operations are +// canceled and the sink is dropped. +struct WritableStreamRpcWrapper { + kj::Own stream; + kj::Promise completionOrRevoke; +}; + +// Wrap the given sink for transfer over capnp RPC. Shared by WritableStream::serialize()'s +// native-sink arm and JsWritableStream's TypeScript arm (whose sink dispatches into the +// isolate to drive the TypeScript writer). +WritableStreamRpcWrapper newWritableStreamRpcAdapter(kj::Own inner); + } // namespace workerd::api From 594fd3ed23bb9fa5b438e5de088184eab01bbf8e Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 19 Aug 2026 14:31:54 -0700 Subject: [PATCH 05/14] Recognize TypeScript-implemented streams in RPC serialization TypeScript-implemented ReadableStream/WritableStream instances are plain JS class instances with no JSG wrapper, so the serializer cannot route them to the stream serialization functions the way it routes the legacy JSG-wrapped streams. Implement the ExternalHandler's trySerializeClassInstance hook on RpcSerializerExternalHandler: recognize the streams by brand check, then write the same serialization tag and wire form the legacy streams use, so the receiving side needs no knowledge of which implementation the sender runs. --- src/workerd/api/worker-rpc.c++ | 20 ++++++++++++++++++++ src/workerd/api/worker-rpc.h | 8 ++++++++ 2 files changed, 28 insertions(+) diff --git a/src/workerd/api/worker-rpc.c++ b/src/workerd/api/worker-rpc.c++ index 932a55291c0..6c5ce7d371e 100644 --- a/src/workerd/api/worker-rpc.c++ +++ b/src/workerd/api/worker-rpc.c++ @@ -1984,6 +1984,26 @@ void RpcSerializerExternalHandler::serializeProxy( }); } +bool RpcSerializerExternalHandler::trySerializeClassInstance( + jsg::Lock& js, jsg::Serializer& serializer, v8::Local object) { + // TypeScript-implemented streams are recognized by the implementation's private brand. + // tryUnwrapTs answers kj::none whenever the typescript_implemented_streams flag is off (the + // brand-check export does not exist then), so legacy isolates are unaffected. The serialized + // form is written with the same tag and wire protocol as the legacy JSG-wrapped streams, so + // the peer needs no knowledge of which implementation the sender runs. + KJ_IF_SOME(stream, JsReadableStream::tryUnwrapTs(js, object)) { + serializer.writeRawUint32(static_cast(rpc::SerializationTag::READABLE_STREAM)); + stream.serialize(js, serializer); + return true; + } + KJ_IF_SOME(stream, JsWritableStream::tryUnwrapTs(js, object)) { + serializer.writeRawUint32(static_cast(rpc::SerializationTag::WRITABLE_STREAM)); + stream.serialize(js, serializer); + return true; + } + return false; +} + // JsRpcTarget implementation specific to entrypoints. This is used to deliver the first, top-level // call of an RPC session. class EntrypointJsRpcTarget final: public JsRpcTargetBase { diff --git a/src/workerd/api/worker-rpc.h b/src/workerd/api/worker-rpc.h index 6cc4a4cb226..1d1859d9c3b 100644 --- a/src/workerd/api/worker-rpc.h +++ b/src/workerd/api/worker-rpc.h @@ -96,6 +96,14 @@ class RpcSerializerExternalHandler final: public jsg::Serializer::ExternalHandle void serializeProxy( jsg::Lock& js, jsg::Serializer& serializer, v8::Local proxy) override; + // TypeScript-implemented ReadableStream/WritableStream instances (present when the + // typescript_implemented_streams compat flag is enabled) are plain JS class instances with + // no JSG wrapper, so the serializer cannot route them to the stream serialization functions + // the way it routes the legacy JSG-wrapped streams; instead they are recognized here by + // brand check and transferred through the same wire protocol. + bool trySerializeClassInstance( + jsg::Lock& js, jsg::Serializer& serializer, v8::Local object) override; + private: StubOwnership stubOwnership; rpc::JsValue::ExternalPusher::Client externalPusher; From f47c85f7ad7b1c1dfad6eb1d236d81e496946974 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 19 Aug 2026 14:37:24 -0700 Subject: [PATCH 06/14] Dispatch RPC stream deserialization on the streams implementation flag ReadableStream::deserialize and WritableStream::deserialize now return JsReadableStream/JsWritableStream and construct the received stream through the bridges' create(), which dispatches on the typescript_implemented_streams compat flag: under the flag a received stream is TypeScript-backed (and an instance of the global stream class), while legacy isolates construct exactly what they did before. The wire protocol is unchanged, so either implementation can send to either. --- src/workerd/api/js-readable-stream.h | 5 ----- src/workerd/api/streams/readable.c++ | 7 +++++-- src/workerd/api/streams/readable.h | 9 ++++++++- src/workerd/api/streams/writable.c++ | 9 ++++++--- src/workerd/api/streams/writable.h | 10 +++++++++- 5 files changed, 28 insertions(+), 12 deletions(-) diff --git a/src/workerd/api/js-readable-stream.h b/src/workerd/api/js-readable-stream.h index 663f683cf53..bfc8f7d86ed 100644 --- a/src/workerd/api/js-readable-stream.h +++ b/src/workerd/api/js-readable-stream.h @@ -103,11 +103,6 @@ class JsReadableStream final { // ReadableStream is used. Buffer-backed construction (the data constructors above) // dispatches the same way; see bufferBackedImpl(). // - // TODO(streams-ts): every JsReadableStream operation (pumpTo, unwrap, tee, detach, - // serialize, and the pipe dispatch cells) now has a TypeScript arm; the remaining RPC - // gap is the receive path (ReadableStream::deserialize constructs legacy streams - // unconditionally) and serializer recognition of TS streams passed directly as RPC - // values. static JsReadableStream create( jsg::Lock& js, IoContext& ioContext, kj::Own source); diff --git a/src/workerd/api/streams/readable.c++ b/src/workerd/api/streams/readable.c++ index fb7c30f33b5..69302e80dba 100644 --- a/src/workerd/api/streams/readable.c++ +++ b/src/workerd/api/streams/readable.c++ @@ -7,6 +7,7 @@ #include "internal.h" #include "writable.h" +#include #include #include #include @@ -763,7 +764,7 @@ void ReadableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) { })); } -jsg::Ref ReadableStream::deserialize( +JsReadableStream ReadableStream::deserialize( jsg::Lock& js, rpc::SerializationTag tag, jsg::Deserializer& deserializer) { auto& handler = KJ_REQUIRE_NONNULL( deserializer.getExternalHandler(), "got ReadableStream on non-RPC serialized object?"); @@ -784,7 +785,9 @@ jsg::Ref ReadableStream::deserialize( kj::Own in = ioctx.getExternalPusher()->unwrapStream(rs.getStream()); - return js.alloc(ioctx, + // JsReadableStream::create() dispatches on the typescript_implemented_streams compat flag, + // so the received stream is backed by whichever implementation this isolate runs. + return JsReadableStream::create(js, ioctx, kj::heap(newSystemStream(kj::mv(in), encoding, ioctx), ioctx)); } diff --git a/src/workerd/api/streams/readable.h b/src/workerd/api/streams/readable.h index 7399cf1a325..b2d9372b7a9 100644 --- a/src/workerd/api/streams/readable.h +++ b/src/workerd/api/streams/readable.h @@ -13,6 +13,7 @@ namespace workerd::api { class ReadableStreamDefaultReader; class ReadableStreamBYOBReader; +class JsReadableStream; class ReaderImpl final { public: @@ -474,7 +475,13 @@ class ReadableStream: public kj::PtrTarget, public jsg::Object { void signalEof(jsg::Lock& js); void serialize(jsg::Lock& js, jsg::Serializer& serializer); - static jsg::Ref deserialize( + + // Deserializes to a JsReadableStream (rather than a jsg::Ref) so that the + // received stream is backed by whichever stream implementation this isolate runs: under the + // typescript_implemented_streams compat flag the result wraps a TypeScript-implemented + // stream (and is an instance of the global ReadableStream class), otherwise a legacy + // stream exactly as before. Wire-compatible with peers running either implementation. + static JsReadableStream deserialize( jsg::Lock& js, rpc::SerializationTag tag, jsg::Deserializer& deserializer); JSG_SERIALIZABLE(rpc::SerializationTag::READABLE_STREAM); diff --git a/src/workerd/api/streams/writable.c++ b/src/workerd/api/streams/writable.c++ index 95a1d18cbde..b0e6314265e 100644 --- a/src/workerd/api/streams/writable.c++ +++ b/src/workerd/api/streams/writable.c++ @@ -4,6 +4,7 @@ #include "writable.h" +#include #include #include #include @@ -560,7 +561,7 @@ void WritableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) { } } -jsg::Ref WritableStream::deserialize( +JsWritableStream WritableStream::deserialize( jsg::Lock& js, rpc::SerializationTag tag, jsg::Deserializer& deserializer) { auto& handler = KJ_REQUIRE_NONNULL( deserializer.getExternalHandler(), "got WritableStream on non-RPC serialized object?"); @@ -581,8 +582,10 @@ jsg::Ref WritableStream::deserialize( auto stream = ioctx.getByteStreamFactory().capnpToKjExplicitEnd(ws.getByteStream()); auto sink = newSystemStream(kj::mv(stream), encoding, ioctx); - return js.alloc( - ioctx, kj::mv(sink), ioctx.getMetrics().tryCreateWritableByteStreamObserver()); + // JsWritableStream::create() dispatches on the typescript_implemented_streams compat flag, + // so the received stream is backed by whichever implementation this isolate runs. + return JsWritableStream::create( + js, ioctx, kj::mv(sink), ioctx.getMetrics().tryCreateWritableByteStreamObserver()); } void WritableStreamDefaultWriter::visitForMemoryInfo(jsg::MemoryTracker& tracker) const { diff --git a/src/workerd/api/streams/writable.h b/src/workerd/api/streams/writable.h index f45c9103d32..274276265b7 100644 --- a/src/workerd/api/streams/writable.h +++ b/src/workerd/api/streams/writable.h @@ -15,6 +15,8 @@ class ExplicitEndOutputStream; namespace workerd::api { +class JsWritableStream; + class WritableStreamDefaultWriter: public jsg::Object, public WritableStreamController::Writer { public: explicit WritableStreamDefaultWriter(); @@ -203,7 +205,13 @@ class WritableStream: public jsg::Object, public kj::PtrTarget { } void serialize(jsg::Lock& js, jsg::Serializer& serializer); - static jsg::Ref deserialize( + + // Deserializes to a JsWritableStream (rather than a jsg::Ref) so that the + // received stream is backed by whichever stream implementation this isolate runs: under the + // typescript_implemented_streams compat flag the result wraps a TypeScript-implemented + // stream (and is an instance of the global WritableStream class), otherwise a legacy + // stream exactly as before. Wire-compatible with peers running either implementation. + static JsWritableStream deserialize( jsg::Lock& js, rpc::SerializationTag tag, jsg::Deserializer& deserializer); JSG_SERIALIZABLE(rpc::SerializationTag::WRITABLE_STREAM); From 999e22e74f9a1d78fd09d9931b892da701687394 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 19 Aug 2026 15:40:16 -0700 Subject: [PATCH 07/14] Hydrate RPC stream externals before the value graph read V8's deserializer forbids JavaScript execution for the entire value graph read (v8::internal::DisallowJavascriptExecution in ValueDeserializer::ReadObject), but constructing a TypeScript-implemented stream means running its class constructor. Constructing received streams inside ReadableStream::deserialize / WritableStream::deserialize therefore cannot work for the TypeScript implementation. Exploit the fact that the externals table arrives complete before deserialization begins: RpcDeserializerExternalHandler::prepare(), called before readValue(), materializes each stream external -- the full construction the deserialize functions previously performed, through the bridges' implementation-dispatching create() -- into per-external slots, and the deserialize functions claim the prebuilt streams without executing JavaScript. Slots record how many externals each value subsumes so multi-external values (sockets) can join the same mechanism. The pre-pass is gated on the new rpc-externals-hydration autogate and runs for both stream implementations, so it can be verified in production on legacy traffic before the gate is retired in favor of the single hydrated path. With the gate off, deserialization constructs legacy streams in place exactly as before the gate existed (under the experimental TypeScript streams flag that configuration degrades to legacy-backed received streams). --- src/workerd/api/streams/readable.c++ | 32 ++++++++++++++-- src/workerd/api/streams/readable.h | 10 +++++ src/workerd/api/streams/writable.c++ | 32 ++++++++++++++-- src/workerd/api/streams/writable.h | 9 +++++ src/workerd/api/worker-rpc.c++ | 55 ++++++++++++++++++++++++++++ src/workerd/api/worker-rpc.h | 42 +++++++++++++++++++++ src/workerd/util/autogate.h | 10 ++++- 7 files changed, 181 insertions(+), 9 deletions(-) diff --git a/src/workerd/api/streams/readable.c++ b/src/workerd/api/streams/readable.c++ index 69302e80dba..8a2df9e6385 100644 --- a/src/workerd/api/streams/readable.c++ +++ b/src/workerd/api/streams/readable.c++ @@ -764,13 +764,38 @@ void ReadableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) { })); } +JsReadableStream hydrateRpcReadableStream( + jsg::Lock& js, IoContext& ioctx, rpc::JsValue::External::ReadableStream::Reader reader) { + auto encoding = reader.getEncoding(); + + KJ_REQUIRE( + static_cast(encoding) < capnp::Schema::from().getEnumerants().size(), + "unknown StreamEncoding received from peer"); + + kj::Own in = ioctx.getExternalPusher()->unwrapStream(reader.getStream()); + + // JsReadableStream::create() dispatches on the typescript_implemented_streams compat flag, + // so the received stream is backed by whichever implementation this isolate runs. + return JsReadableStream::create(js, ioctx, + kj::heap(newSystemStream(kj::mv(in), encoding, ioctx), ioctx)); +} + JsReadableStream ReadableStream::deserialize( jsg::Lock& js, rpc::SerializationTag tag, jsg::Deserializer& deserializer) { + // No JavaScript may execute here: V8's deserializer forbids it for the duration of the value + // graph read. Everything JS-executing happened in hydrateRpcReadableStream() during + // RpcDeserializerExternalHandler::prepare(); this function only claims the result (or, when + // the rpc-externals-hydration autogate is off, constructs the legacy stream in place, which + // requires no JS). auto& handler = KJ_REQUIRE_NONNULL( deserializer.getExternalHandler(), "got ReadableStream on non-RPC serialized object?"); auto externalHandler = dynamic_cast(&handler); KJ_REQUIRE(externalHandler != nullptr, "got ReadableStream on non-RPC serialized object?"); + KJ_IF_SOME(prebuilt, externalHandler->claimPrebuiltReadable()) { + return kj::mv(prebuilt); + } + auto reader = externalHandler->read(); KJ_REQUIRE(reader.isReadableStream(), "external table slot type doesn't match serialization tag"); @@ -785,10 +810,9 @@ JsReadableStream ReadableStream::deserialize( kj::Own in = ioctx.getExternalPusher()->unwrapStream(rs.getStream()); - // JsReadableStream::create() dispatches on the typescript_implemented_streams compat flag, - // so the received stream is backed by whichever implementation this isolate runs. - return JsReadableStream::create(js, ioctx, - kj::heap(newSystemStream(kj::mv(in), encoding, ioctx), ioctx)); + return JsReadableStream(js.alloc(ioctx, + kj::heap( + newSystemStream(kj::mv(in), encoding, ioctx), ioctx))); } kj::StringPtr ReaderImpl::jsgGetMemoryName() const { diff --git a/src/workerd/api/streams/readable.h b/src/workerd/api/streams/readable.h index b2d9372b7a9..44156d6cf74 100644 --- a/src/workerd/api/streams/readable.h +++ b/src/workerd/api/streams/readable.h @@ -587,4 +587,14 @@ kj::Own newReadableStreamSerializeSink(jsg::Lock& js, StreamEncoding encoding, kj::Maybe expectedLength); +// Materializes a readable stream received over RPC from its external-table entry: adopts the +// pushed ByteStream, wraps it as a system stream of the peer's declared encoding (with deferred +// proxying suppressed, since the stream dies with the RPC session's IoContext), and constructs +// the stream through JsReadableStream::create()'s implementation dispatch. Runs during +// RpcDeserializerExternalHandler::prepare() -- before the V8 graph read -- because the +// TypeScript arm of create() executes JavaScript, which the graph read forbids; +// ReadableStream::deserialize() then claims the result. +JsReadableStream hydrateRpcReadableStream( + jsg::Lock& js, IoContext& ioctx, rpc::JsValue::External::ReadableStream::Reader reader); + } // namespace workerd::api diff --git a/src/workerd/api/streams/writable.c++ b/src/workerd/api/streams/writable.c++ index b0e6314265e..1cbb5fe32f3 100644 --- a/src/workerd/api/streams/writable.c++ +++ b/src/workerd/api/streams/writable.c++ @@ -561,13 +561,39 @@ void WritableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) { } } +JsWritableStream hydrateRpcWritableStream( + jsg::Lock& js, IoContext& ioctx, rpc::JsValue::External::WritableStream::Reader reader) { + auto encoding = reader.getEncoding(); + + KJ_REQUIRE( + static_cast(encoding) < capnp::Schema::from().getEnumerants().size(), + "unknown StreamEncoding received from peer"); + + auto stream = ioctx.getByteStreamFactory().capnpToKjExplicitEnd(reader.getByteStream()); + auto sink = newSystemStream(kj::mv(stream), encoding, ioctx); + + // JsWritableStream::create() dispatches on the typescript_implemented_streams compat flag, + // so the received stream is backed by whichever implementation this isolate runs. + return JsWritableStream::create( + js, ioctx, kj::mv(sink), ioctx.getMetrics().tryCreateWritableByteStreamObserver()); +} + JsWritableStream WritableStream::deserialize( jsg::Lock& js, rpc::SerializationTag tag, jsg::Deserializer& deserializer) { + // No JavaScript may execute here: V8's deserializer forbids it for the duration of the value + // graph read. Everything JS-executing happened in hydrateRpcWritableStream() during + // RpcDeserializerExternalHandler::prepare(); this function only claims the result (or, when + // the rpc-externals-hydration autogate is off, constructs the legacy stream in place, which + // requires no JS). auto& handler = KJ_REQUIRE_NONNULL( deserializer.getExternalHandler(), "got WritableStream on non-RPC serialized object?"); auto externalHandler = dynamic_cast(&handler); KJ_REQUIRE(externalHandler != nullptr, "got WritableStream on non-RPC serialized object?"); + KJ_IF_SOME(prebuilt, externalHandler->claimPrebuiltWritable()) { + return kj::mv(prebuilt); + } + auto reader = externalHandler->read(); KJ_REQUIRE(reader.isWritableStream(), "external table slot type doesn't match serialization tag"); @@ -582,10 +608,8 @@ JsWritableStream WritableStream::deserialize( auto stream = ioctx.getByteStreamFactory().capnpToKjExplicitEnd(ws.getByteStream()); auto sink = newSystemStream(kj::mv(stream), encoding, ioctx); - // JsWritableStream::create() dispatches on the typescript_implemented_streams compat flag, - // so the received stream is backed by whichever implementation this isolate runs. - return JsWritableStream::create( - js, ioctx, kj::mv(sink), ioctx.getMetrics().tryCreateWritableByteStreamObserver()); + return JsWritableStream(js.alloc( + ioctx, kj::mv(sink), ioctx.getMetrics().tryCreateWritableByteStreamObserver())); } void WritableStreamDefaultWriter::visitForMemoryInfo(jsg::MemoryTracker& tracker) const { diff --git a/src/workerd/api/streams/writable.h b/src/workerd/api/streams/writable.h index 274276265b7..41c5eaa5ce8 100644 --- a/src/workerd/api/streams/writable.h +++ b/src/workerd/api/streams/writable.h @@ -244,4 +244,13 @@ struct WritableStreamRpcWrapper { // isolate to drive the TypeScript writer). WritableStreamRpcWrapper newWritableStreamRpcAdapter(kj::Own inner); +// Materializes a writable stream received over RPC from its external-table entry: adopts the +// peer's ByteStream, wraps it as a system sink of the declared encoding, and constructs the +// stream through JsWritableStream::create()'s implementation dispatch. Runs during +// RpcDeserializerExternalHandler::prepare() -- before the V8 graph read -- because the +// TypeScript arm of create() executes JavaScript, which the graph read forbids; +// WritableStream::deserialize() then claims the result. +JsWritableStream hydrateRpcWritableStream( + jsg::Lock& js, IoContext& ioctx, rpc::JsValue::External::WritableStream::Reader reader); + } // namespace workerd::api diff --git a/src/workerd/api/worker-rpc.c++ b/src/workerd/api/worker-rpc.c++ index 6c5ce7d371e..c578a678610 100644 --- a/src/workerd/api/worker-rpc.c++ +++ b/src/workerd/api/worker-rpc.c++ @@ -70,6 +70,55 @@ rpc::JsValue::External::Reader RpcDeserializerExternalHandler::read() { return externals[i++]; } +void RpcDeserializerExternalHandler::prepare(jsg::Lock& js, IoContext& ioctx) { + if (!util::Autogate::isEnabled(util::AutogateKey::RPC_EXTERNALS_HYDRATION)) return; + KJ_ASSERT(!prepared, "prepare() may only be called once"); + + slots.resize(externals.size()); + for (uint index = 0; index < externals.size(); index++) { + auto external = externals[index]; + switch (external.which()) { + case rpc::JsValue::External::READABLE_STREAM: + slots[index].value = hydrateRpcReadableStream(js, ioctx, external.getReadableStream()); + break; + case rpc::JsValue::External::WRITABLE_STREAM: + slots[index].value = hydrateRpcWritableStream(js, ioctx, external.getWritableStream()); + break; + default: + // Every other external type deserializes without executing JavaScript, directly under + // the graph read; no hydration needed. + break; + } + } + prepared = true; +} + +template +kj::Maybe RpcDeserializerExternalHandler::claimPrebuilt() { + if (!prepared) return kj::none; + KJ_ASSERT(i < slots.size()); + auto& slot = slots[i]; + auto& value = + KJ_REQUIRE_NONNULL(slot.value, "external table slot type doesn't match serialization tag"); + KJ_REQUIRE(value.template is(), "external table slot type doesn't match serialization tag"); + T result = kj::mv(value.template get()); + slot.value = kj::none; + i += slot.span; + return kj::mv(result); +} + +kj::Maybe RpcDeserializerExternalHandler::claimPrebuiltReadable() { + return claimPrebuilt(); +} + +kj::Maybe RpcDeserializerExternalHandler::claimPrebuiltWritable() { + return claimPrebuilt(); +} + +kj::Maybe> RpcDeserializerExternalHandler::claimPrebuiltSocket() { + return claimPrebuilt>(); +} + namespace { // Call to construct an `rpc::JsValue` from a JS value. @@ -124,6 +173,12 @@ DeserializeResult deserializeJsValue(jsg::Lock& js, rpc::JsValue::Reader reader) auto disposalGroup = kj::heap(); RpcDeserializerExternalHandler externalHandler(reader.getExternals(), *disposalGroup); + if (reader.getExternals().size() > 0) { + // Materialize the externals that need JavaScript execution (streams, sockets) before + // readValue() begins the graph read, during which JS execution is forbidden. See + // RpcDeserializerExternalHandler::prepare(). + externalHandler.prepare(js, IoContext::current()); + } jsg::Deserializer deserializer(js, reader.getV8Serialized(), kj::none, kj::none, jsg::Deserializer::Options{ diff --git a/src/workerd/api/worker-rpc.h b/src/workerd/api/worker-rpc.h index 1d1859d9c3b..e8a5f6330ef 100644 --- a/src/workerd/api/worker-rpc.h +++ b/src/workerd/api/worker-rpc.h @@ -14,6 +14,8 @@ // // See worker-interface.capnp for the underlying protocol. +#include +#include #include #include #include @@ -127,6 +129,32 @@ class RpcDeserializerExternalHandler final: public jsg::Deserializer::ExternalHa // Read and return the next external. rpc::JsValue::External::Reader read(); + // Materialize the externals that require JavaScript execution to deserialize -- streams and + // sockets -- BEFORE the V8 value graph is read. V8's deserializer forbids JS execution for the + // duration of the graph read (v8::internal::DisallowJavascriptExecution in + // ValueDeserializer::ReadObject), so anything JS-executing -- in particular constructing + // TypeScript-implemented streams -- must happen in this earlier phase, where JS is legal. The + // prebuilt objects are parked in per-external slots, and the corresponding deserialize + // functions claim them (JS-free) via the claimPrebuilt*() methods below as the graph read + // reaches them. + // + // Must be called before the value is deserialized, at most once. Gated on the + // rpc-externals-hydration autogate: when the gate is off this is a no-op, the claims all + // return kj::none, and deserialization constructs legacy streams in place exactly as it did + // before the gate existed. + void prepare(jsg::Lock& js, IoContext& ioctx); + + // Claim the prebuilt object for the next external, advancing past it (and, for sockets, past + // the stream externals the socket subsumes). Returns kj::none if prepare() did not run (the + // autogate is off); the caller then falls back to constructing in place. If prepare() ran but + // the next external is not of the claimed type, the message is malformed (the V8 tag stream + // disagrees with the external table) and this throws. The socket slot holds the WRAPPED + // socket (Socket is incomplete here); Socket::deserialize() unwraps it through its + // TypeHandler, which reads internal fields only -- no JS. + kj::Maybe claimPrebuiltReadable(); + kj::Maybe claimPrebuiltWritable(); + kj::Maybe> claimPrebuiltSocket(); + // All stubs deserialized as part of a particular parameter or result set are placed in a // common disposal group so that they can be disposed together. RpcStubDisposalGroup& getDisposalGroup() { @@ -137,6 +165,20 @@ class RpcDeserializerExternalHandler final: public jsg::Deserializer::ExternalHa capnp::List::Reader externals; uint i = 0; + // Prebuilt values from prepare(), indexed to match `externals`. `span` records how many + // externals the value subsumes (1 for streams; 3 for sockets, which consume their two + // adjacent stream externals), so claiming advances `i` correctly. Slots for externals that + // need no hydration hold kj::none. + struct Slot { + kj::Maybe>> value; + uint span = 1; + }; + kj::Vector slots; + bool prepared = false; + + template + kj::Maybe claimPrebuilt(); + kj::UnwindDetector unwindDetector; RpcStubDisposalGroup& disposalGroup; }; diff --git a/src/workerd/util/autogate.h b/src/workerd/util/autogate.h index 9f8b3eb5693..853add6984e 100644 --- a/src/workerd/util/autogate.h +++ b/src/workerd/util/autogate.h @@ -106,7 +106,15 @@ namespace workerd::util { /* Allow a Socket to be transferred over JS RPC. When disabled, serializing a Socket fails as \ though the type were not serializable at all, and an incoming transferred Socket is \ rejected. */ \ - V(SOCKET_RPC_TRANSFER) + V(SOCKET_RPC_TRANSFER) \ + /* Materialize stream and socket externals of an incoming RPC value BEFORE the V8 value graph \ + is deserialized (RpcDeserializerExternalHandler::prepare()), with deserialize() claiming the \ + prebuilt objects. V8's deserializer forbids JS execution during the graph read, so this is \ + the only phase in which TypeScript-implemented streams (whose construction runs JS) can be \ + built; the mechanism itself is implementation-agnostic and runs for legacy streams too. When \ + disabled, deserialization constructs legacy streams in place, exactly as before the gate \ + existed. */ \ + V(RPC_EXTERNALS_HYDRATION) // clang-format on // -------------------------------------------------------------------------------------- From 11ee1286064d5dff6d57e7d650994301049a1e13 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 19 Aug 2026 15:40:16 -0700 Subject: [PATCH 08/14] Keep TypeScript stream recognition and state probes free of JS execution Streams received over RPC flow through brand checks and state probes while V8's deserializer still forbids JavaScript execution: unwrap (e.g. of a Request initializer's body) brand-checks the value, and API constructors reached from deserialize functions validate stream state (Body's disturbed check, lock checks). Both previously dispatched into the TypeScript implementation, executing JS. Recognition: the TypeScript constructors now stamp an own, non-enumerable api-symbol brand on every instance, and tryUnwrapTs probes for it -- an own-data-property read, no JS. Proxies are rejected up front (an own-property probe on a proxy would invoke its traps), matching the #-brand's deliberate no-tunneling behavior. This also removes a JS call from every unwrap attempt, the optimization the unwrap perf notes had sketched. State probes: isDisturbed/isLocked answer false without dispatching when JS execution is disallowed. The only no-JS scope in which TypeScript-backed streams are reachable is RPC deserialization, and every stream reachable there is hydration-fresh: constructed by the externals pre-pass within the same task, never seen by user code, with no transition mechanism available inside the scope. As a backstop, the bridge's dispatch helpers now assert that JS execution is allowed, turning any missed probe into a named failure instead of an unsymbolizable V8 fatal. The js-rpc-streams-ts-test config runs the entire js-rpc-test suite with the TypeScript streams implementation and the hydration autogate enabled, pinning stream transfer -- both directions, error and abort propagation, locked-stream rejection, nested Request/Response bodies -- to the same expectations the legacy implementation satisfies. --- src/per_isolate/webstreams/readable.ts | 17 ++++++ src/per_isolate/webstreams/writable.ts | 16 ++++++ src/workerd/api/js-readable-stream.c++ | 35 ++++++++---- src/workerd/api/js-streams-bridge.h | 15 +++++- src/workerd/api/js-writable-stream.c++ | 26 +++++---- src/workerd/api/tests/BUILD.bazel | 7 +++ .../api/tests/js-rpc-streams-ts-test.wd-test | 53 +++++++++++++++++++ 7 files changed, 150 insertions(+), 19 deletions(-) create mode 100644 src/workerd/api/tests/js-rpc-streams-ts-test.wd-test diff --git a/src/per_isolate/webstreams/readable.ts b/src/per_isolate/webstreams/readable.ts index 6e4c55468f7..994af232b0f 100644 --- a/src/per_isolate/webstreams/readable.ts +++ b/src/per_isolate/webstreams/readable.ts @@ -269,6 +269,15 @@ let readableStreamDefaultReaderRead: ( let isReadableStream: (value: unknown) => boolean; let isByteStreamController: (value: unknown) => boolean; +// C++-recognition brand (JsReadableStream::tryUnwrapTs): an own, +// non-enumerable marker stamped on every instance by the constructor, so +// the C++ bridge can recognize TypeScript streams via an own-property +// probe, without executing JavaScript. That constraint is load-bearing: +// unwrap runs during RPC deserialization, inside V8's no-JS-execution +// scope. Proxies deliberately do not convey it (the C++ check rejects +// proxies up front), matching the #-brand's no-tunneling behavior. +const kReadableStreamBrand: symbol = utils.getApiSymbol('kReadableStreamBrand'); + // BACKEND-DISPATCH: the byte-CAPABLE gate (one of the five sanctioned // dispatch points). True for any controller whose backend can satisfy // BYOB reads: the queued byte controller, or ANY native controller — @@ -3566,6 +3575,14 @@ class ReadableStream { underlyingSource: UnderlyingSource = {}, strategy: QueuingStrategy = {} ) { + // The C++-recognition brand (see kReadableStreamBrand). Stamped before + // the early returns below so every instance carries it: internal + // shells and native-backed streams included. + ObjectDefineProperty(this, kReadableStreamBrand, { + __proto__: null, + value: true, + } as PropertyDescriptor); + // Internal shell creation (tee branches): skip controller setup // entirely — the tee wiring attaches the SHARED controller and a // forked cursor afterwards. The private symbol is unreachable from diff --git a/src/per_isolate/webstreams/writable.ts b/src/per_isolate/webstreams/writable.ts index cdbf2079fd2..fc2a5e958cd 100644 --- a/src/per_isolate/webstreams/writable.ts +++ b/src/per_isolate/webstreams/writable.ts @@ -152,6 +152,15 @@ let getWritableStreamController: ( // Boolean brand check for the C++ bridge (jsgTryUnwrap). The assertion form // (assertIsWritableStream) throws; this one answers. let isWritableStream: (value: unknown) => boolean; + +// C++-recognition brand (JsWritableStream::tryUnwrapTs): an own, +// non-enumerable marker stamped on every instance by the constructor, so +// the C++ bridge can recognize TypeScript streams via an own-property +// probe, without executing JavaScript. That constraint is load-bearing: +// unwrap runs during RPC deserialization, inside V8's no-JS-execution +// scope. Proxies deliberately do not convey it (the C++ check rejects +// proxies up front), matching the #-brand's no-tunneling behavior. +const kWritableStreamBrand: symbol = utils.getApiSymbol('kWritableStreamBrand'); let setWritableStreamPendingClosure: (stream: WritableStream) => void; let isWritableStreamPendingClosure: (stream: WritableStream) => boolean; // Permanently neutralizes a stream on behalf of the C++ bridge (e.g. a @@ -634,6 +643,13 @@ class WritableStream { underlyingSink: UnderlyingSink = {}, strategy: QueuingStrategy = {} ) { + // The C++-recognition brand (see kWritableStreamBrand). Stamped first + // so every instance carries it regardless of construction path. + ObjectDefineProperty(this, kWritableStreamBrand, { + __proto__: null, + value: true, + } as PropertyDescriptor); + // --- WebIDL strategy dictionary conversion (BEFORE sink reads) --- // Per WebIDL, dictionary-typed arguments are converted at the IDL // layer before the constructor body runs. strategy is QueuingStrategy diff --git a/src/workerd/api/js-readable-stream.c++ b/src/workerd/api/js-readable-stream.c++ index 8bc5968d596..198c8469a0c 100644 --- a/src/workerd/api/js-readable-stream.c++ +++ b/src/workerd/api/js-readable-stream.c++ @@ -34,10 +34,17 @@ bool getReadableStreamIsDisturbed(jsg::Lock& js, jsg::JsObject obj) { // The TypeScript implementation's private-brand check. True only for genuine // TypeScript-implemented ReadableStream instances (including subclasses); false for -// everything else, including proxies wrapping a stream (private fields do not tunnel -// through proxies, deliberately matching the TS-side behavior). +// everything else, including proxies wrapping a stream: an own-property probe on a proxy +// would invoke its traps, and private fields do not tunnel through proxies either, so +// rejecting proxies up front matches the TS-side #-brand behavior. Runs no JavaScript -- +// recognition must work during RPC deserialization, inside V8's no-JS-execution scope -- so +// it probes for the own api-symbol brand stamped by the TypeScript constructor rather than +// asking the TS implementation. bool isTypeScriptReadableStream(jsg::Lock& js, jsg::JsObject obj) { - return webstreams::dispatchCall(js, "isReadableStream", obj).isTrue(); + if (v8::Local(obj)->IsProxy()) { + return false; + } + return obj.has(js, js.symbolInternal("kReadableStreamBrand"), jsg::JsObject::HasOption::OWN); } bool getReadableStreamIsLocked(jsg::Lock& js, jsg::JsObject obj) { @@ -579,12 +586,6 @@ kj::Maybe JsReadableStream::tryUnwrapTs( return kj::none; } KJ_IF_SOME(obj, JSG_TRY_CAST_OBJECT(jsg::JsValue(handle))) { - // PERF NOTE: this is a JS call per unwrap attempt on any object-typed value. Since - // JsReadableStream is typically the first alternative in consumer OneOfs (e.g. - // Body::Initializer), object bodies that are NOT streams (ArrayBuffer, Blob, FormData, - // ...) pay it before falling through. If this shows up in profiles, the alternative is - // an own api-symbol marker stamped by the conduit constructor (same machinery as - // kNativeSource) -- see the design doc's unwrap decision entry. if (isTypeScriptReadableStream(js, obj)) { return JsReadableStream(js, obj.addRef(js)); } @@ -634,6 +635,17 @@ bool JsReadableStream::isDisturbed(jsg::Lock& js) { return cachedIsDisturbed = stream->isDisturbed(); } KJ_CASE_ONEOF(obj, jsg::JsRef) { + if (js.isJavascriptExecutionDisallowed()) { + // Asking the TypeScript side would execute JS, which is forbidden here. The only + // no-JS scope in which TypeScript-backed streams are reachable is RPC + // deserialization (V8 forbids JS for the whole value-graph read; the legacy + // queue's drain scope never touches TS-backed streams), and every TS stream + // reachable there is hydration-fresh: it was just constructed by + // RpcDeserializerExternalHandler::prepare(), user code has never had it, and no + // transition mechanism exists inside the scope. Fresh streams are undisturbed by + // construction. + return false; + } return cachedIsDisturbed = getReadableStreamIsDisturbed(js, obj.getHandle(js)); } } @@ -649,6 +661,11 @@ bool JsReadableStream::isLocked(jsg::Lock& js) { return stream->isLocked(); } KJ_CASE_ONEOF(obj, jsg::JsRef) { + if (js.isJavascriptExecutionDisallowed()) { + // Hydration-fresh by the same reasoning as isDisturbed() above; fresh streams are + // unlocked by construction. + return false; + } return getReadableStreamIsLocked(js, obj.getHandle(js)); } } diff --git a/src/workerd/api/js-streams-bridge.h b/src/workerd/api/js-streams-bridge.h index 9ced20efc9a..be8d89ebc0c 100644 --- a/src/workerd/api/js-streams-bridge.h +++ b/src/workerd/api/js-streams-bridge.h @@ -28,8 +28,18 @@ namespace workerd::api::webstreams { jsg::JsFunction getCppExport(jsg::Lock& js, kj::StringPtr name); // Calls the named webstreams/cpp_exports function with undefined as the receiver. +// +// The assert catches dispatch attempts inside a no-JS scope (most importantly V8's +// deserializer, which forbids JS execution for the whole value-graph read) and turns what +// would be an unsymbolizable V8 fatal ("Invoke in DisallowJavascriptExecutionScope") into a +// named failure. Callers that can legitimately be reached inside such a scope must answer +// without dispatching -- see e.g. the state-probe suppression in JsReadableStream:: +// isDisturbed()/isLocked(), which rely on streams in such scopes being hydration-fresh. template jsg::JsValue dispatchCall(jsg::Lock& js, kj::StringPtr name, Args... args) { + KJ_ASSERT(!js.isJavascriptExecutionDisallowed(), + "attempted to dispatch into the TypeScript streams implementation during a no-JS scope", + name); auto func = getCppExport(js, name); return func.call(js, js.undefined(), kj::fwd(args)...); } @@ -37,9 +47,12 @@ jsg::JsValue dispatchCall(jsg::Lock& js, kj::StringPtr name, Args... args) { // Calls the named method on the given object, with the object itself as the receiver. // Used to invoke the TypeScript conduit's controller facade methods (enqueue, close, // respond, ...). The facade objects are module-owned TypeScript code, not user objects, -// so a missing method indicates an internal error. +// so a missing method indicates an internal error. Carries the same no-JS-scope assert as +// dispatchCall(), for the same reason. template jsg::JsValue invokeMethod(jsg::Lock& js, jsg::JsObject obj, kj::StringPtr name, Args... args) { + KJ_ASSERT(!js.isJavascriptExecutionDisallowed(), + "attempted to invoke a TypeScript streams method during a no-JS scope", name); auto func = KJ_REQUIRE_NONNULL(JSG_TRY_CAST_FUNCTION(obj.get(js, name)), "method not found", name); return func.call(js, obj, args...); diff --git a/src/workerd/api/js-writable-stream.c++ b/src/workerd/api/js-writable-stream.c++ index df89ff262cd..e11a873b013 100644 --- a/src/workerd/api/js-writable-stream.c++ +++ b/src/workerd/api/js-writable-stream.c++ @@ -17,12 +17,18 @@ namespace workerd::api { namespace { -// The TypeScript implementation's private-brand check. True only for genuine -// TypeScript-implemented WritableStream instances (including subclasses); false for -// everything else, including proxies wrapping a stream (private fields do not tunnel -// through proxies, deliberately matching the TS-side behavior). +// True only for genuine TypeScript-implemented WritableStream instances (including +// subclasses); false for everything else, including proxies wrapping a stream: an +// own-property probe on a proxy would invoke its traps, and private fields do not tunnel +// through proxies either, so rejecting proxies up front matches the TS-side #-brand +// behavior. Runs no JavaScript -- recognition must work during RPC deserialization, inside +// V8's no-JS-execution scope -- so it probes for the own api-symbol brand stamped by the +// TypeScript constructor rather than asking the TS implementation. bool isTypeScriptWritableStream(jsg::Lock& js, jsg::JsObject obj) { - return webstreams::dispatchCall(js, "isWritableStream", obj).isTrue(); + if (v8::Local(obj)->IsProxy()) { + return false; + } + return obj.has(js, js.symbolInternal("kWritableStreamBrand"), jsg::JsObject::HasOption::OWN); } bool getWritableStreamIsLocked(jsg::Lock& js, jsg::JsObject obj) { @@ -324,6 +330,12 @@ bool JsWritableStream::isLocked(jsg::Lock& js) { return stream->isLocked(); } KJ_CASE_ONEOF(obj, jsg::JsRef) { + if (js.isJavascriptExecutionDisallowed()) { + // Asking the TypeScript side would execute JS, which is forbidden here. TS-backed + // streams reachable in a no-JS scope are hydration-fresh (see the fuller reasoning + // at JsReadableStream::isDisturbed()); fresh streams are unlocked by construction. + return false; + } return getWritableStreamIsLocked(js, obj.getHandle(js)); } } @@ -557,10 +569,6 @@ kj::Maybe JsWritableStream::tryUnwrapTs( return kj::none; } KJ_IF_SOME(obj, JSG_TRY_CAST_OBJECT(jsg::JsValue(handle))) { - // PERF NOTE: this is a JS call per unwrap attempt on any object-typed value (same - // caveat as JsReadableStream::tryUnwrapTs; see the alternative sketched there). Today - // the only unwrap consumer is JsReadableWritablePair's dictionary tier, so the cost is - // confined to pair-shaped inputs. if (isTypeScriptWritableStream(js, obj)) { return JsWritableStream(js, obj.addRef(js)); } diff --git a/src/workerd/api/tests/BUILD.bazel b/src/workerd/api/tests/BUILD.bazel index 05116d77b3d..5dcbb32b6f7 100644 --- a/src/workerd/api/tests/BUILD.bazel +++ b/src/workerd/api/tests/BUILD.bazel @@ -610,6 +610,13 @@ wd_test( tags = ["resources:socket:1"], ) +wd_test( + src = "js-rpc-streams-ts-test.wd-test", + args = ["--experimental"], + data = ["js-rpc-test.js"], + tags = ["resources:socket:1"], +) + wd_test( src = "js-rpc-params-ownership-test.wd-test", args = ["--experimental"], diff --git a/src/workerd/api/tests/js-rpc-streams-ts-test.wd-test b/src/workerd/api/tests/js-rpc-streams-ts-test.wd-test new file mode 100644 index 00000000000..556fecb4dda --- /dev/null +++ b/src/workerd/api/tests/js-rpc-streams-ts-test.wd-test @@ -0,0 +1,53 @@ +# The full js-rpc-test suite run with the TypeScript streams implementation enabled +# (typescript_implemented_streams flag + the per-isolate bootstrap autogate), pinning JS RPC +# stream transfer -- both directions, error/abort propagation, locked-stream rejection -- +# for TypeScript-backed streams against the same expectations the legacy implementation +# satisfies. +using Workerd = import "/workerd/workerd.capnp"; + +const unitTests :Workerd.Config = ( + services = [ + ( name = "js-rpc-test", + worker = ( + modules = [ + (name = "worker", esModule = embed "js-rpc-test.js") + ], + compatibilityFlags = [ + "nodejs_compat", + "fetcher_no_get_put_delete", + "enable_abortsignal_rpc", + "enhanced_error_serialization", + "enable_ctx_exports", + "experimental", + "streams_enable_constructors", + "http_headers_getsetcookie", + "fetch_legacy_url", + "rpc_params_transfer_stubs", + "typescript_implemented_streams", + ], + bindings = [ + (name = "self", service = (name = "js-rpc-test", entrypoint = "nonClass")), + (name = "MyService", service = (name = "js-rpc-test", entrypoint = "MyService")), + (name = "MyServiceProxy", service = (name = "js-rpc-test", entrypoint = "MyServiceProxy")), + (name = "MyActor", durableObjectNamespace = "MyActor"), + (name = "ActorNoExtends", durableObjectNamespace = "ActorNoExtends"), + (name = "defaultExport", service = "js-rpc-test"), + (name = "twelve", json = "12"), + (name = "GreeterFactory", service = (name = "js-rpc-test", entrypoint = "GreeterFactory")), + ], + + durableObjectNamespaces = [ + (className = "MyActor", uniqueKey = "foo"), + (className = "ActorNoExtends", uniqueKey = "bar"), + ], + + durableObjectStorage = (inMemory = void), + ) + ), + ], + v8Flags = [ "--expose-gc" ], + autogates = [ + "workerd-autogate-per-isolate-javascript-bootstrap", + "workerd-autogate-rpc-externals-hydration", + ], +); From d8089f444ec89431ca47d12d95757b249e7723aa Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 19 Aug 2026 15:50:28 -0700 Subject: [PATCH 09/14] Hydrate transferred sockets before the value graph read Socket::deserialize's construction was already carefully split into scope-safe kj work and deferred JS wiring for the legacy streams, but under the TypeScript streams implementation its stream construction and EOF plumbing execute JavaScript, which the deserializer's no-JS scope forbids. Move the whole construction body into hydrateRpcSocket(), called by the externals pre-pass with JavaScript legal: the socket external subsumes its two adjacent stream externals as a span-3 slot, and Socket::deserialize claims the prebuilt socket (unwrapping through its TypeHandler -- an internal-field read, no JS). With the hydration autogate off, the same body runs in place as before, and the socket-rpc-transfer kill switch keeps both its rejection semantics and its gating of hydration itself. Extend the no-JS-scope backstop to getCppExport(), the chokepoint every bridge call into the TypeScript implementation passes through -- including constructor invocations, which bypass dispatchCall() and previously produced the raw V8 fatal. The js-rpc-socket-streams-ts-test config runs the socket-loopback js-rpc suite (including the socket-transfer assertions) under the TypeScript streams implementation with hydration enabled. --- src/workerd/api/js-streams-bridge.c++ | 6 + src/workerd/api/sockets.c++ | 57 ++++++--- src/workerd/api/sockets.h | 32 ++++- src/workerd/api/tests/BUILD.bazel | 10 ++ .../js-rpc-socket-streams-ts-test.wd-test | 120 ++++++++++++++++++ src/workerd/api/worker-rpc.c++ | 20 +++ 6 files changed, 226 insertions(+), 19 deletions(-) create mode 100644 src/workerd/api/tests/js-rpc-socket-streams-ts-test.wd-test diff --git a/src/workerd/api/js-streams-bridge.c++ b/src/workerd/api/js-streams-bridge.c++ index 6918a9b68b9..1e2c1efdeb2 100644 --- a/src/workerd/api/js-streams-bridge.c++ +++ b/src/workerd/api/js-streams-bridge.c++ @@ -8,6 +8,12 @@ namespace workerd::api::webstreams { jsg::JsFunction getCppExport(jsg::Lock& js, kj::StringPtr name) { + // Every bridge path that invokes the TypeScript implementation obtains its function here + // (including constructor invocations that don't go through dispatchCall), so this is the + // complete chokepoint for the no-JS-scope backstop: callers are about to execute JS, which + // is forbidden e.g. during RPC deserialization. See the fuller comment on dispatchCall(). + KJ_ASSERT(!js.isJavascriptExecutionDisallowed(), + "attempted to call into the TypeScript streams implementation during a no-JS scope", name); auto cppExports = KJ_REQUIRE_NONNULL(tryGetBootstrapExport(js, "webstreams/cpp_exports")); auto cppExportsObj = KJ_REQUIRE_NONNULL(JSG_TRY_CAST_OBJECT(cppExports)); return KJ_REQUIRE_NONNULL(JSG_TRY_CAST_FUNCTION(cppExportsObj.get(js, name))); diff --git a/src/workerd/api/sockets.c++ b/src/workerd/api/sockets.c++ index 5d3bcc3a915..6a4f9efdf82 100644 --- a/src/workerd/api/sockets.c++ +++ b/src/workerd/api/sockets.c++ @@ -850,10 +850,14 @@ void Socket::serialize(jsg::Lock& js, jsg::Serializer& serializer) { writable.serialize(js, serializer); } -jsg::Ref Socket::deserialize( - jsg::Lock& js, rpc::SerializationTag tag, jsg::Deserializer& deserializer) { +jsg::Ref Socket::deserialize(jsg::Lock& js, + rpc::SerializationTag tag, + jsg::Deserializer& deserializer, + const jsg::TypeHandler>& socketHandler) { // Only a peer with the gate on can produce this tag. Reject rather than accept it, so that - // turning the gate off is a complete kill switch. + // turning the gate off is a complete kill switch. (The same gate check keeps + // RpcDeserializerExternalHandler::prepare() from hydrating socket externals, so the claim + // below stays empty and this rejection is reached.) JSG_REQUIRE(util::Autogate::isEnabled(util::AutogateKey::SOCKET_RPC_TRANSFER), DOMDataCloneError, "Transferring a Socket over RPC is not supported."); @@ -863,15 +867,33 @@ jsg::Ref Socket::deserialize( JSG_REQUIRE( externalHandler != nullptr, DOMDataCloneError, "Socket can only be deserialized from RPC."); - auto& ioContext = IoContext::current(); + KJ_IF_SOME(prebuilt, externalHandler->claimPrebuiltSocket()) { + // Hydrated before the graph read; unwrapping reads internal fields only, so no JS + // executes here (forbidden under the deserializer's no-JS scope). + return KJ_ASSERT_NONNULL(socketHandler.tryUnwrap(js, prebuilt.getHandle(js)), + "hydrated socket slot did not hold a Socket"); + } + + // Not hydrated (rpc-externals-hydration autogate off): construct in place. The explicit + // sequencing matters -- the externals must be consumed in Socket::serialize()'s order. + auto socketExternal = externalHandler->read(); + auto readableExternal = externalHandler->read(); + auto writableExternal = externalHandler->read(); + return hydrateRpcSocket( + js, IoContext::current(), socketExternal, readableExternal, writableExternal); +} - // Read the externals in the same order Socket::serialize() wrote them: (1) socket metadata, +jsg::Ref hydrateRpcSocket(jsg::Lock& js, + IoContext& ioContext, + rpc::JsValue::External::Reader socketExternal, + rpc::JsValue::External::Reader readableExternal, + rpc::JsValue::External::Reader writableExternal) { + // The externals arrive in the 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. // (1) Socket metadata. - auto socketExternal = externalHandler->read(); JSG_REQUIRE(socketExternal.isSocket(), DOMDataCloneError, "external table slot type doesn't match serialization tag"); auto socketData = socketExternal.getSocket(); @@ -888,8 +910,7 @@ jsg::Ref Socket::deserialize( } // (2) Readable side: recover the raw input stream from the pushed ByteStream (mirrors - // ReadableStream::deserialize). - auto readableExternal = externalHandler->read(); + // the readable stream's own hydration). JSG_REQUIRE(readableExternal.isReadableStream(), DOMDataCloneError, "external table slot type doesn't match serialization tag"); auto rs = readableExternal.getReadableStream(); @@ -899,8 +920,7 @@ jsg::Ref Socket::deserialize( kj::Own input = ioContext.getExternalPusher()->unwrapStream(rs.getStream()); // (3) Writable side: recover the raw output stream from the peer's ByteStream (mirrors - // WritableStream::deserialize). - auto writableExternal = externalHandler->read(); + // the writable stream's own hydration). JSG_REQUIRE(writableExternal.isWritableStream(), DOMDataCloneError, "external table slot type doesn't match serialization tag"); auto ws = writableExternal.getWritableStream(); @@ -971,12 +991,15 @@ jsg::Ref Socket::deserialize( kj::mv(watchForDisconnectTask), kj::mv(options), kj::mv(tlsStarter), secureTransport, kj::none /* domain */, isDefaultFetchPort, kj::mv(openedPrPair)); - // handleReadableEof() and wireClosedToDisconnect() both attach jsg `.then()` continuations, which - // invoke V8 and are thus forbidden inside the deserialize scope (JS execution is disallowed here). - // Defer them to microtasks that run once readValue() returns and JS is permitted again; wrapping - // and enqueuing don't themselves invoke JS. (The jsg promise deferral primitives can't be used: - // they construct a jsg promise synchronously, invoking V8 and aborting under the disallow scope.) - // The kj-side signals were already set up, so no event can be missed while the microtasks pend. + // handleReadableEof() and wireClosedToDisconnect() both attach jsg `.then()` continuations, + // which invoke V8 -- forbidden when this body runs as Socket::deserialize()'s in-place + // fallback inside the deserialize scope (JS execution is disallowed there). Defer them to + // microtasks that run once readValue() returns and JS is permitted again; wrapping and + // enqueuing don't themselves invoke JS. (The jsg promise deferral primitives can't be used: + // they construct a jsg promise synchronously, invoking V8 and aborting under the disallow + // scope.) The hydration path runs with JS allowed and shares this body; the deferral is + // equally correct there. The kj-side signals were already set up, so no event can be missed + // while the microtasks pend. // // The bodies touch IoContext-bound state, but the isolate's microtask queue can in principle be // drained with no active IoContext, so both bail out early if there is no current context (the @@ -1013,7 +1036,7 @@ jsg::Ref Socket::deserialize( // `opened` was resolved synchronously above, so the transferred socket is immediately in the // OPENED state and can itself be re-serialized for a further RPC hop. - socket.get()->openedState = OpenedState::OPENED; + socket.get()->openedState = Socket::OpenedState::OPENED; return socket; } diff --git a/src/workerd/api/sockets.h b/src/workerd/api/sockets.h index 3745d91dfe8..88c22d668e3 100644 --- a/src/workerd/api/sockets.h +++ b/src/workerd/api/sockets.h @@ -164,8 +164,15 @@ class Socket: public jsg::Object { // RPC serialization support void serialize(jsg::Lock& js, jsg::Serializer& serializer); - static jsg::Ref deserialize( - jsg::Lock& js, rpc::SerializationTag tag, jsg::Deserializer& deserializer); + + // Claims the socket prebuilt by RpcDeserializerExternalHandler::prepare() (see + // hydrateRpcSocket below), or, when the rpc-externals-hydration autogate is off, constructs + // it in place. The TypeHandler unwraps the prebuilt slot's wrapped socket -- an + // internal-field read, safe under the deserializer's no-JS scope. + static jsg::Ref deserialize(jsg::Lock& js, + rpc::SerializationTag tag, + jsg::Deserializer& deserializer, + const jsg::TypeHandler>& socketHandler); JSG_RESOURCE_TYPE(Socket) { JSG_READONLY_PROTOTYPE_PROPERTY(readable, getReadable); @@ -246,6 +253,13 @@ class Socket: public jsg::Object { enum class OpenedState : uint8_t { PENDING, OPENED, FAILED }; OpenedState openedState = OpenedState::PENDING; + // Materializes transferred sockets (including marking them OPENED); see its declaration below. + friend jsg::Ref hydrateRpcSocket(jsg::Lock& js, + IoContext& ioContext, + rpc::JsValue::External::Reader socketExternal, + rpc::JsValue::External::Reader readableExternal, + rpc::JsValue::External::Reader writableExternal); + kj::Promise> processConnection(); jsg::Promise maybeCloseWriteSide(jsg::Lock& js); jsg::Promise closeImplOld(jsg::Lock& js); @@ -289,6 +303,20 @@ jsg::Ref connectImpl(jsg::Lock& js, AnySocketAddress address, jsg::Optional options); +// Materializes a socket received over RPC from its three external-table entries (socket +// metadata, then the readable and writable stream halves, in Socket::serialize()'s order), +// validating the entry types. Runs during RpcDeserializerExternalHandler::prepare() -- before +// the V8 graph read -- because stream construction executes JavaScript under the TypeScript +// streams implementation, which the graph read forbids; Socket::deserialize() then claims the +// result. Also serves as Socket::deserialize()'s in-place fallback when the +// rpc-externals-hydration autogate is off (that path predates the gate and remains scope-safe +// only for legacy streams). +jsg::Ref hydrateRpcSocket(jsg::Lock& js, + IoContext& ioContext, + rpc::JsValue::External::Reader socketExternal, + rpc::JsValue::External::Reader readableExternal, + rpc::JsValue::External::Reader writableExternal); + class SocketsModule final: public jsg::Object { public: SocketsModule() = default; diff --git a/src/workerd/api/tests/BUILD.bazel b/src/workerd/api/tests/BUILD.bazel index 5dcbb32b6f7..a27b5e9007a 100644 --- a/src/workerd/api/tests/BUILD.bazel +++ b/src/workerd/api/tests/BUILD.bazel @@ -983,6 +983,16 @@ wd_test( tags = ["resources:socket:1"], ) +wd_test( + src = "js-rpc-socket-streams-ts-test.wd-test", + args = [ + "--experimental", + "--no-verbose", + ], + data = ["js-rpc-test.js"], + tags = ["resources:socket:1"], +) + wd_test( src = "outbound-interceptor-socket-test.wd-test", args = ["--experimental"], diff --git a/src/workerd/api/tests/js-rpc-socket-streams-ts-test.wd-test b/src/workerd/api/tests/js-rpc-socket-streams-ts-test.wd-test new file mode 100644 index 00000000000..33a4e8ecf8b --- /dev/null +++ b/src/workerd/api/tests/js-rpc-socket-streams-ts-test.wd-test @@ -0,0 +1,120 @@ +# Same as js-rpc-socket-test.wd-test but with the TypeScript streams implementation and the +# rpc-externals-hydration autogate enabled, pinning socket transfer over RPC (and the rest of +# the js-rpc suite) for TypeScript-backed streams. Socket transfer exercises the whole-socket +# hydration slot: the socket external subsumes its two stream externals. +# +# Note that in the BUILD file we explicitly disable info logging for this test because it's too +# noisy. + +using Workerd = import "/workerd/workerd.capnp"; + +const unitTests :Workerd.Config = ( + services = [ + ( name = "js-rpc-test", + worker = ( + modules = [ + (name = "worker", esModule = embed "js-rpc-test.js") + ], + compatibilityFlags = [ + "nodejs_compat", + "fetcher_no_get_put_delete", + "enable_abortsignal_rpc", + "enhanced_error_serialization", + "enable_ctx_exports", + "experimental", + "streams_enable_constructors", + "http_headers_getsetcookie", + "fetch_legacy_url", + "rpc_params_transfer_stubs", + "typescript_implemented_streams"], + bindings = [ + (name = "self", service = "nonClass-loop"), + (name = "MyService", service = "MyService-loop"), + (name = "MyServiceProxy", service = "MyServiceProxy-loop"), + (name = "MyActor", durableObjectNamespace = "MyActor"), + (name = "ActorNoExtends", durableObjectNamespace = "ActorNoExtends"), + (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 = [ + (className = "MyActor", uniqueKey = "foo"), + (className = "ActorNoExtends", uniqueKey = "bar"), + ], + + durableObjectStorage = (inMemory = void), + ) + ), + ( name = "MyService-loop", + external = ( + address = "loopback:MyService-loop", + http = (capnpConnectHost = "cappy") + ) + ), + ( name = "MyServiceProxy-loop", + external = ( + address = "loopback:MyServiceProxy-loop", + http = (capnpConnectHost = "cappy") + ) + ), + ( name = "nonClass-loop", + external = ( + address = "loopback:nonClass-loop", + http = (capnpConnectHost = "cappy") + ) + ), + ( name = "default-loop", + external = ( + address = "loopback:default-loop", + http = (capnpConnectHost = "cappy") + ) + ), + ( name = "GreeterFactory-loop", + external = ( + address = "loopback:GreeterFactory-loop", + http = (capnpConnectHost = "cappy") + ) + ), + ( name = "internet", network = ( allow = ["private"] ) ), + ], + sockets = [ + ( name = "MyService-loop", + address = "loopback:MyService-loop", + service = (name = "js-rpc-test", entrypoint = "MyService"), + http = (capnpConnectHost = "cappy") + ), + ( name = "MyServiceProxy-loop", + address = "loopback:MyServiceProxy-loop", + service = (name = "js-rpc-test", entrypoint = "MyServiceProxy"), + http = (capnpConnectHost = "cappy") + ), + ( name = "nonClass-loop", + address = "loopback:nonClass-loop", + service = (name = "js-rpc-test", entrypoint = "nonClass"), + http = (capnpConnectHost = "cappy") + ), + ( name = "default-loop", + address = "loopback:default-loop", + service = (name = "js-rpc-test"), + http = (capnpConnectHost = "cappy") + ), + ( name = "GreeterFactory-loop", + address = "loopback:GreeterFactory-loop", + service = (name = "js-rpc-test", entrypoint = "GreeterFactory"), + http = (capnpConnectHost = "cappy") + ), + # For testing connect() handler + (name = "tcp", address = "*:8081", tcp = (), service = "js-rpc-test") + ], + v8Flags = [ "--expose-gc" ], + # Required by the socket-transfer assertions guarded by the `socketTransfer` binding above. + autogates = [ + "workerd-autogate-socket-rpc-transfer", + "workerd-autogate-per-isolate-javascript-bootstrap", + "workerd-autogate-rpc-externals-hydration", + ], +); diff --git a/src/workerd/api/worker-rpc.c++ b/src/workerd/api/worker-rpc.c++ index c578a678610..525dd9f9940 100644 --- a/src/workerd/api/worker-rpc.c++ +++ b/src/workerd/api/worker-rpc.c++ @@ -4,6 +4,7 @@ #include #include +#include #include #include #include @@ -84,6 +85,25 @@ void RpcDeserializerExternalHandler::prepare(jsg::Lock& js, IoContext& ioctx) { case rpc::JsValue::External::WRITABLE_STREAM: slots[index].value = hydrateRpcWritableStream(js, ioctx, external.getWritableStream()); break; + case rpc::JsValue::External::SOCKET: { + // The socket transfer kill switch also gates hydration: with it off, the slots stay + // empty and Socket::deserialize() rejects the tag before attempting a claim. + if (!util::Autogate::isEnabled(util::AutogateKey::SOCKET_RPC_TRANSFER)) break; + KJ_REQUIRE(index + 2 < externals.size(), + "socket external is missing its stream externals, possible corruption"); + auto socket = + hydrateRpcSocket(js, ioctx, external, externals[index + 1], externals[index + 2]); + auto& handler = KJ_ASSERT_NONNULL(js.tryGetTypeHandler>()); + slots[index].value = jsg::JsRef(js, + KJ_ASSERT_NONNULL( + jsg::JsValue(handler.wrap(js, kj::mv(socket))).tryCast())); + // The socket subsumed its two adjacent stream externals: record the span for the + // claim's index advance and skip them here so they are not hydrated again (the ++ + // covers the second one). + slots[index].span = 3; + index += 2; + break; + } default: // Every other external type deserializes without executing JavaScript, directly under // the graph read; no hydration needed. From dfe66f5d6f1ca172fb4beb0730098c659d686931 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 19 Aug 2026 15:58:50 -0700 Subject: [PATCH 10/14] Require the hydration autogate to receive streams under TypeScript streams The typescript_implemented_streams compat flag cannot function on the RPC receive path without the rpc-externals-hydration autogate (stream construction executes JavaScript, which the graph read forbids), and supporting the combination by degrading to legacy-backed received streams would mean shipping a second, subtly different behavior for a misconfiguration. Reject it instead: when the gate is off and a stream-bearing value arrives at a TypeScript-streams isolate, RpcDeserializerExternalHandler::prepare() fails with an error naming the missing autogate, before the graph read begins. Stream-free RPC and the send side (which has no no-JS scope) are unaffected, and the in-place fallback paths are now reachable only for legacy-streams isolates, which they are scope-safe for. --- src/workerd/api/sockets.c++ | 7 +++++-- src/workerd/api/streams/readable.c++ | 4 ++++ src/workerd/api/streams/writable.c++ | 4 ++++ src/workerd/api/worker-rpc.c++ | 24 +++++++++++++++++++++++- src/workerd/api/worker-rpc.h | 9 ++++++--- src/workerd/util/autogate.h | 3 ++- 6 files changed, 44 insertions(+), 7 deletions(-) diff --git a/src/workerd/api/sockets.c++ b/src/workerd/api/sockets.c++ index 6a4f9efdf82..d9defde8ed9 100644 --- a/src/workerd/api/sockets.c++ +++ b/src/workerd/api/sockets.c++ @@ -874,8 +874,11 @@ jsg::Ref Socket::deserialize(jsg::Lock& js, "hydrated socket slot did not hold a Socket"); } - // Not hydrated (rpc-externals-hydration autogate off): construct in place. The explicit - // sequencing matters -- the externals must be consumed in Socket::serialize()'s order. + // Not hydrated: only reachable for legacy-streams isolates (the + // typescript_implemented_streams flag requires the hydration gate; see + // RpcDeserializerExternalHandler::prepare()), for which this in-place construction is + // scope-safe. The explicit sequencing matters -- the externals must be consumed in + // Socket::serialize()'s order. auto socketExternal = externalHandler->read(); auto readableExternal = externalHandler->read(); auto writableExternal = externalHandler->read(); diff --git a/src/workerd/api/streams/readable.c++ b/src/workerd/api/streams/readable.c++ index 8a2df9e6385..caa2fbde780 100644 --- a/src/workerd/api/streams/readable.c++ +++ b/src/workerd/api/streams/readable.c++ @@ -796,6 +796,10 @@ JsReadableStream ReadableStream::deserialize( return kj::mv(prebuilt); } + // Not hydrated: the rpc-externals-hydration autogate is off, which + // RpcDeserializerExternalHandler::prepare() only permits for legacy-streams isolates (the + // typescript_implemented_streams flag requires the gate), so constructing the legacy + // stream in place -- which runs no JS -- is the only case here. auto reader = externalHandler->read(); KJ_REQUIRE(reader.isReadableStream(), "external table slot type doesn't match serialization tag"); diff --git a/src/workerd/api/streams/writable.c++ b/src/workerd/api/streams/writable.c++ index 1cbb5fe32f3..2a9ca18d2fe 100644 --- a/src/workerd/api/streams/writable.c++ +++ b/src/workerd/api/streams/writable.c++ @@ -594,6 +594,10 @@ JsWritableStream WritableStream::deserialize( return kj::mv(prebuilt); } + // Not hydrated: only reachable for legacy-streams isolates (the + // typescript_implemented_streams flag requires the hydration gate; see + // RpcDeserializerExternalHandler::prepare()), so the in-place legacy construction -- which + // runs no JS -- is the only case here. auto reader = externalHandler->read(); KJ_REQUIRE(reader.isWritableStream(), "external table slot type doesn't match serialization tag"); diff --git a/src/workerd/api/worker-rpc.c++ b/src/workerd/api/worker-rpc.c++ index 525dd9f9940..e46ef5da8d8 100644 --- a/src/workerd/api/worker-rpc.c++ +++ b/src/workerd/api/worker-rpc.c++ @@ -72,7 +72,29 @@ rpc::JsValue::External::Reader RpcDeserializerExternalHandler::read() { } void RpcDeserializerExternalHandler::prepare(jsg::Lock& js, IoContext& ioctx) { - if (!util::Autogate::isEnabled(util::AutogateKey::RPC_EXTERNALS_HYDRATION)) return; + if (!util::Autogate::isEnabled(util::AutogateKey::RPC_EXTERNALS_HYDRATION)) { + // The TypeScript streams implementation cannot construct received streams during the + // graph read (JS execution is forbidden there), so the typescript_implemented_streams + // compat flag REQUIRES this autogate to receive streams. Reject only configurations + // that actually receive stream-bearing values: stream-free RPC (and the send side, + // which has no such scope) works without the gate. + if (FeatureFlags::get(js).getTypeScriptImplementedStreams()) { + for (auto external: externals) { + switch (external.which()) { + case rpc::JsValue::External::READABLE_STREAM: + case rpc::JsValue::External::WRITABLE_STREAM: + case rpc::JsValue::External::SOCKET: + JSG_FAIL_REQUIRE(Error, + "The typescript_implemented_streams compatibility flag requires the " + "workerd-autogate-rpc-externals-hydration autogate in order to receive " + "streams over RPC."); + default: + break; + } + } + } + return; + } KJ_ASSERT(!prepared, "prepare() may only be called once"); slots.resize(externals.size()); diff --git a/src/workerd/api/worker-rpc.h b/src/workerd/api/worker-rpc.h index e8a5f6330ef..67bd2a6b7c7 100644 --- a/src/workerd/api/worker-rpc.h +++ b/src/workerd/api/worker-rpc.h @@ -139,9 +139,12 @@ class RpcDeserializerExternalHandler final: public jsg::Deserializer::ExternalHa // reaches them. // // Must be called before the value is deserialized, at most once. Gated on the - // rpc-externals-hydration autogate: when the gate is off this is a no-op, the claims all - // return kj::none, and deserialization constructs legacy streams in place exactly as it did - // before the gate existed. + // rpc-externals-hydration autogate: when the gate is off, the claims all return kj::none + // and deserialization constructs legacy streams in place exactly as it did before the gate + // existed. The typescript_implemented_streams flag REQUIRES the gate (TypeScript stream + // construction cannot happen during the graph read), so with the gate off, stream-bearing + // values arriving at a TypeScript-streams isolate are rejected here with a configuration + // error rather than half-supported. void prepare(jsg::Lock& js, IoContext& ioctx); // Claim the prebuilt object for the next external, advancing past it (and, for sockets, past diff --git a/src/workerd/util/autogate.h b/src/workerd/util/autogate.h index 853add6984e..66860b924e2 100644 --- a/src/workerd/util/autogate.h +++ b/src/workerd/util/autogate.h @@ -113,7 +113,8 @@ namespace workerd::util { the only phase in which TypeScript-implemented streams (whose construction runs JS) can be \ built; the mechanism itself is implementation-agnostic and runs for legacy streams too. When \ disabled, deserialization constructs legacy streams in place, exactly as before the gate \ - existed. */ \ + existed; the typescript_implemented_streams compat flag requires this gate to receive \ + streams over RPC (that combination is rejected, not degraded). */ \ V(RPC_EXTERNALS_HYDRATION) // clang-format on // -------------------------------------------------------------------------------------- From 7668f099a1444c4ccc22eae3e93a4b711f630451 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 19 Aug 2026 15:58:50 -0700 Subject: [PATCH 11/14] Pin cross-implementation stream transfer over RPC A runner worker with typescript_implemented_streams calls a peer without it, so every stream RPC crosses stream implementations. Covers all four serializer/deserializer pairings -- TypeScript-serialized to legacy-deserialized in the argument direction (readable and writable) and legacy-serialized to TypeScript-deserialized in the return direction (readable, and both halves of an echo pair nested in an object) -- and asserts that received streams are instances of the receiving isolate's own stream globals. --- src/workerd/api/tests/BUILD.bazel | 6 + .../tests/js-rpc-streams-crossflag-test.js | 108 ++++++++++++++++++ .../js-rpc-streams-crossflag-test.wd-test | 41 +++++++ 3 files changed, 155 insertions(+) create mode 100644 src/workerd/api/tests/js-rpc-streams-crossflag-test.js create mode 100644 src/workerd/api/tests/js-rpc-streams-crossflag-test.wd-test diff --git a/src/workerd/api/tests/BUILD.bazel b/src/workerd/api/tests/BUILD.bazel index a27b5e9007a..ac6bf262bad 100644 --- a/src/workerd/api/tests/BUILD.bazel +++ b/src/workerd/api/tests/BUILD.bazel @@ -617,6 +617,12 @@ wd_test( tags = ["resources:socket:1"], ) +wd_test( + src = "js-rpc-streams-crossflag-test.wd-test", + args = ["--experimental"], + data = ["js-rpc-streams-crossflag-test.js"], +) + wd_test( src = "js-rpc-params-ownership-test.wd-test", args = ["--experimental"], diff --git a/src/workerd/api/tests/js-rpc-streams-crossflag-test.js b/src/workerd/api/tests/js-rpc-streams-crossflag-test.js new file mode 100644 index 00000000000..9220d5bb88b --- /dev/null +++ b/src/workerd/api/tests/js-rpc-streams-crossflag-test.js @@ -0,0 +1,108 @@ +// 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 + +// Cross-implementation stream transfer over JS RPC: the runner worker has the +// typescript_implemented_streams compat flag, the peer worker does not, so every transfer +// below crosses implementations. The wire protocol is implementation-agnostic; each side +// constructs received streams with its own implementation, so received streams are always +// instanceof the receiver's own globals. All four serializer/deserializer pairings are +// covered: TypeScript-serialized -> legacy-deserialized (sendReadable/sendWritable) and +// legacy-serialized -> TypeScript-deserialized (receiveReadable/receivePair). + +import { WorkerEntrypoint } from 'cloudflare:workers'; +import * as assert from 'node:assert'; + +const enc = new TextEncoder(); + +// The legacy-flagged peer. (This same class is also exported by the runner service, but only +// the PEER binding's legacy-flagged instance is ever called.) +export class Peer extends WorkerEntrypoint { + // Receives a runner-created (TypeScript-backed at origin) readable; it deserializes here + // as this worker's own (legacy) implementation. + async readFrom(stream) { + assert.ok(stream instanceof ReadableStream); + return await new Response(stream).text(); + } + + // Receives a runner-created writable and writes a fixed payload into it. + async writeTo(stream) { + assert.ok(stream instanceof WritableStream); + const writer = stream.getWriter(); + await writer.write(enc.encode('written by the peer')); + await writer.close(); + } + + // Returns a peer-created (legacy at origin) readable. + makeReadable() { + return new ReadableStream({ + start(c) { + c.enqueue(enc.encode('made by the peer')); + c.close(); + }, + }); + } + + // Returns both halves of a peer-local identity transform, nested in an object (multiple + // stream externals in one value graph). The runner writes into `writable` and reads the + // echo from `readable`, so bytes traverse the wire in both directions through the peer's + // transform. + makeEchoPair() { + const { readable, writable } = new IdentityTransformStream(); + return { readable, writable }; + } +} + +export default { + async test(controller, env) { + // Both services embed this file, so the peer runs this test export too; only the runner + // (which has the binding) performs the assertions. + if (env.PEER === undefined) return; + + // TS-serialized -> legacy-deserialized (argument direction), readable. + { + const stream = new ReadableStream({ + start(c) { + c.enqueue(enc.encode('made by the runner')); + c.close(); + }, + }); + assert.strictEqual(await env.PEER.readFrom(stream), 'made by the runner'); + } + + // TS-serialized -> legacy-deserialized, writable (peer writes, runner reads back + // through its own TypeScript-backed identity transform). + { + const { readable, writable } = new IdentityTransformStream(); + const promise = env.PEER.writeTo(writable); + assert.strictEqual( + await new Response(readable).text(), + 'written by the peer' + ); + await promise; + } + + // Legacy-serialized -> TS-deserialized (return direction), readable. The received + // stream must be an instance of THIS worker's (TypeScript-implemented) global. + { + const stream = await env.PEER.makeReadable(); + assert.ok(stream instanceof ReadableStream); + assert.strictEqual(await new Response(stream).text(), 'made by the peer'); + } + + // Legacy-serialized -> TS-deserialized, both directions at once through the peer's + // echo pair, nested in an object. + { + const { readable, writable } = await env.PEER.makeEchoPair(); + assert.ok(readable instanceof ReadableStream); + assert.ok(writable instanceof WritableStream); + const writer = writable.getWriter(); + await writer.write(enc.encode('echoed through the peer')); + await writer.close(); + assert.strictEqual( + await new Response(readable).text(), + 'echoed through the peer' + ); + } + }, +}; diff --git a/src/workerd/api/tests/js-rpc-streams-crossflag-test.wd-test b/src/workerd/api/tests/js-rpc-streams-crossflag-test.wd-test new file mode 100644 index 00000000000..6f29eb1781f --- /dev/null +++ b/src/workerd/api/tests/js-rpc-streams-crossflag-test.wd-test @@ -0,0 +1,41 @@ +# Cross-implementation stream transfer: the runner has typescript_implemented_streams, the +# peer does not, so every stream RPC in the test crosses stream implementations (in both +# serialization directions). See the header comment in the JS file. +using Workerd = import "/workerd/workerd.capnp"; + +const unitTests :Workerd.Config = ( + services = [ + ( name = "runner", + worker = ( + modules = [ + (name = "worker", esModule = embed "js-rpc-streams-crossflag-test.js") + ], + compatibilityFlags = [ + "nodejs_compat", + "experimental", + "streams_enable_constructors", + "typescript_implemented_streams", + ], + bindings = [ + (name = "PEER", service = (name = "peer", entrypoint = "Peer")), + ], + ) + ), + ( name = "peer", + worker = ( + modules = [ + (name = "worker", esModule = embed "js-rpc-streams-crossflag-test.js") + ], + compatibilityFlags = [ + "nodejs_compat", + "experimental", + "streams_enable_constructors", + ], + ) + ), + ], + autogates = [ + "workerd-autogate-per-isolate-javascript-bootstrap", + "workerd-autogate-rpc-externals-hydration", + ], +); From a8a7ba4d8f6fdafa65a923c95f20c5749ce099ff Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 19 Aug 2026 16:06:42 -0700 Subject: [PATCH 12/14] Document the stream brand and no-JS-scope rules in the webstreams contract The own-symbol instance brands and the prohibition on dispatching into the TypeScript implementation while JS execution is disallowed are load-bearing parts of the C++/JS contract (RPC deserialization depends on both); record them alongside the other cross-fence rules. --- src/per_isolate/webstreams/AGENTS.md | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/src/per_isolate/webstreams/AGENTS.md b/src/per_isolate/webstreams/AGENTS.md index d9018f121dc..2b44a0ff0a4 100644 --- a/src/per_isolate/webstreams/AGENTS.md +++ b/src/per_isolate/webstreams/AGENTS.md @@ -44,6 +44,22 @@ private-brand dispatch, no `instanceof`) apply here — see and the C++ bridge via the API-symbol registry. The C++ mocks in `js-readable-stream-test.c++` construct real `ReadableStreamNativeSource` objects; no JS-visible marker export exists. +- Every stream instance carries an own, non-enumerable api-symbol brand + (`kReadableStreamBrand` / `kWritableStreamBrand`), stamped at the very + top of the constructor (before any early return). The C++ bridge's + `tryUnwrapTs` recognizes streams by probing it — an own-data-property + read that executes no JS. That constraint is load-bearing: recognition + runs during RPC deserialization, inside V8's no-JS-execution scope. Any + new construction path MUST go through the constructors (or stamp the + brand itself). +- The C++ bridge MUST NOT dispatch into the TypeScript implementation + while JS execution is disallowed (`js.isJavascriptExecutionDisallowed()`, + set during RPC value deserialization): `getCppExport`/`dispatchCall`/ + `invokeMethod` assert this. Bridge operations reachable in that scope + either answer from C++-side knowledge (state probes return the + hydration-fresh answers; see `JsReadableStream::isDisturbed`) or happen + before the scope entirely (stream construction via + `RpcDeserializerExternalHandler::prepare()`'s externals hydration). ## ANTI-PATTERNS From 7b94e1a7e9a4c34b53fd42c07d8e5770cf37cd08 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 19 Aug 2026 18:53:08 -0700 Subject: [PATCH 13/14] Drive RPC writer transfer through frozen internals, not public methods JsWritableStream::serialize's TypeScript arm acquired the writer via WritableStream.prototype.getWriter and TsWriterSink drove it through WritableStreamDefaultWriter.prototype.{write,close,abort} -- all user-patchable: a replaced method could intercept the transferred bytes, fake the transfer while leaving the stream unlocked, or drop remote writes. Route acquisition and every writer operation through new members of the frozen cppExports object instead (acquireWritableStreamWriter has the public getWriter's exact semantics, including the locked TypeError from the shared constructor path; the write/close/abort operations reuse the existing internal algorithms, with a writerAbortInternal added beside them for the released-writer check the public abort performs). The readable arm was already patch-proof (own-symbol extraction and frozen-export dispatch throughout). The regression test transfers both stream directions with booby-trapped prototypes and asserts the traps never fire; before this change it recorded getWriter, write, and close. --- src/per_isolate/webstreams/writable.ts | 40 ++++++++ src/workerd/api/js-writable-stream.c++ | 42 ++++++--- src/workerd/api/tests/BUILD.bazel | 6 ++ .../tests/js-rpc-streams-pollution-test.js | 93 +++++++++++++++++++ .../js-rpc-streams-pollution-test.wd-test | 39 ++++++++ 5 files changed, 205 insertions(+), 15 deletions(-) create mode 100644 src/workerd/api/tests/js-rpc-streams-pollution-test.js create mode 100644 src/workerd/api/tests/js-rpc-streams-pollution-test.wd-test diff --git a/src/per_isolate/webstreams/writable.ts b/src/per_isolate/webstreams/writable.ts index fc2a5e958cd..224742e0167 100644 --- a/src/per_isolate/webstreams/writable.ts +++ b/src/per_isolate/webstreams/writable.ts @@ -232,6 +232,10 @@ let writerWriteInternal: ( let writerCloseInternal: ( writer: WritableStreamDefaultWriter ) => Promise; +let writerAbortInternal: ( + writer: WritableStreamDefaultWriter, + reason: unknown +) => Promise; let writerReleaseInternal: (writer: WritableStreamDefaultWriter) => void; let getWriterReadyPromiseInternal: ( writer: WritableStreamDefaultWriter @@ -1218,6 +1222,19 @@ class WritableStreamDefaultWriter< return promise; }; + writerAbortInternal = ( + writer: WritableStreamDefaultWriter, + reason: unknown + ) => { + const stream = writer.#stream; + if (stream === undefined) { + return PromiseReject( + new TypeError('This writer has been released') + ) as Promise; + } + return writableStreamAbort(stream, reason); + }; + writerCloseInternal = (writer: WritableStreamDefaultWriter) => { const stream = writer.#stream; if (stream === undefined) { @@ -1649,6 +1666,29 @@ const cppExports = ObjectFreeze({ writableStreamAbort, writableStreamClose, writableStreamFlush, + // The RPC-transfer writer operations (JsWritableStream::serialize's + // TsWriterSink): acquisition and per-operation dispatch go through these + // internal algorithms rather than the public prototype methods, which are + // user-patchable — a replaced getWriter/write/close/abort must not be able + // to intercept or fake a stream's RPC transfer. Acquisition has the public + // getWriter's exact semantics (the locked TypeError comes from the same + // constructor path). + acquireWritableStreamWriter( + stream: WritableStream + ): WritableStreamDefaultWriter { + return new WritableStreamDefaultWriter(stream); + }, + writableStreamWriterWrite: ( + writer: WritableStreamDefaultWriter, + chunk: W + ): Promise => writerWriteInternal(writer, chunk), + writableStreamWriterClose: ( + writer: WritableStreamDefaultWriter + ): Promise => writerCloseInternal(writer), + writableStreamWriterAbort: ( + writer: WritableStreamDefaultWriter, + reason: unknown + ): Promise => writerAbortInternal(writer, reason), }); module.exports = { diff --git a/src/workerd/api/js-writable-stream.c++ b/src/workerd/api/js-writable-stream.c++ index e11a873b013..4ce74c7947a 100644 --- a/src/workerd/api/js-writable-stream.c++ +++ b/src/workerd/api/js-writable-stream.c++ @@ -173,7 +173,8 @@ class TsWriterSink final: public WritableStreamSink { return context.run([this, buffer](Worker::Lock& lock) mutable { jsg::Lock& js = lock; auto ab = jsg::JsArrayBuffer::create(js, buffer); - return context.awaitJs(lock, invokeWriter(js, "write"_kj, jsg::JsValue(ab))); + return context.awaitJs( + lock, invokeWriter(js, "writableStreamWriterWrite"_kj, jsg::JsValue(ab))); }); } @@ -197,7 +198,8 @@ class TsWriterSink final: public WritableStreamSink { if (piece.size() == 0) continue; ptr.write(piece); } - return context.awaitJs(lock, invokeWriter(js, "write"_kj, jsg::JsValue(ab))); + return context.awaitJs( + lock, invokeWriter(js, "writableStreamWriterWrite"_kj, jsg::JsValue(ab))); }); } @@ -208,7 +210,7 @@ class TsWriterSink final: public WritableStreamSink { ended = true; return context.run([this](Worker::Lock& lock) mutable { jsg::Lock& js = lock; - return context.awaitJs(lock, invokeWriter(js, "close"_kj)); + return context.awaitJs(lock, invokeWriter(js, "writableStreamWriterClose"_kj)); }); } @@ -228,13 +230,17 @@ class TsWriterSink final: public WritableStreamSink { bool ended = false; // Invoke a writer method under the isolate lock, returning its (required) promise result. + // Drive a writer operation through the frozen cppExports internals (never the public + // writer prototype methods, which are user-patchable: a replaced write/close must not be + // able to intercept or fake a stream's RPC transfer), returning its (required) promise + // result. jsg::Promise invokeWriter( - jsg::Lock& js, kj::StringPtr method, kj::Maybe arg = kj::none) { + jsg::Lock& js, kj::StringPtr op, kj::Maybe arg = kj::none) { auto& w = KJ_UNWRAP_OR(writer, { kj::throwFatalException(disconnectedException()); }); - auto result = webstreams::invokeMethod( - js, w.getHandle(js), method, arg.orDefault(jsg::JsValue(js.undefined()))); + auto result = webstreams::dispatchCall( + js, op, jsg::JsValue(w.getHandle(js)), arg.orDefault(jsg::JsValue(js.undefined()))); return js.toVoidPromise(KJ_REQUIRE_NONNULL( - JSG_TRY_CAST_PROMISE(result), "writer method did not return a promise", method)); + JSG_TRY_CAST_PROMISE(result), "writer operation did not return a promise", op)); } void scheduleAbort(jsg::JsRef writer, kj::Exception reason) { @@ -242,10 +248,11 @@ class TsWriterSink final: public WritableStreamSink { context.run([writer = kj::mv(writer), reason = kj::mv(reason)](Worker::Lock& lock) mutable { jsg::Lock& js = lock; auto ex = js.exceptionToJsValue(kj::mv(reason)); - auto result = - webstreams::invokeMethod(js, writer.getHandle(js), "abort"_kj, ex.getHandle(js)); - auto promise = js.toVoidPromise( - KJ_REQUIRE_NONNULL(JSG_TRY_CAST_PROMISE(result), "abort() did not return a promise")); + // Same internal-dispatch requirement as invokeWriter above. + auto result = webstreams::dispatchCall(js, "writableStreamWriterAbort", + jsg::JsValue(writer.getHandle(js)), jsg::JsValue(ex.getHandle(js))); + auto promise = js.toVoidPromise(KJ_REQUIRE_NONNULL( + JSG_TRY_CAST_PROMISE(result), "writableStreamWriterAbort did not return a promise")); return IoContext::current().awaitJs(lock, kj::mv(promise)); })); } @@ -508,11 +515,16 @@ void JsWritableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) { IoContext& ioctx = IoContext::current(); - // NOTE: We're counting on getWriter() to check that the stream is not locked and other - // common checks. It's important we don't modify the WritableStream before this call. - auto writerValue = webstreams::invokeMethod(js, obj.getHandle(js), "getWriter"_kj); + // NOTE: We're counting on writer acquisition to check that the stream is not locked + // and other common checks. It's important we don't modify the WritableStream before + // this call. Acquisition goes through the frozen cppExports internals -- NOT the + // public getWriter, which is user-patchable and must not be able to fake the + // transfer -- with the public method's exact semantics (same constructor path, + // including the locked TypeError). + auto writerValue = + webstreams::dispatchCall(js, "acquireWritableStreamWriter", obj.getHandle(js)); auto writerObj = KJ_REQUIRE_NONNULL( - JSG_TRY_CAST_OBJECT(writerValue), "getWriter() did not return an object"); + JSG_TRY_CAST_OBJECT(writerValue), "acquireWritableStreamWriter did not return an object"); auto wrapper = newWritableStreamRpcAdapter(kj::heap(ioctx, jsg::JsRef(js, writerObj))); diff --git a/src/workerd/api/tests/BUILD.bazel b/src/workerd/api/tests/BUILD.bazel index ac6bf262bad..5614e1fc107 100644 --- a/src/workerd/api/tests/BUILD.bazel +++ b/src/workerd/api/tests/BUILD.bazel @@ -623,6 +623,12 @@ wd_test( data = ["js-rpc-streams-crossflag-test.js"], ) +wd_test( + src = "js-rpc-streams-pollution-test.wd-test", + args = ["--experimental"], + data = ["js-rpc-streams-pollution-test.js"], +) + wd_test( src = "js-rpc-params-ownership-test.wd-test", args = ["--experimental"], diff --git a/src/workerd/api/tests/js-rpc-streams-pollution-test.js b/src/workerd/api/tests/js-rpc-streams-pollution-test.js new file mode 100644 index 00000000000..68635ab35b0 --- /dev/null +++ b/src/workerd/api/tests/js-rpc-streams-pollution-test.js @@ -0,0 +1,93 @@ +// 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 + +// RPC stream transfer must not dispatch through user-patchable prototype methods: +// serialization of a TypeScript-backed stream drives the stream through internal +// (frozen cppExports) operations, so replacing WritableStream.prototype.getWriter or +// WritableStreamDefaultWriter.prototype.{write,close,abort} (or the readable-side +// getReader) must neither intercept the transferred data nor fake the transfer. + +import { WorkerEntrypoint } from 'cloudflare:workers'; +import * as assert from 'node:assert'; + +const enc = new TextEncoder(); + +export class Peer extends WorkerEntrypoint { + // Writes a payload into a received writable, then closes it. + async writeTo(stream) { + const writer = stream.getWriter(); + await writer.write(enc.encode('delivered intact')); + await writer.close(); + } + + // Reads a received readable fully. + async readFrom(stream) { + return await new Response(stream).text(); + } +} + +export default { + async test(controller, env) { + // Both services embed this file; only the runner (which has the binding) + // performs the assertions. + if (env.PEER === undefined) return; + + const trapped = []; + const trap = (name, impl) => + function (...args) { + trapped.push(name); + return impl.apply(this, args); + }; + + // Booby-trap the public stream surfaces BEFORE any transfer. + const wsProto = WritableStream.prototype; + const writerProto = WritableStreamDefaultWriter.prototype; + const rsProto = ReadableStream.prototype; + const origGetWriter = wsProto.getWriter; + const origWrite = writerProto.write; + const origClose = writerProto.close; + const origAbort = writerProto.abort; + const origGetReader = rsProto.getReader; + wsProto.getWriter = trap('getWriter', origGetWriter); + writerProto.write = trap('write', origWrite); + writerProto.close = trap('close', origClose); + writerProto.abort = trap('abort', origAbort); + rsProto.getReader = trap('getReader', origGetReader); + + try { + // Transfer a writable: the peer writes into it; the payload must arrive + // through the real stream, with no patched method ever invoked. + { + const { readable, writable } = new IdentityTransformStream(); + const promise = env.PEER.writeTo(writable); + const text = await new Response(readable).text(); + assert.strictEqual(text, 'delivered intact'); + await promise; + } + + // Transfer a readable (serialization pumps it): same requirement. + { + const stream = new ReadableStream({ + start(c) { + c.enqueue(enc.encode('pumped intact')); + c.close(); + }, + }); + assert.strictEqual(await env.PEER.readFrom(stream), 'pumped intact'); + } + + assert.deepStrictEqual( + trapped, + [], + `internal machinery dispatched through patched prototypes: ${trapped.join(', ')}` + ); + } finally { + wsProto.getWriter = origGetWriter; + writerProto.write = origWrite; + writerProto.close = origClose; + writerProto.abort = origAbort; + rsProto.getReader = origGetReader; + } + }, +}; diff --git a/src/workerd/api/tests/js-rpc-streams-pollution-test.wd-test b/src/workerd/api/tests/js-rpc-streams-pollution-test.wd-test new file mode 100644 index 00000000000..f6fe285f488 --- /dev/null +++ b/src/workerd/api/tests/js-rpc-streams-pollution-test.wd-test @@ -0,0 +1,39 @@ +# RPC stream transfer with booby-trapped public stream prototypes: serialization must +# drive the streams through internal operations only. See the header comment in the JS. +using Workerd = import "/workerd/workerd.capnp"; + +const unitTests :Workerd.Config = ( + services = [ + ( name = "runner", + worker = ( + modules = [ + (name = "worker", esModule = embed "js-rpc-streams-pollution-test.js") + ], + compatibilityFlags = [ + "nodejs_compat", + "experimental", + "typescript_implemented_streams", + ], + bindings = [ + (name = "PEER", service = (name = "peer", entrypoint = "Peer")), + ], + ) + ), + ( name = "peer", + worker = ( + modules = [ + (name = "worker", esModule = embed "js-rpc-streams-pollution-test.js") + ], + compatibilityFlags = [ + "nodejs_compat", + "experimental", + "typescript_implemented_streams", + ], + ) + ), + ], + autogates = [ + "workerd-autogate-per-isolate-javascript-bootstrap", + "workerd-autogate-rpc-externals-hydration", + ], +); From 4533e7dc593754f1e42d0ff26c426be91557c81f Mon Sep 17 00:00:00 2001 From: James M Snell Date: Fri, 21 Aug 2026 12:56:03 -0700 Subject: [PATCH 14/14] Describe the stream brand accurately and check RPC-backing first The api-symbol brand that isTypeScriptReadableStream() probes stays visible to reflection, so user code can read it off a real stream and stamp it on an object of its own: the check recognizes streams, it does not authenticate them. Say so, along with what makes that safe -- recognition grants nothing on its own, because every operation reached afterwards goes through the TypeScript internal algorithms, whose real #-brand checks reject an impostor with a TypeError before either RPC serialize arm writes anything to the wire, and a symbol-keyed brand cannot arrive over the wire at all. Pin it with a test that brands a bare object, watches it unwrap, and watches the first operation throw. Extract the serializer's RPC-handler requirement into requireReadableStreamRpcSerializer() and resolve it first in both readable serialize arms. Folding that requirement into newReadableStreamSerializeSink() had placed it behind IoContext::current(), so serializing a stream from global scope reported the async-I/O error instead of the DataCloneError that says streams only transfer over RPC -- the order the writable arm already used. Also fix the "has endeded" typo in the RPC disconnect message, in all three copies and the test that asserts it, and drop a stale first line from the comment on TsWriterSink's writer dispatch. --- src/workerd/api/js-readable-stream-test.c++ | 16 ++++++++++ src/workerd/api/js-readable-stream.c++ | 33 +++++++++++++++------ src/workerd/api/js-writable-stream-test.c++ | 16 ++++++++++ src/workerd/api/js-writable-stream.c++ | 17 +++++------ src/workerd/api/streams/readable.c++ | 29 ++++++++++-------- src/workerd/api/streams/readable.h | 26 +++++++++------- src/workerd/api/streams/writable.c++ | 4 +-- src/workerd/api/tests/js-rpc-test.js | 2 +- 8 files changed, 100 insertions(+), 43 deletions(-) diff --git a/src/workerd/api/js-readable-stream-test.c++ b/src/workerd/api/js-readable-stream-test.c++ index 2e3501afef7..e0efa94efd6 100644 --- a/src/workerd/api/js-readable-stream-test.c++ +++ b/src/workerd/api/js-readable-stream-test.c++ @@ -1312,6 +1312,22 @@ KJ_TEST("JsReadableStream::tryUnwrapTs adopts TypeScript streams and rejects imp auto sourceObj = jsg::JsValue(handler.wrap( js, js.alloc(env.context, kj::heap(kData)))); KJ_EXPECT(JsReadableStream::tryUnwrapTs(js, sourceObj) == kj::none); + + // The brand is recognition, not authentication: an api symbol stays reflection-visible, + // so an object carrying a copy of it unwraps. Recognition grants nothing on its own -- + // the TypeScript internal algorithms re-check the real #-brand, so the first operation + // on the adopted impostor throws. + auto impostorObj = js.obj(); + impostorObj.setNonEnumerable(js, js.symbolInternal("kReadableStreamBrand"), js.boolean(true)); + auto impostor = KJ_ASSERT_NONNULL(JsReadableStream::tryUnwrapTs(js, jsg::JsValue(impostorObj))); + bool threw = false; + js.tryCatch( + [&]() { auto locked KJ_UNUSED = impostor.isLocked(js); }, [&](jsg::Value exception) { + threw = true; + auto e = js.exceptionToKj(kj::mv(exception)); + KJ_EXPECT(e.getDescription().contains("TypeError"), e.getDescription()); + }); + KJ_EXPECT(threw, "expected the impostor to fail the TypeScript #-brand check"); }); } diff --git a/src/workerd/api/js-readable-stream.c++ b/src/workerd/api/js-readable-stream.c++ index 198c8469a0c..15943885eb3 100644 --- a/src/workerd/api/js-readable-stream.c++ +++ b/src/workerd/api/js-readable-stream.c++ @@ -32,14 +32,27 @@ bool getReadableStreamIsDisturbed(jsg::Lock& js, jsg::JsObject obj) { return webstreams::dispatchCall(js, "getReadableStreamIsDisturbed", obj).isTrue(); } -// The TypeScript implementation's private-brand check. True only for genuine -// TypeScript-implemented ReadableStream instances (including subclasses); false for -// everything else, including proxies wrapping a stream: an own-property probe on a proxy -// would invoke its traps, and private fields do not tunnel through proxies either, so -// rejecting proxies up front matches the TS-side #-brand behavior. Runs no JavaScript -- -// recognition must work during RPC deserialization, inside V8's no-JS-execution scope -- so -// it probes for the own api-symbol brand stamped by the TypeScript constructor rather than -// asking the TS implementation. +// Recognizes the TypeScript implementation's ReadableStream (including subclasses) by the own +// api-symbol brand its constructor stamps on every instance. Runs no JavaScript -- recognition +// must work during RPC deserialization, inside V8's no-JS-execution scope -- so it probes for +// the brand rather than asking the TS implementation. Proxies answer false: an own-property +// probe on a proxy would invoke its traps, and the TS-side #-brand does not tunnel through +// proxies either. +// +// This is recognition, not authentication. An api symbol stays visible to reflection +// (Object.getOwnPropertySymbols) on every instance, so user code can read it off a real stream +// and stamp it on an object of its own: a true answer means "route this as a TypeScript +// stream", not "this is one". Genuine instances cannot lose the brand, which is stamped +// non-writable and non-configurable. What protects the consumers is that recognition grants +// nothing on its own -- every operation reached afterwards goes through the TS internal +// algorithms, whose real #-brand checks throw a TypeError on an impostor, and on both RPC +// serialize arms that rejection lands before anything is written to the wire. +// +// An impostor also cannot arrive over the wire, because V8's value serializer emits only own +// enumerable string keys (dropping any symbol-keyed brand) and will not serialize an +// unrecognized class instance at all. Every branded object reachable inside the no-JS +// deserialization scope is therefore one this runtime just built -- the premise the state +// probes in isDisturbed() and isLocked() rest on. bool isTypeScriptReadableStream(jsg::Lock& js, jsg::JsObject obj) { if (v8::Local(obj)->IsProxy()) { return false; @@ -1162,12 +1175,14 @@ void JsReadableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) { // Mirrors ReadableStream::serialize(): pumpTo() performs the lock/disturb validation, // so the stream must not be modified before that call (the encoding/length queries are // non-mutating reads). + auto& externalHandler = requireReadableStreamRpcSerializer(serializer); + IoContext& ioctx = IoContext::current(); auto encoding = getPreferredEncoding(js); auto expectedLength = tryGetLength(js, encoding); - auto sink = newReadableStreamSerializeSink(js, serializer, encoding, expectedLength); + auto sink = newReadableStreamSerializeSink(externalHandler, encoding, expectedLength); ioctx.addTask(ioctx.waitForDeferredProxy(pumpTo(js, kj::mv(sink), EndStream::YES)) .catch_([](kj::Exception&& e) { diff --git a/src/workerd/api/js-writable-stream-test.c++ b/src/workerd/api/js-writable-stream-test.c++ index b6f96de4877..1d18e07ad03 100644 --- a/src/workerd/api/js-writable-stream-test.c++ +++ b/src/workerd/api/js-writable-stream-test.c++ @@ -1319,6 +1319,22 @@ KJ_TEST("JsWritableStream::tryUnwrapTs adopts TypeScript streams and rejects imp auto sinkObj = jsg::JsValue(handler.wrap( js, js.alloc(env.context, state.makeSink(), kj::none, kj::none))); KJ_EXPECT(JsWritableStream::tryUnwrapTs(js, sinkObj) == kj::none); + + // The brand is recognition, not authentication: an api symbol stays reflection-visible, + // so an object carrying a copy of it unwraps. Recognition grants nothing on its own -- + // the TypeScript internal algorithms re-check the real #-brand, so the first operation + // on the adopted impostor throws. + auto impostorObj = js.obj(); + impostorObj.setNonEnumerable(js, js.symbolInternal("kWritableStreamBrand"), js.boolean(true)); + auto impostor = KJ_ASSERT_NONNULL(JsWritableStream::tryUnwrapTs(js, jsg::JsValue(impostorObj))); + bool threw = false; + js.tryCatch( + [&]() { auto locked KJ_UNUSED = impostor.isLocked(js); }, [&](jsg::Value exception) { + threw = true; + auto e = js.exceptionToKj(kj::mv(exception)); + KJ_EXPECT(e.getDescription().contains("TypeError"), e.getDescription()); + }); + KJ_EXPECT(threw, "expected the impostor to fail the TypeScript #-brand check"); }); } diff --git a/src/workerd/api/js-writable-stream.c++ b/src/workerd/api/js-writable-stream.c++ index 4ce74c7947a..e8fbcdd5085 100644 --- a/src/workerd/api/js-writable-stream.c++ +++ b/src/workerd/api/js-writable-stream.c++ @@ -17,13 +17,13 @@ namespace workerd::api { namespace { -// True only for genuine TypeScript-implemented WritableStream instances (including -// subclasses); false for everything else, including proxies wrapping a stream: an -// own-property probe on a proxy would invoke its traps, and private fields do not tunnel -// through proxies either, so rejecting proxies up front matches the TS-side #-brand -// behavior. Runs no JavaScript -- recognition must work during RPC deserialization, inside -// V8's no-JS-execution scope -- so it probes for the own api-symbol brand stamped by the -// TypeScript constructor rather than asking the TS implementation. +// Recognizes the TypeScript implementation's WritableStream (including subclasses) by the own +// api-symbol brand its constructor stamps on every instance; the writable counterpart of +// isTypeScriptReadableStream(), where the reasoning is spelled out. Runs no JavaScript, answers +// false for proxies, and is recognition rather than authentication: user code can stamp the +// reflection-visible brand on an object of its own, so consumers re-validate against the real +// #-brand -- here, the writer acquisition that JsWritableStream::serialize() performs before it +// writes anything to the wire. bool isTypeScriptWritableStream(jsg::Lock& js, jsg::JsObject obj) { if (v8::Local(obj)->IsProxy()) { return false; @@ -229,7 +229,6 @@ class TsWriterSink final: public WritableStreamSink { kj::Maybe> writer; bool ended = false; - // Invoke a writer method under the isolate lock, returning its (required) promise result. // Drive a writer operation through the frozen cppExports internals (never the public // writer prototype methods, which are user-patchable: a replaced write/close must not be // able to intercept or fake a stream's RPC transfer), returning its (required) promise @@ -260,7 +259,7 @@ class TsWriterSink final: public WritableStreamSink { static kj::Exception disconnectedException() { return JSG_KJ_EXCEPTION(DISCONNECTED, Error, "WritableStream received over RPC was disconnected because the remote execution context " - "has endeded."); + "has ended."); } }; diff --git a/src/workerd/api/streams/readable.c++ b/src/workerd/api/streams/readable.c++ index caa2fbde780..68456cc982f 100644 --- a/src/workerd/api/streams/readable.c++ +++ b/src/workerd/api/streams/readable.c++ @@ -703,8 +703,17 @@ kj::Own newNoDeferredProxyReadableStream( return kj::heap(kj::mv(inner), context); } -kj::Own newReadableStreamSerializeSink(jsg::Lock& js, - jsg::Serializer& serializer, +RpcSerializerExternalHandler& requireReadableStreamRpcSerializer(jsg::Serializer& serializer) { + auto& handler = JSG_REQUIRE_NONNULL(serializer.getExternalHandler(), DOMDataCloneError, + "ReadableStream can only be serialized for RPC."); + auto externalHandler = dynamic_cast(&handler); + JSG_REQUIRE(externalHandler != nullptr, DOMDataCloneError, + "ReadableStream can only be serialized for RPC."); + return *externalHandler; +} + +kj::Own newReadableStreamSerializeSink( + RpcSerializerExternalHandler& externalHandler, StreamEncoding encoding, kj::Maybe expectedLength) { // Serialize by effectively creating a `JsRpcStub` around the stream and serializing that. @@ -712,23 +721,17 @@ kj::Own newReadableStreamSerializeSink(jsg::Lock& js, // a `JsRpcStub` locally. So do the important parts of `JsRpcStub::constructor()` followed by // `JsRpcStub::serialize()`. - auto& handler = JSG_REQUIRE_NONNULL(serializer.getExternalHandler(), DOMDataCloneError, - "ReadableStream can only be serialized for RPC."); - auto externalHandler = dynamic_cast(&handler); - JSG_REQUIRE(externalHandler != nullptr, DOMDataCloneError, - "ReadableStream can only be serialized for RPC."); - IoContext& ioctx = IoContext::current(); capnp::ByteStream::Client streamCap = [&]() { - auto req = externalHandler->getExternalPusher().pushByteStreamRequest(capnp::MessageSize{2, 0}); + auto req = externalHandler.getExternalPusher().pushByteStreamRequest(capnp::MessageSize{2, 0}); KJ_IF_SOME(el, expectedLength) { req.setLengthPlusOne(el + 1); } auto pipeline = req.sendForPipeline(); - externalHandler->write([encoding, expectedLength, source = pipeline.getSource()]( - rpc::JsValue::External::Builder builder) mutable { + externalHandler.write([encoding, expectedLength, source = pipeline.getSource()]( + rpc::JsValue::External::Builder builder) mutable { auto rs = builder.initReadableStream(); rs.setStream(kj::mv(source)); rs.setEncoding(encoding); @@ -748,13 +751,15 @@ void ReadableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) { // and other common checks. It's important that we don't modify the stream in any way before // that call. + auto& externalHandler = requireReadableStreamRpcSerializer(serializer); + IoContext& ioctx = IoContext::current(); auto& controller = getController(); StreamEncoding encoding = controller.getPreferredEncoding(); auto expectedLength = controller.tryGetLength(encoding); - auto sink = newReadableStreamSerializeSink(js, serializer, encoding, expectedLength); + auto sink = newReadableStreamSerializeSink(externalHandler, encoding, expectedLength); ioctx.addTask( ioctx.waitForDeferredProxy(pumpTo(js, kj::mv(sink), true)).catch_([](kj::Exception&& e) { diff --git a/src/workerd/api/streams/readable.h b/src/workerd/api/streams/readable.h index 44156d6cf74..b3607a1ec59 100644 --- a/src/workerd/api/streams/readable.h +++ b/src/workerd/api/streams/readable.h @@ -14,6 +14,7 @@ namespace workerd::api { class ReadableStreamDefaultReader; class ReadableStreamBYOBReader; class JsReadableStream; +class RpcSerializerExternalHandler; class ReaderImpl final { public: @@ -574,16 +575,21 @@ class CountQueuingStrategy: public jsg::Object { kj::Own newNoDeferredProxyReadableStream( IoContext& context, kj::Own inner); -// Builds the wire plumbing for transferring a readable stream over RPC: requires `serializer` -// to be RPC-backed (throws DOMDataCloneError otherwise), pushes a ByteStream to the peer, -// writes the external-table entry describing it (encoding plus expected length, when known), -// and returns the local sink the stream's remaining content must be pumped into (ending the -// sink when the stream ends). Shared by ReadableStream::serialize() and JsReadableStream's -// TypeScript arm, which differ only in how the encoding/length are obtained and how the pump -// is driven. The encoding/length reads happen before the handler requirement is checked; both -// are non-mutating, so the reordering relative to the thrown error is unobservable. -kj::Own newReadableStreamSerializeSink(jsg::Lock& js, - jsg::Serializer& serializer, +// Resolves the RPC handler backing `serializer`, throwing DOMDataCloneError when there is none +// (structuredClone(), for one, cannot transfer a stream). Both readable-stream serialize arms +// resolve the handler before anything else they need, so an unsupported serialize attempt +// reports this error rather than whichever later requirement happens to fail first -- notably +// IoContext::current(), which is unavailable in global scope. +RpcSerializerExternalHandler& requireReadableStreamRpcSerializer(jsg::Serializer& serializer); + +// Builds the wire plumbing for transferring a readable stream over RPC: pushes a ByteStream to +// the peer, writes the external-table entry describing it (encoding plus expected length, when +// known), and returns the local sink the stream's remaining content must be pumped into (ending +// the sink when the stream ends). Shared by ReadableStream::serialize() and JsReadableStream's +// TypeScript arm, which differ only in how the encoding/length are obtained and how the pump is +// driven. +kj::Own newReadableStreamSerializeSink( + RpcSerializerExternalHandler& externalHandler, StreamEncoding encoding, kj::Maybe expectedLength); diff --git a/src/workerd/api/streams/writable.c++ b/src/workerd/api/streams/writable.c++ index 2a9ca18d2fe..19735ba3ab4 100644 --- a/src/workerd/api/streams/writable.c++ +++ b/src/workerd/api/streams/writable.c++ @@ -324,7 +324,7 @@ class WritableStreamRpcAdapter final: public capnp::ExplicitEndOutputStream { static kj::Exception cancellationException() { return JSG_KJ_EXCEPTION(DISCONNECTED, Error, "WritableStream received over RPC was disconnected because the remote execution context " - "has endeded."); + "has ended."); } }; @@ -494,7 +494,7 @@ class WritableStreamJsRpcAdapter final: public capnp::ExplicitEndOutputStream { static kj::Exception cancellationException() { return JSG_KJ_EXCEPTION(DISCONNECTED, Error, "WritableStream received over RPC was disconnected because the remote execution context " - "has endeded."); + "has ended."); } }; diff --git a/src/workerd/api/tests/js-rpc-test.js b/src/workerd/api/tests/js-rpc-test.js index 2eed06ad363..a07166e452c 100644 --- a/src/workerd/api/tests/js-rpc-test.js +++ b/src/workerd/api/tests/js-rpc-test.js @@ -1616,7 +1616,7 @@ export let streams = { assert.strictEqual( reason.message, 'WritableStream received over RPC was disconnected because the remote execution ' + - 'context has endeded.' + 'context has ended.' ); }