Implement jsrpc serialization/deserialization for TS web streams

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
 

src/per_isolate/webstreams/readable.ts

@@ -269,6 +269,15 @@ let readableStreamDefaultReaderRead: <R>(
 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<R> {
     underlyingSource: UnderlyingSource<R> = {},
     strategy: QueuingStrategy<R> = {}
   ) {
+    // 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

src/per_isolate/webstreams/writable.ts

@@ -152,6 +152,15 @@ let getWritableStreamController: <W>(
 // 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: <W>(stream: WritableStream<W>) => void;
 let isWritableStreamPendingClosure: <W>(stream: WritableStream<W>) => boolean;
 // Permanently neutralizes a stream on behalf of the C++ bridge (e.g. a
@@ -223,6 +232,10 @@ let writerWriteInternal: <W>(
 let writerCloseInternal: <W>(
   writer: WritableStreamDefaultWriter<W>
 ) => Promise<void>;
+let writerAbortInternal: <W>(
+  writer: WritableStreamDefaultWriter<W>,
+  reason: unknown
+) => Promise<void>;
 let writerReleaseInternal: <W>(writer: WritableStreamDefaultWriter<W>) => void;
 let getWriterReadyPromiseInternal: <W>(
   writer: WritableStreamDefaultWriter<W>
@@ -634,6 +647,13 @@ class WritableStream<W = unknown> {
     underlyingSink: UnderlyingSink<W> = {},
     strategy: QueuingStrategy<W> = {}
   ) {
+    // 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 = <W>(
+      writer: WritableStreamDefaultWriter<W>,
+      reason: unknown
+    ) => {
+      const stream = writer.#stream;
+      if (stream === undefined) {
+        return PromiseReject(
+          new TypeError('This writer has been released')
+        ) as Promise<void>;
+      }
+      return writableStreamAbort(stream, reason);
+    };
+
     writerCloseInternal = <W>(writer: WritableStreamDefaultWriter<W>) => {
       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<W>(
+    stream: WritableStream<W>
+  ): WritableStreamDefaultWriter<W> {
+    return new WritableStreamDefaultWriter<W>(stream);
+  },
+  writableStreamWriterWrite: <W>(
+    writer: WritableStreamDefaultWriter<W>,
+    chunk: W
+  ): Promise<void> => writerWriteInternal(writer, chunk),
+  writableStreamWriterClose: <W>(
+    writer: WritableStreamDefaultWriter<W>
+  ): Promise<void> => writerCloseInternal(writer),
+  writableStreamWriterAbort: <W>(
+    writer: WritableStreamDefaultWriter<W>,
+    reason: unknown
+  ): Promise<void> => writerAbortInternal(writer, reason),
 });
 
 module.exports = {

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<ReadableStreamNativeSource>(env.context, kj::heap<ContentSource>(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<size_t> 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<EncodedLengthSource>());
+      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<EncodedLengthSource>());
+      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<size_t> tryRead(void* buffer, size_t minBytes, size_t maxBytes) override {
+    auto amount = kj::min(maxBytes, data.size() - offset);
+    kj::arrayPtr(static_cast<kj::byte*>(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<void> {
+    auto& js = env.js;
+
+    auto source =
+        js.alloc<ReadableStreamNativeSource>(env.context, kj::heap<EncodedContentSource>(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<void> {