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 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..224742e0167 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 @@ -223,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 @@ -634,6 +647,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 @@ -1202,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) { @@ -1633,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-readable-stream-test.c++ b/src/workerd/api/js-readable-stream-test.c++ index 8d532c8c45b..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"); }); } @@ -1724,8 +1740,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 +1754,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 +1798,114 @@ 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("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; + 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..15943885eb3 100644 --- a/src/workerd/api/js-readable-stream.c++ +++ b/src/workerd/api/js-readable-stream.c++ @@ -32,12 +32,32 @@ 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 (private fields do not tunnel -// through proxies, deliberately matching the TS-side behavior). +// 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) { - 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 +599,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 +648,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 +674,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)); } } @@ -765,6 +795,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) { @@ -1116,7 +1172,24 @@ 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). + auto& externalHandler = requireReadableStreamRpcSerializer(serializer); + + IoContext& ioctx = IoContext::current(); + + auto encoding = getPreferredEncoding(js); + auto expectedLength = tryGetLength(js, encoding); + + auto sink = newReadableStreamSerializeSink(externalHandler, 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! + })); } } } @@ -1457,6 +1530,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..bfc8f7d86ed 100644 --- a/src/workerd/api/js-readable-stream.h +++ b/src/workerd/api/js-readable-stream.h @@ -103,10 +103,6 @@ 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. static JsReadableStream create( jsg::Lock& js, IoContext& ioContext, kj::Own source); @@ -158,6 +154,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 +386,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); 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/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-test.c++ b/src/workerd/api/js-writable-stream-test.c++ index b4d2a87d9a9..1d18e07ad03 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; @@ -1297,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 ec0e0f2a09c..e8fbcdd5085 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 @@ -14,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). +// 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) { - 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) { @@ -126,6 +135,134 @@ 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, "writableStreamWriterWrite"_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, "writableStreamWriterWrite"_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, "writableStreamWriterClose"_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; + + // 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 op, kj::Maybe arg = kj::none) { + auto& w = KJ_UNWRAP_OR(writer, { kj::throwFatalException(disconnectedException()); }); + 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 operation did not return a promise", op)); + } + + 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)); + // 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)); + })); + } + + static kj::Exception disconnectedException() { + return JSG_KJ_EXCEPTION(DISCONNECTED, Error, + "WritableStream received over RPC was disconnected because the remote execution context " + "has ended."); + } +}; + } // namespace JsWritableStream::JsWritableStream(jsg::Ref stream) @@ -199,6 +336,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)); } } @@ -359,7 +502,43 @@ 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 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), "acquireWritableStreamWriter 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); + }); } } } @@ -401,10 +580,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/sockets.c++ b/src/workerd/api/sockets.c++ index 5d3bcc3a915..d9defde8ed9 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,36 @@ 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: 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(); + 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 +913,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 +923,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 +994,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 +1039,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/streams/readable.c++ b/src/workerd/api/streams/readable.c++ index 05c5e8389f9..68456cc982f 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 @@ -702,37 +703,35 @@ 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. - // 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()`. - +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; +} - // 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. +kj::Own newReadableStreamSerializeSink( + RpcSerializerExternalHandler& externalHandler, + 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()`. 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}); + 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); @@ -744,7 +743,23 @@ 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. + + auto& externalHandler = requireReadableStreamRpcSerializer(serializer); + + IoContext& ioctx = IoContext::current(); + + auto& controller = getController(); + StreamEncoding encoding = controller.getPreferredEncoding(); + auto expectedLength = controller.tryGetLength(encoding); + + auto sink = newReadableStreamSerializeSink(externalHandler, encoding, expectedLength); ioctx.addTask( ioctx.waitForDeferredProxy(pumpTo(js, kj::mv(sink), true)).catch_([](kj::Exception&& e) { @@ -754,13 +769,42 @@ void ReadableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) { })); } -jsg::Ref ReadableStream::deserialize( +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); + } + + // 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"); @@ -775,8 +819,9 @@ jsg::Ref ReadableStream::deserialize( kj::Own in = ioctx.getExternalPusher()->unwrapStream(rs.getStream()); - return js.alloc(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 4840601d7dc..b3607a1ec59 100644 --- a/src/workerd/api/streams/readable.h +++ b/src/workerd/api/streams/readable.h @@ -13,6 +13,8 @@ namespace workerd::api { class ReadableStreamDefaultReader; class ReadableStreamBYOBReader; +class JsReadableStream; +class RpcSerializerExternalHandler; class ReaderImpl final { public: @@ -474,7 +476,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); @@ -567,4 +575,32 @@ class CountQueuingStrategy: public jsg::Object { kj::Own newNoDeferredProxyReadableStream( IoContext& context, kj::Own inner); +// 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); + +// 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 523c215127a..19735ba3ab4 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 @@ -323,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."); } }; @@ -493,12 +494,18 @@ 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."); } }; } // 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 +527,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 { @@ -554,13 +561,43 @@ void WritableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) { } } -jsg::Ref WritableStream::deserialize( +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); + } + + // 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"); @@ -575,8 +612,8 @@ 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()); + 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 46444760995..41c5eaa5ce8 100644 --- a/src/workerd/api/streams/writable.h +++ b/src/workerd/api/streams/writable.h @@ -9,8 +9,14 @@ #include #include +namespace capnp { +class ExplicitEndOutputStream; +} + namespace workerd::api { +class JsWritableStream; + class WritableStreamDefaultWriter: public jsg::Object, public WritableStreamController::Writer { public: explicit WritableStreamDefaultWriter(); @@ -199,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); @@ -216,4 +228,29 @@ 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); + +// 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/tests/BUILD.bazel b/src/workerd/api/tests/BUILD.bazel index 05116d77b3d..5614e1fc107 100644 --- a/src/workerd/api/tests/BUILD.bazel +++ b/src/workerd/api/tests/BUILD.bazel @@ -610,6 +610,25 @@ 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-streams-crossflag-test.wd-test", + args = ["--experimental"], + 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"], @@ -976,6 +995,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/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", + ], +); 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", + ], +); 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", + ], +); 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.' ); } diff --git a/src/workerd/api/worker-rpc.c++ b/src/workerd/api/worker-rpc.c++ index 932a55291c0..e46ef5da8d8 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 @@ -70,6 +71,96 @@ 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)) { + // 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()); + 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; + 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. + 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 +215,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{ @@ -1984,6 +2081,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..67bd2a6b7c7 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 @@ -96,6 +98,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; @@ -119,6 +129,35 @@ 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, 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 + // 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() { @@ -129,6 +168,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/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 { diff --git a/src/workerd/util/autogate.h b/src/workerd/util/autogate.h index 9f8b3eb5693..66860b924e2 100644 --- a/src/workerd/util/autogate.h +++ b/src/workerd/util/autogate.h @@ -106,7 +106,16 @@ 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; 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 // --------------------------------------------------------------------------------------