From 72bef1e72da7fa09528197d8b129b61f65f41a41 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Tue, 27 Oct 2026 14:01:46 +0100 Subject: [PATCH] stream: keep async context of fromWeb writes Writable.fromWeb() and Duplex.fromWeb() dispatch writes to the WritableStream directly and complete them through a shared write tracker. A write issued before the WritableStream had started was queued and processed from its start step, so the sink ran in the async context of the code that created the WritableStream, and the tracker invoked the Node.js write callback synchronously from whatever context completed the sink write, so the writes and final() it released ran there too. The promise-based adapter waited for writer.ready and received the result through a reaction, both in the context of the Node.js write. Dispatch writes issued before the start from a writer.ready reaction, failing them if the stream was closed or errored meanwhile, and deliver the write callback in the context of the Node.js write, from a reaction when it completes in another context. Refs: https://github.com/nodejs/node/issues/66906 Refs: https://github.com/nodejs/node/pull/65548 Signed-off-by: Matteo Collina --- lib/internal/webstreams/adapters.js | 68 +++++++++++- lib/internal/webstreams/writablestream.js | 19 ++++ ...hatwg-webstreams-adapters-async-context.js | 105 ++++++++++++++++++ 3 files changed, 191 insertions(+), 1 deletion(-) create mode 100644 test/parallel/test-whatwg-webstreams-adapters-async-context.js diff --git a/lib/internal/webstreams/adapters.js b/lib/internal/webstreams/adapters.js index a41b668f014c..fc5ada12be3c 100644 --- a/lib/internal/webstreams/adapters.js +++ b/lib/internal/webstreams/adapters.js @@ -32,6 +32,8 @@ const { const { WritableStream, isWritableStream, + writableStreamDefaultWriterIsStarted, + writableStreamDefaultWriterWriteFailure, writableStreamDefaultWriterWriteWithRequest, } = require('internal/webstreams/writablestream'); @@ -62,6 +64,11 @@ const { kResolvedPromise, } = require('internal/webstreams/util'); +const { + getContinuationPreservedEmbedderData: currentContextFrame, + setContinuationPreservedEmbedderData: setContextFrame, +} = internalBinding('async_context_frame'); + const { isAnyArrayBuffer, } = require('internal/util/types'); @@ -157,7 +164,9 @@ const noop = () => {}; // batch is in flight at a time, so one callback slot is enough. // `promise` is non-undefined so writable-stream queue entries are // discriminated from kNilRequest by `promise === undefined`; -// `failed`/`failure` latch a write that could not proceed. +// `failed`/`failure` latch a write that could not proceed. The callback +// runs in the async context of the Node.js write that set it, as the +// reaction to writer.write() did (see settle()). function createWriteTracker(getDestination) { return { promise: null, @@ -165,6 +174,7 @@ function createWriteTracker(getDestination) { failed: false, failure: undefined, callback: undefined, + contextFrame: undefined, error: undefined, hasError: false, dispatching: false, @@ -187,6 +197,22 @@ function createWriteTracker(getDestination) { } }, settle() { + const contextFrame = this.contextFrame; + this.contextFrame = undefined; + const prior = currentContextFrame(); + if (prior === contextFrame) { + this.settleInContext(); + return; + } + // Completed from another async context (e.g. a write queued while + // the destination was starting): deliver from a reaction registered + // in the context of the write, once the destination has finished + // processing it. + setContextFrame(contextFrame); + PromisePrototypeThen(kResolvedPromise, () => this.settleInContext()); + setContextFrame(prior); + }, + settleInContext() { const callback = this.callback; this.callback = undefined; let error; @@ -224,7 +250,37 @@ function createWriteTracker(getDestination) { }; } +// A write issued before the destination has started would be queued and +// processed from the start step, in the async context of the code that +// created the WritableStream. Dispatch it from a writer.ready reaction +// instead, as the promise-based adapter did, so the sink runs in the +// context of the Node.js write. The stream may have been closed or +// errored (e.g. by destroy()) by then. +function deferDispatch(writer, tracker, dispatchNow) { + const dispatch = () => { + const failure = writableStreamDefaultWriterWriteFailure(writer); + if (failure !== undefined) { + tracker.failed = true; + tracker.failure = failure; + tracker.settle(); + return; + } + dispatchNow(); + }; + PromisePrototypeThen(writer.ready, dispatch, dispatch); +} + function dispatchWrite(writer, chunk, tracker) { + tracker.contextFrame = currentContextFrame(); + if (!writableStreamDefaultWriterIsStarted(writer)) { + deferDispatch(writer, tracker, + () => dispatchWriteNow(writer, chunk, tracker)); + return; + } + dispatchWriteNow(writer, chunk, tracker); +} + +function dispatchWriteNow(writer, chunk, tracker) { tracker.dispatching = true; writableStreamDefaultWriterWriteWithRequest(writer, chunk, tracker); tracker.dispatching = false; @@ -233,6 +289,16 @@ function dispatchWrite(writer, chunk, tracker) { } function dispatchWritev(writer, chunks, tracker) { + tracker.contextFrame = currentContextFrame(); + if (!writableStreamDefaultWriterIsStarted(writer)) { + deferDispatch(writer, tracker, + () => dispatchWritevNow(writer, chunks, tracker)); + return; + } + dispatchWritevNow(writer, chunks, tracker); +} + +function dispatchWritevNow(writer, chunks, tracker) { tracker.dispatching = true; for (let i = 0; i < chunks.length; i++) writableStreamDefaultWriterWriteWithRequest(writer, chunks[i].chunk, tracker); diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index 31f2a1f875dc..36e26c71db65 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -1001,6 +1001,23 @@ function writableStreamDefaultWriterWrite(writer, chunk) { return promise; } +function writableStreamDefaultWriterIsStarted(writer) { + const stream = writer[kState].stream; + return stream === undefined || + stream[kState].controller[kState].started; +} + +// Returns the error a write through the writer would fail with, or +// undefined if the stream still accepts writes. +function writableStreamDefaultWriterWriteFailure(writer) { + const streamState = writer[kState].stream[kState]; + const { state } = streamState; + if (state === 'errored' || state === 'erroring') + return streamState.storedError; + if (streamState.closeQueuedOrInFlight || state === 'closed') + return new ERR_INVALID_STATE.TypeError('WritableStream is closed'); +} + // Variant of writableStreamDefaultWriterWrite for pipeTo: the caller // provides a shared request object instead of a per-write promise record. // `pending` is incremented before the controller write, which can settle @@ -1439,6 +1456,8 @@ module.exports = { writableStreamCloseQueuedOrInFlight, writableStreamAddWriteRequest, writableStreamDefaultWriterWrite, + writableStreamDefaultWriterIsStarted, + writableStreamDefaultWriterWriteFailure, writableStreamDefaultWriterWriteWithRequest, writableStreamDefaultWriterRelease, writableStreamDefaultWriterGetDesiredSize, diff --git a/test/parallel/test-whatwg-webstreams-adapters-async-context.js b/test/parallel/test-whatwg-webstreams-adapters-async-context.js new file mode 100644 index 000000000000..94240018e45a --- /dev/null +++ b/test/parallel/test-whatwg-webstreams-adapters-async-context.js @@ -0,0 +1,105 @@ +'use strict'; + +// Writes through Writable.fromWeb() and Duplex.fromWeb() must reach the +// WritableStream sink in the async context of the Node.js write, also when +// they are issued before the WritableStream has started. + +const common = require('../common'); +const assert = require('assert'); +const { AsyncLocalStorage } = require('async_hooks'); +const { Duplex, Writable } = require('stream'); +const { setImmediate: tick } = require('timers/promises'); + +const als = new AsyncLocalStorage(); + +function store() { + return als.getStore() ?? 'none'; +} + +function createSink(seen, start) { + return new WritableStream({ + start, + write(chunk) { seen.push(`write-${chunk}:${store()}`); }, + close() { seen.push(`close:${store()}`); }, + }); +} + +const tests = []; +function test(name, expected, fn) { + tests.push({ name, expected, fn }); +} + +test('Writable.fromWeb() written before start', [ + 'write-a:writer', 'write-b:writer', 'close:writer', +], async (seen) => { + const ws = als.run('creator', () => createSink(seen)); + const writable = Writable.fromWeb(ws, { decodeStrings: false }); + await als.run('writer', () => new Promise((resolve) => { + writable.write('a'); + writable.write('b'); + writable.end(resolve); + })); +}); + +test('Writable.fromWeb() writev before start', [ + 'write-a:writer', 'write-b:writer', 'close:writer', +], async (seen) => { + const ws = als.run('creator', () => createSink(seen)); + const writable = Writable.fromWeb(ws, { decodeStrings: false }); + await als.run('writer', () => new Promise((resolve) => { + writable.cork(); + writable.write('a'); + writable.write('b'); + writable.uncork(); + writable.end(resolve); + })); +}); + +test('Writable.fromWeb() written during an async start', [ + // The WritableStream processes the chunk queued while it was starting + // from its start step. The Node.js write callback still runs in the + // writer's context, so the writes it releases do too. + 'write-a:creator', 'write-b:writer', 'close:writer', +], async (seen) => { + const ws = als.run('creator', () => createSink(seen, () => tick())); + const writable = Writable.fromWeb(ws, { decodeStrings: false }); + await als.run('writer', () => new Promise((resolve) => { + writable.write('a'); + writable.write('b'); + writable.end(resolve); + })); +}); + +test('Writable.fromWeb() written after start', [ + 'write-a:w1', 'write-b:w2', 'close:ender', +], async (seen) => { + const ws = als.run('creator', () => createSink(seen)); + const writable = Writable.fromWeb(ws, { decodeStrings: false }); + await tick(); + await als.run('w1', () => new Promise((resolve) => writable.write('a', resolve))); + await als.run('w2', () => new Promise((resolve) => writable.write('b', resolve))); + await als.run('ender', () => new Promise((resolve) => writable.end(resolve))); +}); + +test('Duplex.fromWeb() written before start', [ + 'write-a:writer', 'close:writer', +], async (seen) => { + const pair = als.run('creator', () => ({ + readable: new ReadableStream({ start(c) { c.close(); } }), + writable: createSink(seen), + })); + const duplex = Duplex.fromWeb(pair, { decodeStrings: false }); + duplex.resume(); + await als.run('writer', () => new Promise((resolve) => { + duplex.write('a'); + duplex.end(resolve); + })); +}); + +(async () => { + for (const { name, expected, fn } of tests) { + const seen = []; + await fn(seen); + assert.deepStrictEqual(seen, expected, name); + } +})().then(common.mustCall());