From 6345743a16aaa2390f5577ea9568f8eb509b9f8b Mon Sep 17 00:00:00 2001 From: James M Snell Date: Tue, 6 Oct 2026 21:59:52 +0000 Subject: [PATCH 1/3] Start a native source pull or sink write before marking it in flight ReadableStreamNativeSource and WritableStreamNativeSink set pullInFlight or writeInFlight before dereferencing their IoOwns, which throws when called from a request other than the one that created them. The pull or write then failed with "Cannot perform I/O on behalf of a different request", but the flag stayed set, so every later pull threw "pull() is already in flight." and every later write threw "write() is already in flight.". A cancel would defer the source's release to a read that never started, a tee would wait on it, and an abort or pipeFrom would treat the sink as busy. The sink's observer also counted a chunk that never passed through. Today a throwing hook errors the TS stream, so the JS layer does not call the source or sink again; the stale flag is visible only from C++. With the sync fast-path autogate on, the fast path's own catch releases the source or sink first and masks it. Both now dereference the IoOwns and start the read or write first, and only then set the flag (and, for the sink, count the chunk). Pins: js-readable-stream-test.c++ "ReadableStreamNativeSource pull from another request leaves no pull in flight" (default and BYOB pulls) and js-writable-stream-test.c++ "WritableStreamNativeSink write from another request leaves no write in flight": two attempts from a second request; the first reports the cross-request error, and neither reports an operation in flight. Without the source change, both fail in their default variants (all-autogates passes through the fast path). Compatibility: the change is behind the experimental typescript_implemented_streams flag (the native source and sink back only TS streams). --- src/workerd/api/js-readable-stream-test.c++ | 55 +++++++++++++++++++++ src/workerd/api/js-readable-stream.c++ | 17 +++++-- src/workerd/api/js-writable-stream-test.c++ | 45 +++++++++++++++++ src/workerd/api/js-writable-stream.c++ | 20 +++++--- 4 files changed, 126 insertions(+), 11 deletions(-) diff --git a/src/workerd/api/js-readable-stream-test.c++ b/src/workerd/api/js-readable-stream-test.c++ index 815f88155c1..31901068180 100644 --- a/src/workerd/api/js-readable-stream-test.c++ +++ b/src/workerd/api/js-readable-stream-test.c++ @@ -767,6 +767,61 @@ KJ_TEST("ReadableStreamNativeSource BYOB pull fills the view; zero-byte EOF clos }); } +// Calls `op` twice from a request other than the one that created the object it uses, +// recording each outcome: a synchronous throw, or the promise's settlement. +template +kj::Promise recordCrossRequestOutcomes( + const TestFixture::Environment& env, kj::Vector& outcomes, Op op) { + auto& js = env.js; + for (int i = 0; i < 2; i++) { + js.tryCatch([&] { + op(js) + .then(js, [&outcomes](jsg::Lock& js) { outcomes.add(kj::str("resolved")); }, + [&outcomes](jsg::Lock& js, jsg::Value e) { + outcomes.add(kj::str("rejected: ", jsg::JsValue(e.getHandle(js)))); + }).markAsHandled(js); + }, [&](jsg::Value e) { outcomes.add(kj::str("threw: ", jsg::JsValue(e.getHandle(js)))); }); + } + return env.context.awaitJs(js, env.context.awaitIo(js, kj::Promise(kj::READY_NOW))); +} + +KJ_TEST("ReadableStreamNativeSource pull from another request leaves no pull in flight") { + // The source's IoOwn belongs to the request that created it; dereferencing it from + // another request throws. A pull that fails that way never started a read, so a + // second pull must fail the same way rather than report a pull still in flight. + TestFixture testFixture; + MockControllerState state; + // One source per read mode, both created by the first request. + kj::Vector> held; + testFixture.runInIoContext([&](const TestFixture::Environment& env) { + for (int i = 0; i < 2; i++) { + held.add( + env.js.alloc(env.context, kj::heap(kData))); + } + }); + for (bool byob: {false, true}) { + kj::Vector outcomes; + testFixture.runInIoContext([&](const TestFixture::Environment& env) -> kj::Promise { + auto& js = env.js; + auto source = held[byob].addRef(); + auto controller = byob + ? makeMockController(js, state, + makeMockByobRequest(js, state, jsg::JsUint8Array::create(js, size_t(64)), 1)) + : makeMockController(js, state, js.null()); + return recordCrossRequestOutcomes(env, outcomes, + [&, source = kj::mv(source), controller = controller.addRef(js)](jsg::Lock& js) mutable { + return source->pull(js, controller.getHandle(js), freshSignal(js)); + }); + }); + KJ_ASSERT(outcomes.size() == 2, outcomes); + for (auto& outcome: outcomes) { + KJ_EXPECT(outcome.find("already in flight"_kj) == kj::none, byob, outcomes); + } + KJ_EXPECT(outcomes[0].find("different request"_kj) != kj::none, byob, outcomes); + } + testFixture.runInIoContext([&](const TestFixture::Environment& env) { held.clear(); }); +} + KJ_TEST("ReadableStreamNativeSource BYOB under-delivery responds then closes (fused EOF)") { TestFixture testFixture; MockControllerState state; diff --git a/src/workerd/api/js-readable-stream.c++ b/src/workerd/api/js-readable-stream.c++ index 501cef8e1ce..8de536f6bc2 100644 --- a/src/workerd/api/js-readable-stream.c++ +++ b/src/workerd/api/js-readable-stream.c++ @@ -1484,10 +1484,15 @@ jsg::Promise ReadableStreamNativeSource::pullDefault( } auto& ioContext = IoContext::current(); + // Dereferencing the IoOwn throws when called from another request: start the read + // before marking it in flight, so a pull that never started leaves no flag behind. + auto& source = *active.source; + auto startLength = source.tryGetLength(StreamEncoding::IDENTITY); + auto read = source.tryRead(scratch.begin(), 1, scratch.size()); pullInFlight = true; - inFlightReadStartLength = active.source->tryGetLength(StreamEncoding::IDENTITY); + inFlightReadStartLength = startLength; return ioContext - .awaitIo(js, active.source->tryRead(scratch.begin(), 1, scratch.size()), + .awaitIo(js, kj::mv(read), [self = JSG_THIS, controller = controller.addRef(js), signal = kj::mv(signal)]( jsg::Lock& js, size_t amount) mutable { self->pullInFlight = false; @@ -1613,10 +1618,14 @@ jsg::Promise ReadableStreamNativeSource::pullByob(jsg::Lock& js, } auto& ioContext = IoContext::current(); + // As in pullDefault: start the read before marking it in flight. + auto& source = *active.source; + auto startLength = source.tryGetLength(StreamEncoding::IDENTITY); + auto read = source.tryRead(scratch.begin(), minBytes, maxBytes); pullInFlight = true; - inFlightReadStartLength = active.source->tryGetLength(StreamEncoding::IDENTITY); + inFlightReadStartLength = startLength; return ioContext - .awaitIo(js, active.source->tryRead(scratch.begin(), minBytes, maxBytes), + .awaitIo(js, kj::mv(read), [self = JSG_THIS, controller = controller.addRef(js), byobRequest = byobRequest.addRef(js), view = view.addRef(js), signal = kj::mv(signal), minBytes, elementSize](jsg::Lock& js, size_t amount) mutable { diff --git a/src/workerd/api/js-writable-stream-test.c++ b/src/workerd/api/js-writable-stream-test.c++ index cb04e45066e..481fd634127 100644 --- a/src/workerd/api/js-writable-stream-test.c++ +++ b/src/workerd/api/js-writable-stream-test.c++ @@ -877,6 +877,51 @@ KJ_TEST("WritableStreamNativeSink write fast path failure rejects and releases t }); } +// Calls `op` twice from a request other than the one that created the object it uses, +// recording each outcome: a synchronous throw, or the promise's settlement. +template +kj::Promise recordCrossRequestOutcomes( + const TestFixture::Environment& env, kj::Vector& outcomes, Op op) { + auto& js = env.js; + for (int i = 0; i < 2; i++) { + js.tryCatch([&] { + op(js) + .then(js, [&outcomes](jsg::Lock& js) { outcomes.add(kj::str("resolved")); }, + [&outcomes](jsg::Lock& js, jsg::Value e) { + outcomes.add(kj::str("rejected: ", jsg::JsValue(e.getHandle(js)))); + }).markAsHandled(js); + }, [&](jsg::Value e) { outcomes.add(kj::str("threw: ", jsg::JsValue(e.getHandle(js)))); }); + } + return env.context.awaitJs(js, env.context.awaitIo(js, kj::Promise(kj::READY_NOW))); +} + +KJ_TEST("WritableStreamNativeSink write from another request leaves no write in flight") { + // The sink's IoOwn belongs to the request that created it; dereferencing it from + // another request throws. A write that fails that way never started, so a second write + // must fail the same way rather than report a write still in flight. + TestFixture testFixture; + SinkState state; + kj::Maybe> held; + testFixture.runInIoContext([&](const TestFixture::Environment& env) { + held = + env.js.alloc(env.context, state.makeSink(), kj::none, kj::none); + }); + kj::Vector outcomes; + testFixture.runInIoContext([&](const TestFixture::Environment& env) -> kj::Promise { + auto sink = KJ_ASSERT_NONNULL(held).addRef(); + return recordCrossRequestOutcomes(env, outcomes, [sink = kj::mv(sink)](jsg::Lock& js) mutable { + return sink->write(js, jsg::JsValue(jsg::JsUint8Array::create(js, "hi"_kjb))); + }); + }); + KJ_ASSERT(outcomes.size() == 2, outcomes); + for (auto& outcome: outcomes) { + KJ_EXPECT(outcome.find("already in flight"_kj) == kj::none, outcomes); + } + KJ_EXPECT(outcomes[0].find("different request"_kj) != kj::none, outcomes); + KJ_EXPECT(state.written.size() == 0); + testFixture.runInIoContext([&](const TestFixture::Environment& env) { held = kj::none; }); +} + KJ_TEST("JsWritableStream create under the flag produces a TypeScript-backed stream") { auto fixture = makeTsStreamsFixture(); SinkState state; diff --git a/src/workerd/api/js-writable-stream.c++ b/src/workerd/api/js-writable-stream.c++ index 0b0536c8025..f91859d1209 100644 --- a/src/workerd/api/js-writable-stream.c++ +++ b/src/workerd/api/js-writable-stream.c++ @@ -725,10 +725,6 @@ jsg::Promise WritableStreamNativeSink::write( } } - KJ_IF_SOME(o, observer) { - o->onChunkEnqueued(len); - } - writeInFlight = true; // Durable Object output gate: the write's bytes must not become externally // observable while an output lock is pending, exactly like the legacy internal // controller, which stores an output lock with every queued write event and awaits @@ -737,15 +733,25 @@ jsg::Promise WritableStreamNativeSink::write( // settlement, detach() cancels the wrapped wait+write before it releases the sink, // and pipeFrom() refuses to move the sink while it is set. The write's I/O runs // outside the isolate lock; the copied bytes ride the promise. + // + // Dereferencing the IoOwns throws when called from another request: the write is + // started before it is marked in flight (and counted by the observer), so a write + // that never started leaves no flag behind. + auto& sink = *active.sink; + auto& canceler = *active.canceler; kj::Promise promise = nullptr; KJ_IF_SOME(lock, ioContext.waitForOutputLocksIfNecessary()) { - promise = lock.then([&sink = *active.sink, data = kj::mv(data)]() mutable { + promise = lock.then([&sink, data = kj::mv(data)]() mutable { return sink.write(data.asPtr()).attach(kj::mv(data)); }); } else { - promise = active.sink->write(data.asPtr()).attach(kj::mv(data)); + promise = sink.write(data.asPtr()).attach(kj::mv(data)); } - promise = active.canceler->wrap(kj::mv(promise)); + promise = canceler.wrap(kj::mv(promise)); + KJ_IF_SOME(o, observer) { + o->onChunkEnqueued(len); + } + writeInFlight = true; return ioContext .awaitIo(js, kj::mv(promise), [self = JSG_THIS, len](jsg::Lock& js) mutable { self->writeInFlight = false; From 662d443e67230b994ddd929fe3d3810066b1d40e Mon Sep 17 00:00:00 2001 From: James M Snell Date: Tue, 6 Oct 2026 22:05:37 +0000 Subject: [PATCH 2/3] Freeze the TS streams cpp_exports table and give it a null prototype The C++ bridge fetches every TypeScript streams entry point with getCppExport(), an ordinary property get on the webstreams/cpp_exports module's exports. That object was a plain object literal spreading the two frozen halves (readable.ts and writable.ts cppExports): unfrozen, and inheriting from Object.prototype. Every name the bridge asks for is an own property today, so nothing resolves through the prototype now, and user code cannot reach the table itself (it lives in the bootstrap's C++-side module cache). But a name that went missing, through a rename or a typo, would resolve to whatever user code had put on Object.prototype, and the bridge would call it with internal objects. The table is now frozen and has a null prototype, so such a lookup fails the bridge's KJ_REQUIRE_NONNULL instead. Pin: js-writable-stream-test.c++ "The cpp_exports table has a null prototype and is frozen" checks the prototype, that a function planted on Object.prototype under a missing name is not returned by getCppExport(), and that a new property does not stick. Without the source change, all three assertions fail. The webstreams AGENTS.md key rules record both properties. Compatibility: no user-visible change; the table is internal to the experimental typescript_implemented_streams implementation. --- src/per_isolate/webstreams/AGENTS.md | 4 ++++ src/per_isolate/webstreams/cpp_exports.ts | 11 ++++++++-- src/workerd/api/js-writable-stream-test.c++ | 23 +++++++++++++++++++++ 3 files changed, 36 insertions(+), 2 deletions(-) diff --git a/src/per_isolate/webstreams/AGENTS.md b/src/per_isolate/webstreams/AGENTS.md index 9fcd0aea676..aa4be11acd7 100644 --- a/src/per_isolate/webstreams/AGENTS.md +++ b/src/per_isolate/webstreams/AGENTS.md @@ -165,6 +165,10 @@ connection. argument — the source checks `signal.aborted` before delivery and stashes bytes for redelivery if aborted (race buffering lives source-side; the JS conduit is uniformly bufferless). +- The `webstreams/cpp_exports` table, which the C++ bridge reads with an + ordinary property get (`getCppExport`), is frozen and has a null + prototype, so a name missing from it cannot resolve through + `Object.prototype`. Keep both when adding exports. - `nativeStreamInternals` (markers, extraction symbols, conduit construction) is module-private, consumed only by readable.ts/writable.ts and the C++ bridge via the API-symbol registry. The C++ mocks in diff --git a/src/per_isolate/webstreams/cpp_exports.ts b/src/per_isolate/webstreams/cpp_exports.ts index 78f5996f490..e4d385877fb 100644 --- a/src/per_isolate/webstreams/cpp_exports.ts +++ b/src/per_isolate/webstreams/cpp_exports.ts @@ -1,9 +1,16 @@ 'use strict'; +const { ObjectFreeze } = primordials; + const { cppExports: readableCppExports } = require('webstreams/readable'); const { cppExports: writableCppExports } = require('webstreams/writable'); -module.exports = { +// The C++ bridge looks names up here with an ordinary property get +// (getCppExport in js-streams-bridge.c++). The table has a null prototype, +// so a name missing from it fails the lookup instead of resolving through a +// polluted Object.prototype, and it is frozen, like the two halves. +module.exports = ObjectFreeze({ + __proto__: null, ...readableCppExports, ...writableCppExports, -}; +}); diff --git a/src/workerd/api/js-writable-stream-test.c++ b/src/workerd/api/js-writable-stream-test.c++ index 481fd634127..6a51d81a518 100644 --- a/src/workerd/api/js-writable-stream-test.c++ +++ b/src/workerd/api/js-writable-stream-test.c++ @@ -1637,6 +1637,29 @@ KJ_TEST("JsWritableStream create TS arm: closure waitable rejection skips the si KJ_EXPECT(!state.aborted); } +KJ_TEST("The cpp_exports table has a null prototype and is frozen") { + // getCppExport() reads the table with an ordinary property get, so a name missing from + // it must not resolve through Object.prototype, where user code can put a function. + 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()); + KJ_EXPECT(exportsObj.getPrototype(js).isNull()); + + auto objectPrototype = KJ_ASSERT_NONNULL(js.obj().getPrototype(js).tryCast()); + auto planted = exportsObj.get(js, "WritableStream"_kj); + objectPrototype.set(js, "plantedCppExport"_kj, planted); + KJ_EXPECT_THROW(FAILED, webstreams::getCppExport(js, "plantedCppExport"_kj)); + objectPrototype.delete_(js, "plantedCppExport"_kj); + + // Frozen: a new property does not stick. + js.tryCatch([&] { exportsObj.set(js, "addedCppExport"_kj, planted); }, [](jsg::Value) {}); + KJ_EXPECT(!exportsObj.has(js, "addedCppExport"_kj, jsg::JsObject::HasOption::OWN)); + }); +} + KJ_TEST("JsWritableStream serialize of a TypeScript-backed stream requires an RPC serializer") { auto fixture = makeTsStreamsFixture(); fixture.runInIoContext([&](const TestFixture::Environment& env) { From 6edb0715af88ca9ed9fb2b924153a75924791481 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Fri, 9 Oct 2026 14:37:16 +0000 Subject: [PATCH 3/3] Use static_cast in js-readable-stream-test to satisfy clang-tidy --- src/workerd/api/js-readable-stream-test.c++ | 8 ++++---- src/workerd/api/js-readable-stream.c++ | 4 ++-- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/workerd/api/js-readable-stream-test.c++ b/src/workerd/api/js-readable-stream-test.c++ index 31901068180..551cb406cde 100644 --- a/src/workerd/api/js-readable-stream-test.c++ +++ b/src/workerd/api/js-readable-stream-test.c++ @@ -804,10 +804,10 @@ KJ_TEST("ReadableStreamNativeSource pull from another request leaves no pull in testFixture.runInIoContext([&](const TestFixture::Environment& env) -> kj::Promise { auto& js = env.js; auto source = held[byob].addRef(); - auto controller = byob - ? makeMockController(js, state, - makeMockByobRequest(js, state, jsg::JsUint8Array::create(js, size_t(64)), 1)) - : makeMockController(js, state, js.null()); + auto controller = byob ? makeMockController(js, state, + makeMockByobRequest(js, state, + jsg::JsUint8Array::create(js, static_cast(64)), 1)) + : makeMockController(js, state, js.null()); return recordCrossRequestOutcomes(env, outcomes, [&, source = kj::mv(source), controller = controller.addRef(js)](jsg::Lock& js) mutable { return source->pull(js, controller.getHandle(js), freshSignal(js)); diff --git a/src/workerd/api/js-readable-stream.c++ b/src/workerd/api/js-readable-stream.c++ index 8de536f6bc2..57ee9078091 100644 --- a/src/workerd/api/js-readable-stream.c++ +++ b/src/workerd/api/js-readable-stream.c++ @@ -1490,7 +1490,7 @@ jsg::Promise ReadableStreamNativeSource::pullDefault( auto startLength = source.tryGetLength(StreamEncoding::IDENTITY); auto read = source.tryRead(scratch.begin(), 1, scratch.size()); pullInFlight = true; - inFlightReadStartLength = startLength; + inFlightReadStartLength = kj::mv(startLength); return ioContext .awaitIo(js, kj::mv(read), [self = JSG_THIS, controller = controller.addRef(js), signal = kj::mv(signal)]( @@ -1623,7 +1623,7 @@ jsg::Promise ReadableStreamNativeSource::pullByob(jsg::Lock& js, auto startLength = source.tryGetLength(StreamEncoding::IDENTITY); auto read = source.tryRead(scratch.begin(), minBytes, maxBytes); pullInFlight = true; - inFlightReadStartLength = startLength; + inFlightReadStartLength = kj::mv(startLength); return ioContext .awaitIo(js, kj::mv(read), [self = JSG_THIS, controller = controller.addRef(js),