Start a native source pull or sink write before marking it in flight

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

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,
-};
+});

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 <typename Op>
+kj::Promise<void> recordCrossRequestOutcomes(
+    const TestFixture::Environment& env, kj::Vector<kj::String>& 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<void>(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<jsg::Ref<ReadableStreamNativeSource>> held;
+  testFixture.runInIoContext([&](const TestFixture::Environment& env) {
+    for (int i = 0; i < 2; i++) {
+      held.add(
+          env.js.alloc<ReadableStreamNativeSource>(env.context, kj::heap<ContentSource>(kData)));
+    }
+  });
+  for (bool byob: {false, true}) {
+    kj::Vector<kj::String> outcomes;
+    testFixture.runInIoContext([&](const TestFixture::Environment& env) -> kj::Promise<void> {
+      auto& js = env.js;
+      auto source = held[byob].addRef();
+      auto controller = byob ? makeMockController(js, state,
+                                   makeMockByobRequest(js, state,
+                                       jsg::JsUint8Array::create(js, static_cast<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;

src/workerd/api/js-readable-stream.c++

@@ -1484,10 +1484,15 @@ jsg::Promise<void> 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<void> 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 {

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 <typename Op>
+kj::Promise<void> recordCrossRequestOutcomes(
+    const TestFixture::Environment& env, kj::Vector<kj::String>& 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<void>(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<jsg::Ref<WritableStreamNativeSink>> held;
+  testFixture.runInIoContext([&](const TestFixture::Environment& env) {
+    held =
+        env.js.alloc<WritableStreamNativeSink>(env.context, state.makeSink(), kj::none, kj::none);
+  });
+  kj::Vector<kj::String> outcomes;
+  testFixture.runInIoContext([&](const TestFixture::Environment& env) -> kj::Promise<void> {
+    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<jsg::JsObject>());
+    KJ_EXPECT(exportsObj.getPrototype(js).isNull());
+
+    auto objectPrototype = KJ_ASSERT_NONNULL(js.obj().getPrototype(js).tryCast<jsg::JsObject>());
+    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) {

src/workerd/api/js-writable-stream.c++

@@ -725,10 +725,6 @@ jsg::Promise<void> 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<void> 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<void> 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;