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;