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-readable-stream-test.c++ b/src/workerd/api/js-readable-stream-test.c++ index 815f88155c1..551cb406cde 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, 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)); + }); + }); + 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..57ee9078091 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 = kj::mv(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 = kj::mv(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..6a51d81a518 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; @@ -1592,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) { 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;