From 2eb5b37a5ab0295fbb4675c8fc5c9bbb8a3827da Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 28 Oct 2026 00:37:12 +0000 Subject: [PATCH] Release the endpoints of a native TypeScript streams pipe Under typescript_implemented_streams, pipeTo() from a native-backed readable (a fetch() or service-binding body, a buffer-backed body, Blob.stream(), a socket's readable) into a native-backed writable (a connect() socket's writable, a writable received over JS RPC) runs at the C++ layer through the sink's pipeFrom hook. Extracting the two native halves locks both streams with internal handles, which the fast path never released: once the pipe settled, both endpoints stayed locked for good, so the writable could not be closed, aborted or written again, and the source could not be read or cancelled. The JS pipe path, and the C++ implementation, release both. Once pipeFrom settles, the fast path now finishes the pipe as the JS pipe's shutdown and finalize would: it closes the destination through its internal writer when the pipe succeeded, or aborts it with the pipe's error, and then releases the writer and the source's reader. The extracted sink's hooks no longer reach the C++ sink, so neither reaches the connection. The source has been closed since its extraction and stays closed, so it reads done whatever the outcome. On an abort that is the spec's cancel. C++ instead errors the source when the pipe is aborted or the source fails, and with an already-aborted signal touches neither endpoint. The sockets suite pins these as a divergence. Tests: nativePipeReleasesEndpoints in the sockets suite (success, mid-pipe abort, an already-aborted signal, a failing service-binding body), and the cross-implementation RPC test for a writable received over RPC, with and without a write made before the pipe, and with a failing source. typescript_implemented_streams is still experimental, with no enable date, so the change needs no compatibility flag. Fixes: https://github.com/cloudflare/workerd/issues/8049 --- src/per_isolate/webstreams/AGENTS.md | 9 +- src/per_isolate/webstreams/readable.ts | 64 ++++++++- src/per_isolate/webstreams/writable.ts | 12 +- src/tests/streams/sockets/AGENTS.md | 10 +- src/tests/streams/sockets/main.js | 1 + src/tests/streams/sockets/socket-streams.js | 132 ++++++++++++++++++ .../tests/js-rpc-streams-crossflag-test.js | 49 +++++++ 7 files changed, 269 insertions(+), 8 deletions(-) diff --git a/src/per_isolate/webstreams/AGENTS.md b/src/per_isolate/webstreams/AGENTS.md index aa0fe26f4c4..94941493391 100644 --- a/src/per_isolate/webstreams/AGENTS.md +++ b/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 diff --git a/src/per_isolate/webstreams/readable.ts b/src/per_isolate/webstreams/readable.ts index 46d4fb9ea27..d62aa82fb31 100644 --- a/src/per_isolate/webstreams/readable.ts +++ b/src/per_isolate/webstreams/readable.ts @@ -2814,6 +2814,52 @@ function convertAndCheckPipe( 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( + piped: Promise, + reader: object, + destination: WritableStreamType, + writer: WritableStreamDefaultWriterType +): Promise { + 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; +} + // 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( // 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(stream: ReadableStream): void { setReadableStreamDisturbed(stream); acquireReadableStreamDefaultReader(stream); @@ -3862,7 +3910,17 @@ class ReadableStream { '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; + return finishNativePipe( + uncurryThis(pipeFrom)(nativeSink, nativeSource, converted), + reader, + destination, + writer + ); } return readableStreamPipeThroughTo(source, destination, converted); } catch (e) { diff --git a/src/per_isolate/webstreams/writable.ts b/src/per_isolate/webstreams/writable.ts index 9500139ddb4..b668b2f8d93 100644 --- a/src/per_isolate/webstreams/writable.ts +++ b/src/per_isolate/webstreams/writable.ts @@ -164,6 +164,9 @@ let setWritableStreamWriter: ( stream: WritableStream, writer: WritableStreamDefaultWriter | undefined ) => void; +let getWritableStreamWriter: ( + stream: WritableStream +) => WritableStreamDefaultWriter | undefined; let writableStreamAbort: StreamOp<[reason?: unknown], Promise>; let writableStreamCloseInternal: StreamOp<[], Promise>; let writableStreamAddWriteRequest: StreamOp<[], Promise>; @@ -389,6 +392,7 @@ class WritableStream { 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 { 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(this); return sink; }; @@ -1798,6 +1803,9 @@ module.exports = { writerCloseWithErrorPropagation, writerRelease: (writer: WritableStreamDefaultWriter) => writerReleaseInternal(writer), + // The writer holding the stream's lock, if any. + getWriter: (stream: WritableStream) => + getWritableStreamWriter(stream), writableStreamAbort: (stream: WritableStream, reason: unknown) => writableStreamAbort(stream, reason), getState: (stream: WritableStream) => getWritableStreamState(stream), diff --git a/src/tests/streams/sockets/AGENTS.md b/src/tests/streams/sockets/AGENTS.md index 116000beee4..fb48f8a3bcc 100644 --- a/src/tests/streams/sockets/AGENTS.md +++ b/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 | diff --git a/src/tests/streams/sockets/main.js b/src/tests/streams/sockets/main.js index d5a97ef8044..8599d2469ca 100644 --- a/src/tests/streams/sockets/main.js +++ b/src/tests/streams/sockets/main.js @@ -19,6 +19,7 @@ export { pipeSocketToSocketClosesSource, pipeBehindUnawaitedWrite, pipeBehindWriteBeforeStart, + nativePipeReleasesEndpoints, pipeOptionGetterLocksNativeEndpoint, cancelReadableSettlesSocket, detachRejectsPendingWrites, diff --git a/src/tests/streams/sockets/socket-streams.js b/src/tests/streams/sockets/socket-streams.js index 6f5bcca2ffa..d018c78debc 100644 --- a/src/tests/streams/sockets/socket-streams.js +++ b/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:')) { diff --git a/src/workerd/api/tests/js-rpc-streams-crossflag-test.js b/src/workerd/api/tests/js-rpc-streams-crossflag-test.js index 2c9d177f8d8..13f7de009c3 100644 --- a/src/workerd/api/tests/js-rpc-streams-crossflag-test.js +++ b/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;