stream: keep async context of fromWeb writes

@@ -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,14 +164,17 @@ 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,     pending: 0,     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);

@@ -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,

@@ -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());