Release the endpoints of a native TypeScript streams pipe

src/per_isolate/webstreams/AGENTS.md

@@ -120,8 +120,13 @@ native-backed tee, a native-backed stream whose source is extracted
 bridge's `detach()`: `new Request(request)`, socket upgrades), and a
 detached queued stream other than the source's own stream (a tee
 branch, or a copy detached again). The legacy C++ streams leave the
-same streams locked, disturbed and closed. The Node.js interop
-closed-promise settles then, and the interop error hook, which acts only
+same streams locked, disturbed and closed. The exception to the lock is
+a native-to-native `pipeTo()`: once the pipe settles, its source is
+unlocked (still closed), and its destination, whose native sink the pipe
+also extracted, is closed or errored with the pipe's error and unlocked
+(`finishNativePipe` in readable.ts), as a pipe releases its endpoints.
+The Node.js interop closed-promise settles then, and the interop error
+hook, which acts only
 on a readable stream, cannot reach the moved source. A queued source's
 own stream, once detached, is the exception: it stays the controller's
 stream, closing and erroring with the source, and the hook on it errors

src/per_isolate/webstreams/readable.ts

@@ -2814,6 +2814,52 @@ function convertAndCheckPipe<R>(
   return converted;
 }
 
+// Finishes a native+native pipe once the sink's pipeFrom hook, which ran it
+// at the C++ layer, has settled. Both endpoints were extracted, so neither
+// stream's own algorithms took part in the pipe. As pipeToInternal's
+// shutdown and finalize would, the destination is closed (the pipe
+// succeeded) or aborted with the pipe's error, through the writer
+// extraction locked it with; then both endpoints are unlocked. The
+// extracted sink's hooks no longer reach the C++ sink, which the pipe has
+// already ended or aborted. The source has been closed since its
+// extraction (see HANDED-OFF STREAMS in webstreams/AGENTS.md) and stays
+// closed, whatever the outcome.
+function finishNativePipe<R>(
+  piped: Promise<void>,
+  reader: object,
+  destination: WritableStreamType<R>,
+  writer: WritableStreamDefaultWriterType<R>
+): Promise<void> {
+  const finalize = (): void => {
+    writableInternals.writerRelease(writer);
+    readableStreamReaderGenericRelease(reader);
+  };
+  return PromisePrototypeThen(
+    piped,
+    () =>
+      PromisePrototypeThen(
+        writableInternals.writerCloseWithErrorPropagation(writer),
+        finalize,
+        (error: unknown) => {
+          finalize();
+          throw error;
+        }
+      ),
+    (error: unknown) =>
+      PromisePrototypeThen(
+        writableInternals.writableStreamAbort(destination, error),
+        () => {
+          finalize();
+          throw error;
+        },
+        (abortError: unknown) => {
+          finalize();
+          throw abortError;
+        }
+      )
+  ) as Promise<void>;
+}
+
 // 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
@@ -3549,8 +3595,10 @@ function forkQueuedCursors<R>(
 // Marks a stream whose data has moved to another stream (tee, detach,
 // extraction; see HANDED-OFF STREAMS in webstreams/AGENTS.md). It is
 // disturbed, as the legacy C++ streams leave it, so a body that was teed
-// reports bodyUsed and cannot become another body. It stays locked for good,
-// by an internal reader that is never exposed and so never released.
+// reports bodyUsed and cannot become another body. It is locked by an
+// internal reader that is never exposed: for good, except after a
+// native+native pipe, which releases it once the pipe settles
+// (finishNativePipe), as a pipe releases its reader.
 function markHandedOff<R>(stream: ReadableStream<R>): void {
   setReadableStreamDisturbed(stream);
   acquireReadableStreamDefaultReader(stream);
@@ -3862,7 +3910,17 @@ class ReadableStream<R> {
               'Native sink is missing the required pipeFrom hook'
             );
           }
-          return uncurryThis(pipeFrom)(nativeSink, nativeSource, converted);
+          // Extraction locked both endpoints with internal handles.
+          const reader = source.#reader as object;
+          const writer = writableInternals.getWriter(
+            destination
+          ) as WritableStreamDefaultWriterType<R>;
+          return finishNativePipe(
+            uncurryThis(pipeFrom)(nativeSink, nativeSource, converted),
+            reader,
+            destination,
+            writer
+          );
         }
         return readableStreamPipeThroughTo(source, destination, converted);
       } catch (e) {

src/per_isolate/webstreams/writable.ts

@@ -164,6 +164,9 @@ let setWritableStreamWriter: <W>(
   stream: WritableStream<W>,
   writer: WritableStreamDefaultWriter<W> | undefined
 ) => void;
+let getWritableStreamWriter: <W>(
+  stream: WritableStream<W>
+) => WritableStreamDefaultWriter<W> | undefined;
 let writableStreamAbort: StreamOp<[reason?: unknown], Promise<void>>;
 let writableStreamCloseInternal: StreamOp<[], Promise<void>>;
 let writableStreamAddWriteRequest: StreamOp<[], Promise<void>>;
@@ -389,6 +392,7 @@ class WritableStream<W = unknown> {
     setWritableStreamWriter = (stream, writer) => {
       stream.#writer = writer;
     };
+    getWritableStreamWriter = (stream) => stream.#writer;
     getWritableStreamController = (stream) => stream.#controller;
 
     // JS-to-C++ extraction for native-backed writable streams.
@@ -405,8 +409,9 @@ class WritableStream<W = unknown> {
       if (sink === undefined) {
         throw new TypeError('This stream is not backed by a native sink');
       }
-      // Permanently lock via an internal writer (never exposed, never
-      // released) — same pattern as readable-side extraction.
+      // Lock via an internal writer, never exposed. The native+native pipe,
+      // which extracts the sink, closes or aborts the stream through it and
+      // releases it once the pipe settles (readableStreamPipeTo).
       new WritableStreamDefaultWriter<W>(this);
       return sink;
     };
@@ -1798,6 +1803,9 @@ module.exports = {
     writerCloseWithErrorPropagation,
     writerRelease: <W>(writer: WritableStreamDefaultWriter<W>) =>
       writerReleaseInternal(writer),
+    // The writer holding the stream's lock, if any.
+    getWriter: <W>(stream: WritableStream<W>) =>
+      getWritableStreamWriter(stream),
     writableStreamAbort: <W>(stream: WritableStream<W>, reason: unknown) =>
       writableStreamAbort(stream, reason),
     getState: <W>(stream: WritableStream<W>) => getWritableStreamState(stream),

src/tests/streams/sockets/AGENTS.md

@@ -28,7 +28,7 @@ service allowing `private`. The suite's `.wd-test` files take no
 
 ## Coverage
 
-Two implementation divergences are pinned:
+Three implementation divergences are pinned:
 
 1. `cancelReadableSettlesSocket`: canceling a pending read rejects with a
    re-created `Error` carrying the cancel reason under C++, while
@@ -41,6 +41,13 @@ Two implementation divergences are pinned:
    (`awaitIoLegacy`, then `drain()`). Neither order is specified
    (`startTls` and flush are workerd extensions); the TypeScript one
    reports the failed writes before the upgrade.
+3. `nativePipeReleasesEndpoints`: once a native-to-native pipe settles,
+   TypeScript's source reads done whatever the outcome, as it has been
+   closed since its native half was extracted at the start; on an abort
+   that is the spec's cancel. C++ errors the source with the pipe's error
+   when the pipe is aborted or the source fails, and with an
+   already-aborted signal touches neither endpoint. The spec leaves a
+   failed source errored, as C++ does.
 
 | Test | Shape |
 | --- | --- |
@@ -56,6 +63,7 @@ Two implementation divergences are pinned:
 | `pipeSocketToSocket` | greet socket's readable piped into the echo socket's writable |
 | `pipeSocketToSocketClosesSource` | the same native-to-native pipe leaves the source locked and (TS, via the interop closed-promise) closed at the start |
 | `pipeBehindUnawaitedWrite` | header write not awaited, writer released, Response body piped in: echo is header then body; both endpoints unlocked after the pipe |
+| `nativePipeReleasesEndpoints` | a Response body, an idle socket's readable (aborted mid-pipe), a Response body with an already-aborted signal, and a service-binding body that fails, each piped into a socket writable: both endpoints unlocked after the pipe; the writable closed or errored with the pipe's error (C++: untouched by the already-aborted signal); the source's state diverges (3) |
 | `pipeBehindWriteBeforeStart` | the same within connect()'s turn, with the header still queued before the writable starts |
 | `pipeOptionGetterLocksNativeEndpoint` | an option getter that takes a reader on a Response body, or a writer on the socket writable, fails the native-to-native pipe with a TypeError and leaves both endpoints as they were: the reader still reads the whole body, the body piped toward the locked writable still reads in full, and the writer's write is echoed |
 | `cancelReadableSettlesSocket` | reader.cancel settles a pending peer read (C++ rejects; TS resolves done), then socket.close()/closed settle |

src/tests/streams/sockets/main.js

@@ -19,6 +19,7 @@ export {
   pipeSocketToSocketClosesSource,
   pipeBehindUnawaitedWrite,
   pipeBehindWriteBeforeStart,
+  nativePipeReleasesEndpoints,
   pipeOptionGetterLocksNativeEndpoint,
   cancelReadableSettlesSocket,
   detachRejectsPendingWrites,

src/tests/streams/sockets/socket-streams.js

@@ -15,6 +15,10 @@
 // closed promise resolves; C++ resolves closed first and rejects the writes
 // an event-loop turn later, once the canceled write's I/O reports back
 // (detachRejectsPendingWrites).
+//
+// The source's state once a native-to-native pipe settles diverges:
+// TypeScript's reads done, C++ errors it when the pipe is aborted or the
+// source fails (nativePipeReleasesEndpoints).
 
 import { connect } from 'cloudflare:sockets';
 import { strictEqual, ok, rejects, deepStrictEqual } from 'node:assert';
@@ -342,6 +346,115 @@ export const pipeBehindWriteBeforeStart = {
   },
 };
 
+async function settled(promise) {
+  return promise.then(
+    () => 'resolved',
+    (e) => `rejected ${e.message}`
+  );
+}
+
+// What a read on the unlocked source settles with: 'done', 'value', or
+// the rejection.
+async function sourceOutcome(source) {
+  const reader = source.getReader();
+  const outcome = await reader.read().then(
+    ({ done }) => (done ? 'done' : 'value'),
+    (e) => `rejected ${e.message}`
+  );
+  reader.releaseLock();
+  return outcome;
+}
+
+// How the unlocked writable's writer.closed settles, 'pending' if it has
+// not within 50 ms.
+async function writableOutcome(writable) {
+  const writer = writable.getWriter();
+  const outcome = await Promise.race([
+    settled(writer.closed),
+    scheduler.wait(50).then(() => 'pending'),
+  ]);
+  writer.releaseLock();
+  return outcome;
+}
+
+async function pipeOutcome(source, writable, options) {
+  const pipe = await settled(source.pipeTo(writable, options));
+  strictEqual(source.locked, false, 'source locked');
+  strictEqual(writable.locked, false, 'writable locked');
+  return {
+    pipe,
+    source: await sourceOutcome(source),
+    writable: await writableOutcome(writable),
+  };
+}
+
+// A native-to-native pipe into a socket writable unlocks both endpoints
+// once it settles, however it settles.
+//
+// DIVERGENCE: the source's final state. TypeScript extracts the source's
+// native half at the start, closing it (pipeSocketToSocketClosesSource),
+// so it reads done whatever the outcome; this follows the spec when the
+// pipe is aborted, which cancels the source. C++ errors the source with
+// the pipe's error when it is aborted or the source fails, and with an
+// already-aborted signal touches neither endpoint.
+export const nativePipeReleasesEndpoints = {
+  async test(ctrl, env) {
+    {
+      const socket = connect(echoAddress(env));
+      const echoed = drainToBytes(socket.readable);
+      deepStrictEqual(
+        await pipeOutcome(new Response('BODY').body, socket.writable),
+        { pipe: 'resolved', source: 'done', writable: 'resolved' }
+      );
+      strictEqual(dec.decode(await echoed), 'BODY');
+      await socket.close();
+    }
+    {
+      const idle = connect(echoAddress(env));
+      const socket = connect(echoAddress(env));
+      const ac = new AbortController();
+      const outcome = pipeOutcome(idle.readable, socket.writable, {
+        signal: ac.signal,
+      });
+      await scheduler.wait(20);
+      ac.abort(new Error('stop'));
+      deepStrictEqual(await outcome, {
+        pipe: 'rejected stop',
+        source: usingTsImpl ? 'done' : 'rejected stop',
+        writable: 'rejected stop',
+      });
+      await Promise.all([idle.close(), socket.close()]);
+    }
+    {
+      const socket = connect(echoAddress(env));
+      const signal = AbortSignal.abort(new Error('early'));
+      deepStrictEqual(
+        await pipeOutcome(new Response('BODY').body, socket.writable, {
+          signal,
+        }),
+        usingTsImpl
+          ? {
+              pipe: 'rejected early',
+              source: 'done',
+              writable: 'rejected early',
+            }
+          : { pipe: 'rejected early', source: 'value', writable: 'pending' }
+      );
+      await socket.close();
+    }
+    {
+      const socket = connect(echoAddress(env));
+      const body = (await env.SELF.fetch('http://self/failing')).body;
+      deepStrictEqual(await pipeOutcome(body, socket.writable), {
+        pipe: 'rejected the body failed',
+        source: usingTsImpl ? 'done' : 'rejected the body failed',
+        writable: 'rejected the body failed',
+      });
+      await socket.close();
+    }
+  },
+};
+
 // A pipe reads its options before it checks the locks, so an option
 // getter that locks an endpoint of a native-to-native pipe fails the pipe
 // before it takes either endpoint's native half. That matters most for
@@ -550,6 +663,25 @@ export const largeEchoVolume = {
 // and socket.close() aborts it. Whatever a later writer's closed promise then
 // reports must be a real error, never undefined.
 export default {
+  // /failing: a body that delivers one chunk and then errors, for the
+  // native pipes whose source fails.
+  async fetch(request) {
+    let sent = false;
+    return new Response(
+      new ReadableStream({
+        async pull(c) {
+          if (!sent) {
+            sent = true;
+            c.enqueue(enc.encode('PART|'));
+            return;
+          }
+          await scheduler.wait(10);
+          c.error(new Error('the body failed'));
+        },
+      })
+    );
+  },
+
   async connect(socket, env) {
     const { localAddress } = await socket.opened;
     if (localAddress.startsWith('echo:')) {

src/workerd/api/tests/js-rpc-streams-crossflag-test.js

@@ -153,6 +153,23 @@ export default {
     if (url.hostname === 'body') {
       return new Response(url.searchParams.get('text'));
     }
+    if (url.hostname === 'failing') {
+      // One chunk, then an error.
+      let sent = false;
+      return new Response(
+        new ReadableStream({
+          async pull(c) {
+            if (!sent) {
+              sent = true;
+              c.enqueue(enc.encode('part'));
+              return;
+            }
+            await scheduler.wait(10);
+            c.error(new Error('the body failed'));
+          },
+        })
+      );
+    }
     return Response.json({
       length: request.headers.get('Content-Length'),
       text: await request.text(),
@@ -263,6 +280,38 @@ export const stringChunksCrossAsUtf8 = {
   },
 };
 
+// A native body piped into a writable received over RPC: once the pipe settles, both endpoints
+// are unlocked, whether it resolved (with or without a write made before it) or rejected on a
+// source error.
+export const nativePipeIntoRpcWritableReleasesLocks = {
+  async test(controller, env) {
+    if (!(await isCrossImplementationRunner(env))) return;
+    for (const header of [undefined, 'header|']) {
+      const { readable, writable } =
+        await env.PEER.makeConstructorBackedEchoPair();
+      const text = new Response(readable).text();
+      if (header !== undefined) {
+        const writer = writable.getWriter();
+        await writer.write(enc.encode(header));
+        writer.releaseLock();
+      }
+      const body = new Response(enc.encode('body')).body;
+      await body.pipeTo(writable);
+      assert.strictEqual(await text, `${header ?? ''}body`);
+      assert.strictEqual(body.locked, false, header);
+      assert.strictEqual(writable.locked, false, header);
+    }
+
+    const { readable, writable } =
+      await env.PEER.makeConstructorBackedEchoPair();
+    readable.pipeTo(new WritableStream()).catch(() => {});
+    const body = (await env.HTTP.fetch('http://failing/')).body;
+    await assert.rejects(body.pipeTo(writable), { message: 'the body failed' });
+    assert.strictEqual(body.locked, false);
+    assert.strictEqual(writable.locked, false);
+  },
+};
+
 export const readableLength = {
   async test(controller, env) {
     if (!(await isCrossImplementationRunner(env))) return;