Release the pipe source when validation fails after locking (#7498)

Author: jasnell

src/per_isolate/webstreams/readable.ts

@@ -18,6 +18,7 @@ import type {
   UnderlyingDefaultSource,
   UnderlyingSource,
   WritableStream as WritableStreamType,
+  WritableStreamDefaultWriter as WritableStreamDefaultWriterType,
 } from './types';
 import type {
   ByteQueueEntry,
@@ -227,7 +228,7 @@ let readableStreamCancel: <R>(
 let readableStreamPipeThroughTo: <R>(
   source: ReadableStream<R>,
   destination: WritableStreamType<R>,
-  options?: StreamPipeOptions
+  options: ConvertedPipeOptions
 ) => Promise<void>;
 let readableStreamPipeTo: <R>(
   source: ReadableStream<R>,
@@ -2911,35 +2912,36 @@ class ReadableStreamDrainingReader<R> {
   }
 }
 
-// The pipe (spec ReadableStreamPipeTo). Internal operations only on both
-// ends — locks are held for the duration. Chunks are read only while the
-// destination desires them: the pump moves buffered chunks until
-// desiredSize runs out, and reads the next chunk when none is buffered. It
-// resumes from the destination's ready hook when a write completes, and
-// when a read delivers, so a buffered backlog moves without a promise per
-// wake-up. Writes are not awaited individually (the close path queues
-// behind them, and write failures surface through the destination's closed
-// promise). A shutdown waits for the writes made before running its
-// action, including the write of a chunk that a pending read delivers
-// during the wait.
-function pipeToInternal<R>(
-  source: ReadableStream<R>,
-  destination: WritableStreamType<R>,
-  options: StreamPipeOptions = kEmptyDictionary as StreamPipeOptions
-): Promise<void> {
-  // Spec-mandated read order (§4.9.1): preventAbort, preventCancel,
-  // preventClose, signal. WPT piping/throwing-options.any.js verifies
-  // that getter side-effects occur in exactly this sequence.
-  const preventAbort = !!options.preventAbort;
-  const preventCancel = !!options.preventCancel;
-  const preventClose = !!options.preventClose;
-  const signal = options.signal;
+// StreamPipeOptions after WebIDL conversion: plain data, no getters.
+interface ConvertedPipeOptions {
+  readonly preventAbort: boolean;
+  readonly preventCancel: boolean;
+  readonly preventClose: boolean;
+  readonly signal: AbortSignal | undefined;
+}
+
+// WebIDL conversion of StreamPipeOptions. The pipe methods run it before
+// their locked checks, so a getter cannot change a lock after it is checked.
+// Members are read once each in spec order (WPT
+// piping/throwing-options.any.js).
+function convertPipeOptions(options: unknown): ConvertedPipeOptions {
+  // WebIDL: null and undefined become {}.
+  if (options == null) options = kEmptyDictionary;
+  if (!isActualObject(options)) {
+    throw new TypeError('Pipe options must be an object');
+  }
+  const dict = options as StreamPipeOptions;
+  const preventAbort = !!dict.preventAbort;
+  const preventCancel = !!dict.preventCancel;
+  const preventClose = !!dict.preventClose;
+  const signal = dict.signal;
   if (signal !== undefined) {
     // Brand check. Under the modern JSG layout the captured `aborted` getter
     // throws for non-AbortSignal receivers; under the instance-property
     // layout (old compat dates) the capture is a plain read that cannot
     // brand-check, so additionally require the boolean a genuine signal's
-    // own data property carries.
+    // own data property carries. (A forged {aborted: boolean} slips through
+    // under old dates only; the native fast path's C++ unwrap rejects it.)
     let aborted: unknown;
     try {
       aborted = AbortSignalAbortedGet(signal);
@@ -2950,10 +2952,45 @@ function pipeToInternal<R>(
       throw new TypeError('options.signal must be an AbortSignal');
     }
   }
+  return {
+    __proto__: null,
+    preventAbort,
+    preventCancel,
+    preventClose,
+    signal,
+  } as ConvertedPipeOptions;
+}
+
+// The pipe (spec ReadableStreamPipeTo). Internal operations only on both
+// ends — locks are held for the duration. Chunks are read only while the
+// destination desires them: the pump moves buffered chunks until
+// desiredSize runs out, and reads the next chunk when none is buffered. It
+// resumes from the destination's ready hook when a write completes, and
+// when a read delivers, so a buffered backlog moves without a promise per
+// wake-up. Writes are not awaited individually (the close path queues
+// behind them, and write failures surface through the destination's closed
+// promise). A shutdown waits for the writes made before running its
+// action, including the write of a chunk that a pending read delivers
+// during the wait.
+function pipeToInternal<R>(
+  source: ReadableStream<R>,
+  destination: WritableStreamType<R>,
+  options: ConvertedPipeOptions
+): Promise<void> {
+  const { preventAbort, preventCancel, preventClose, signal } = options;
 
-  // Lock both ends.
+  // Lock both ends. The callers have checked both locks; release the reader
+  // if the writer still cannot be acquired.
   const reader = new ReadableStreamDefaultReader<R>(source);
-  const writer = writableInternals.acquireWriter(destination);
+  let writer: WritableStreamDefaultWriterType<R>;
+  try {
+    writer = writableInternals.acquireWriter(destination);
+  } catch (e) {
+    // The catch here is purely defensive. The acquireWriter
+    // should not actually throw.
+    readableStreamReaderGenericRelease(reader);
+    throw e;
+  }
   setReadableStreamDisturbed(source);
 
   const { promise, resolve, reject } =
@@ -3529,7 +3566,7 @@ class ReadableStream<R> {
     readableStreamPipeThroughTo = <R>(
       source: ReadableStream<R>,
       destination: WritableStreamType<R>,
-      options?: StreamPipeOptions
+      options: ConvertedPipeOptions
     ) => {
       // The pending-closure gate (see #pendingClosure). The prototype
       // pipeThrough reaches pipeToInternal through here WITHOUT passing
@@ -3558,9 +3595,20 @@ class ReadableStream<R> {
     readableStreamPipeTo = <R>(
       source: ReadableStream<R>,
       destination: WritableStreamType<R>,
-      options: StreamPipeOptions = kEmptyDictionary as StreamPipeOptions
+      options: unknown = kEmptyDictionary
     ): Promise<void> => {
       try {
+        // WebIDL argument conversion first: the destination brand check,
+        // then the options, all before the locked checks and before the
+        // fast path below permanently consumes both endpoints. Both paths
+        // receive the converted values and never re-read the user's object.
+        if (!writableInternals.isWritableStream(destination)) {
+          throw new TypeError(
+            "Failed to execute 'pipeTo': destination is not a WritableStream"
+          );
+        }
+        const converted = convertPipeOptions(options);
+        const { preventAbort, preventCancel, preventClose } = converted;
         if (isReadableStreamLocked(source)) {
           throw new TypeError('Cannot pipe a stream that is locked');
         }
@@ -3570,47 +3618,6 @@ class ReadableStream<R> {
         if (source.#pendingClosure) {
           throw pendingClosureError();
         }
-        // WebIDL: null coerces to {} for optional dictionaries.
-        if (options === null) options = kEmptyDictionary as StreamPipeOptions;
-        if (!isActualObject(options)) {
-          throw new TypeError('Pipe options must be an object');
-        }
-        // Convert the options ONCE, before any extraction: the fast path
-        // below permanently consumes both endpoints, so validation (and
-        // the user-observable getter side effects, in the spec-mandated
-        // §4.9.1 order) must happen while both streams are still
-        // untouched. Both paths below receive the converted values as
-        // plain data properties; neither re-reads the user's options
-        // object.
-        const preventAbort = !!options.preventAbort;
-        const preventCancel = !!options.preventCancel;
-        const preventClose = !!options.preventClose;
-        const signal = options.signal;
-        if (signal !== undefined) {
-          // Brand check (same check and error text as the JS pump). Under
-          // the modern JSG layout the captured `aborted` getter throws for
-          // non-AbortSignal receivers; under the instance-property layout
-          // (old compat dates) the capture is a plain read that cannot
-          // brand-check, so additionally require the boolean a genuine
-          // signal's own data property carries. (A deliberately forged
-          // {aborted: boolean} object can slip through under old dates
-          // only; the fast path's C++ typed unwrap still rejects it.)
-          let aborted: unknown;
-          try {
-            aborted = AbortSignalAbortedGet(signal);
-          } catch {
-            throw new TypeError('options.signal must be an AbortSignal');
-          }
-          if (typeof aborted !== 'boolean') {
-            throw new TypeError('options.signal must be an AbortSignal');
-          }
-        }
-        const converted = {
-          preventAbort,
-          preventCancel,
-          preventClose,
-          signal,
-        } as StreamPipeOptions;
 
         // PIPE DISPATCH: if both source and dest are native-backed, take
         // the fast path -- extract both and let the sink's pipeFrom hook
@@ -3637,16 +3644,16 @@ class ReadableStream<R> {
         // TODO(streams-ts): revisit extending the fast path to the
         // prevent* options (e.g. reversible extraction) so those pipes can
         // also run entirely at the C++ layer.
-        // Brand-check the destination before probing for native markers so
-        // a Proxy's getOwnPropertyDescriptor trap cannot observe the symbol.
+        // The destination is brand-checked above, so a Proxy's
+        // getOwnPropertyDescriptor trap cannot observe the symbol.
         const sourceExtractor = ObjectGetOwnPropertyDescriptor(
           source,
           kExtractNativeSource
         )?.value as ((this: ReadableStream<R>) => object) | undefined;
-        const sinkExtractor = writableInternals.isWritableStream(destination)
-          ? (ObjectGetOwnPropertyDescriptor(destination, kExtractNativeSink)
-              ?.value as ((this: object) => object) | undefined)
-          : undefined;
+        const sinkExtractor = ObjectGetOwnPropertyDescriptor(
+          destination,
+          kExtractNativeSink
+        )?.value as ((this: object) => object) | undefined;
         if (
           sourceExtractor !== undefined &&
           sinkExtractor !== undefined &&
@@ -3674,7 +3681,7 @@ class ReadableStream<R> {
             unknown
           >;
           const pipeFrom = nativeSink.pipeFrom as
-            | ((source: object, opts: StreamPipeOptions) => Promise<void>)
+            | ((source: object, opts: ConvertedPipeOptions) => Promise<void>)
             | undefined;
           if (pipeFrom === undefined) {
             throw new TypeError(
@@ -4356,29 +4363,30 @@ class ReadableStream<R> {
     options: StreamPipeOptions = kEmptyDictionary as StreamPipeOptions
   ): ReadableStreamType<T> {
     assertIsReadableStream(this);
-    if (isReadableStreamLocked(this)) {
-      throw new TypeError('Cannot pipe a stream that is locked');
-    }
-    // WebIDL dictionary conversion: alphabetical member order means
-    // "readable" is read (and brand-checked) BEFORE "writable".  The
-    // WPT pipe-through.any.js tests verify that a bad readable stops
-    // the writable getter from ever being called.
+    // WebIDL argument conversion precedes the locked checks, so no user
+    // code runs between those checks and pipeToInternal taking the locks.
+    // Dictionary members are read in alphabetical order: "readable" is read
+    // and brand-checked before "writable" (WPT pipe-through.any.js).
     const readable = transform.readable;
     if (!isReadableStream(readable)) {
       throw new TypeError(
         "Failed to execute 'pipeThrough': readable is not a ReadableStream"
       );
     }
     const writable = transform.writable;
-    if (writable.locked) {
-      throw new TypeError('Cannot pipe to a locked writable stream');
+    if (!writableInternals.isWritableStream(writable)) {
+      throw new TypeError(
+        "Failed to execute 'pipeThrough': writable is not a WritableStream"
+      );
     }
-    // WebIDL: null coerces to {} for optional dictionaries.
-    if (options === null) options = kEmptyDictionary as StreamPipeOptions;
-    if (!isActualObject(options)) {
-      throw new TypeError('Pipe options must be an object');
+    const converted = convertPipeOptions(options);
+    if (isReadableStreamLocked(this)) {
+      throw new TypeError('Cannot pipe a stream that is locked');
+    }
+    if (writableInternals.isWritableStreamLocked(writable)) {
+      throw new TypeError('Cannot pipe to a locked writable stream');
     }
-    const promise = readableStreamPipeThroughTo(this, writable, options);
+    const promise = readableStreamPipeThroughTo(this, writable, converted);
     markPromiseHandled(promise);
     return readable;
   }

src/tests/streams/piping/AGENTS.md

@@ -40,8 +40,9 @@ ends, preventAbort/preventCancel incl. TRUTHY coercion, dest stays
 usable under preventAbort); option plumbing (getter order
 [preventAbort, preventCancel, preventClose, signal], throwing-getter
 identity with no locks taken, bad-signal TypeError); pipeThrough
-locked-endpoint sync throws; custom error type/instance preservation;
-cancel-propagation through native identity AND JS transforms (source
+locked-endpoint sync throws; a bad destination, an option getter that
+locks it, or a shadowed `locked` fails without locking the source;
+custom error type/instance preservation; cancel-propagation through native identity AND JS transforms (source
 ends CLOSED, all locks release); external close()/abort() on a piped
 (locked) destination rejects while the pipe proceeds; backward write-
 error propagation with hook identity; backpressure through
@@ -75,7 +76,7 @@ the source FIRST, then releasing the write (`pipeStopsPullingWhenDestStalls`).
 | Module | Coverage |
 | --- | --- |
 | `pipe-matrix.js` | migrated pipe-streams-test.js wholesale (35): pipeThrough + pipeTo across JS↔native in all directions, prevent* combos, pre-aborted and mid-read AbortSignals, tee'd pipes, queued-destination close (ledger #1-#4, #13) |
-| `api-surface.js` | brand checks (ledger #5), option getter order, throwing getters, invalid signal, locked pipeThrough endpoints |
+| `api-surface.js` | brand checks (ledger #5), option getter order, throwing getters, invalid signal, locked pipeThrough endpoints, lock safety when validation fails |
 | `abort-signal.js` | the pipe's AbortSignal is an abort algorithm: a synthetic 'abort' event is ignored, a listener's stopImmediatePropagation() cannot block the abort, an abort after the pipe settles does nothing |
 | `error-propagation.js` | forward matrix (starts-errored × prevent* × truthy), hwm-0 dest (ledger #6), custom-error preservation (migrated from streams-error-edge-cases-test.js) |
 | `close-propagation.js` | the WPT-disabled backward territory, bounded: external close/abort on piped dest, write-throw backward propagation, idle dest-controller error and one with a write in flight (ledger #7) |

src/tests/streams/piping/api-surface.js

@@ -118,3 +118,64 @@ export const pipeThroughLockedEndpoints = {
     strictEqual(rs2.locked, false);
   },
 };
+
+// A TypeError without internal names (`#writer`), and no lock taken.
+const cleanTypeError = (e) => e.name === 'TypeError' && !/#/.test(e.message);
+
+// A non-WritableStream destination fails before the source is locked.
+export const badDestinationLeavesSourceUnlocked = {
+  async test() {
+    const rs = new ReadableStream({});
+    throws(
+      () => rs.pipeThrough({ readable: new ReadableStream(), writable: {} }),
+      cleanTypeError
+    );
+    strictEqual(rs.locked, false);
+    await rejects(rs.pipeTo({}), cleanTypeError);
+    strictEqual(rs.locked, false);
+  },
+};
+
+// Options are read before the locked checks, so an option getter that
+// locks the destination fails the pipe without locking the source.
+export const optionGetterLocksDestination = {
+  async test() {
+    const rs = new ReadableStream({});
+    const t = new TransformStream();
+    throws(
+      () =>
+        rs.pipeThrough(t, {
+          get preventAbort() {
+            t.writable.getWriter();
+            return false;
+          },
+        }),
+      cleanTypeError
+    );
+    strictEqual(rs.locked, false);
+    const ws = new WritableStream({});
+    await rejects(
+      rs.pipeTo(ws, {
+        get preventAbort() {
+          ws.getWriter();
+          return false;
+        },
+      }),
+      cleanTypeError
+    );
+    strictEqual(rs.locked, false);
+  },
+};
+
+// pipeThrough checks the writable's lock internally, not through its
+// user-visible `locked`.
+export const shadowedWritableLocked = {
+  test() {
+    const rs = new ReadableStream({});
+    const t = new TransformStream();
+    t.writable.getWriter();
+    Object.defineProperty(t.writable, 'locked', { value: false });
+    throws(() => rs.pipeThrough(t), cleanTypeError);
+    strictEqual(rs.locked, false);
+  },
+};

src/tests/streams/piping/main.js

@@ -48,6 +48,9 @@ export {
   invalidSignalRejected,
   brandChecks,
   pipeThroughLockedEndpoints,
+  badDestinationLeavesSourceUnlocked,
+  optionGetterLocksDestination,
+  shadowedWritableLocked,
 } from 'api-surface';
 
 export {