child_process: deliver each stdio event after the nextTicks and promise jobs of the one before

src/js/builtins/ProcessObjectInternals.ts

@@ -348,12 +348,13 @@ export function initializeNextTickQueue(process: typeof globalThis.process, next
   var drainMicrotasks = drainMicrotasksFn;
 
   const { validateFunction } = require("internal/validators");
+  const asyncHooksTick = require("internal/async_hooks_tick");
 
   var setup;
   setup = () => {
     const { FixedQueue } = require("internal/fixed_queue");
     queue = new FixedQueue();
-    tickInitHooks = require("internal/async_hooks_tick").tickInitHooks;
+    tickInitHooks = asyncHooksTick.tickInitHooks;
 
     function processTicksAndRejections() {
       var tock;
@@ -393,7 +394,8 @@ export function initializeNextTickQueue(process: typeof globalThis.process, next
         }
 
         drainMicrotasks();
-      } while (!queue.isEmpty());
+        // With both queues empty, internal/process/after_tick_drain.ts (once something loaded it) runs one callback.
+      } while (!queue.isEmpty() || asyncHooksTick.runAfterTickDrainCallback?.());
     }
 
     $putInternalField(nextTickQueue, 0, 0);
@@ -402,6 +404,15 @@ export function initializeNextTickQueue(process: typeof globalThis.process, next
     setup = undefined;
   };
 
+  // For internal/process/after_tick_drain.ts. JSNextTickQueue::drain only calls the tick loop while field 0 is set.
+  asyncHooksTick.ensureTickLoop = () => {
+    if (setup) {
+      setup();
+      process = globalThis.process;
+    }
+    $putInternalField(nextTickQueue, 0, 1);
+  };
+
   function nextTick(cb, ...args) {
     validateFunction(cb, "callback");
     if (setup) {
@@ -420,7 +431,6 @@ export function initializeNextTickQueue(process: typeof globalThis.process, next
     if (tickInitHooks.length !== 0) {
       // node fires one TickObject init per process.nextTick() call, at
       // construction time (before the callback runs).
-      const asyncHooksTick = require("internal/async_hooks_tick");
       const asyncId = asyncHooksTick.newAsyncId();
       // Snapshot: enable()/disable() from inside a hook must not affect the
       // in-flight dispatch (node stages such mutations in tmp_array until

src/js/internal/async_hooks_tick.ts

@@ -17,4 +17,8 @@ export default {
   newAsyncId() {
     return ++nextAsyncId;
   },
+  // For internal/process/after_tick_drain.ts. Set when process.nextTick is first read, which creates the tick queue.
+  ensureTickLoop: undefined as (() => void) | undefined,
+  // Set by internal/process/after_tick_drain.ts. The tick loop calls it each time both queues are empty.
+  runAfterTickDrainCallback: undefined as (() => boolean) | undefined,
 };

src/js/internal/process/after_tick_drain.ts

@@ -0,0 +1,46 @@
+// Node empties the nextTick queue and the promise job queue between two libuv callbacks: https://github.com/nodejs/node/blob/v24.9.0/lib/internal/process/task_queues.js#L72-L109
+const asyncHooksTick = require("internal/async_hooks_tick");
+
+type AfterTickDrainCallback = {
+  callback: (arg: any) => void;
+  arg: unknown;
+  frame: import("../../node/async_hooks").Frame | undefined;
+};
+
+// A queue: entries before `head` are done. Bounded by the number of callers, which each wait for their turn.
+const entries: (AfterTickDrainCallback | undefined)[] = [];
+let head = 0;
+
+// Runs `callback` once both queues are empty, inside the drain that is running. One callback for each such point.
+function runAfterTickDrain(callback: (arg: any) => void, arg?: unknown) {
+  let ensureTickLoop = asyncHooksTick.ensureTickLoop;
+  if (ensureTickLoop === undefined) {
+    // The first read of process.nextTick creates the tick queue. Fake timers read it before they replace it.
+    void process.nextTick;
+    ensureTickLoop = asyncHooksTick.ensureTickLoop;
+    // process.nextTick was overwritten before its first read: there is no tick loop to wait for.
+    if (ensureTickLoop === undefined) return process.nextTick(callback, arg);
+  }
+  ensureTickLoop();
+  $arrayPush(entries, { callback, arg, frame: $getInternalField($asyncContext, 0) });
+}
+
+// The tick loop calls this each time both queues are empty. False: nothing waited.
+function runAfterTickDrainCallback() {
+  if (head === entries.length) return false;
+  const { callback, arg, frame } = entries[head]!;
+  entries[head++] = undefined;
+  if (head === entries.length) {
+    entries.length = 0;
+    head = 0;
+  }
+  const restore = $getInternalField($asyncContext, 0);
+  $putInternalField($asyncContext, 0, frame);
+  // No catch and no finally, as for a tick: JSNextTickQueue::drain reports the throw and calls the tick loop again.
+  callback(arg);
+  $putInternalField($asyncContext, 0, restore);
+  return true;
+}
+asyncHooksTick.runAfterTickDrainCallback = runAfterTickDrainCallback;
+
+export default { runAfterTickDrain };

src/js/internal/streams/native-readable.ts

@@ -12,6 +12,7 @@ const transferToNativeReadable = $newCppFunction(
   1,
 );
 const { errorOrDestroy } = require("internal/streams/destroy");
+const { runAfterTickDrain } = require("internal/process/after_tick_drain");
 
 const kRefCount = Symbol("refCount");
 const kCloseState = Symbol("closeState");
@@ -20,6 +21,15 @@ const kHighWaterMark = Symbol("highWaterMark");
 const kPendingRead = Symbol("pendingRead");
 const kHasResized = Symbol("hasResized");
 const kRemainingChunk = Symbol("remainingChunk");
+const kSawEof = Symbol("sawEof");
+const kReadEvents = Symbol("readEvents");
+const kReadEventsHead = Symbol("readEventsHead");
+const kReadEventQueued = Symbol("readEventQueued");
+
+// What a read gave, in the order Node would report it: see `kReadEvents`.
+const kChunkEvent = 0;
+const kEofEvent = 1;
+const kErrorEvent = 2;
 
 const MIN_BUFFER_SIZE = 512;
 let dynamicallyAdjustChunkSize = (_?) => (
@@ -49,6 +59,11 @@ interface NativeReadable extends NodeReadable {
   [kHighWaterMark]: number;
   [kHasResized]: boolean;
   [kRemainingChunk]: Buffer | undefined;
+  [kSawEof]: boolean;
+  // "libuv-handle" only: (kind, value) pairs from `kReadEventsHead` on. At most the results of one `_read`.
+  [kReadEvents]: unknown[] | undefined;
+  [kReadEventsHead]: number;
+  [kReadEventQueued]: boolean;
   debugId: number;
 }
 
@@ -65,12 +80,19 @@ interface NativePtr {
 
 let debugId = 0;
 
-function constructNativeReadable(readableStream: ReadableStream, options): NativeReadable {
+// `nodeSource`: what feeds the equivalent stream in Node. "libuv-handle" is child_process stdio, a net.Socket there.
+function constructNativeReadable(
+  readableStream: ReadableStream,
+  options,
+  nodeSource: "libuv-handle" | "webstream",
+): NativeReadable {
   $assert(typeof readableStream === "object" && readableStream instanceof ReadableStream, "Invalid readable stream");
   const bunNativePtr = (readableStream as any).$bunNativePtr;
   $assert(typeof bunNativePtr === "object", "Invalid native ptr");
 
-  const stream = new Readable(options) as NativeReadable;
+  const isHandle = nodeSource === "libuv-handle";
+  // A net.Socket emits 'close' itself, from the close callback of its handle: see `destroy`.
+  const stream = new Readable(isHandle ? { ...options, emitClose: false } : options) as NativeReadable;
   stream._read = read;
   stream._destroy = destroy;
 
@@ -86,6 +108,10 @@ function constructNativeReadable(readableStream: ReadableStream, options): Nativ
   stream[kPendingRead] = false;
   stream[kHasResized] = !dynamicallyAdjustChunkSize();
   stream[kCloseState] = [false];
+  stream[kSawEof] = false;
+  stream[kReadEvents] = isHandle ? [] : undefined;
+  stream[kReadEventsHead] = 0;
+  stream[kReadEventQueued] = false;
 
   const highWaterMark = options.highWaterMark;
   stream[kHighWaterMark] = typeof highWaterMark === "number" ? highWaterMark : 256 * 1024;
@@ -142,7 +168,12 @@ function read(this: NativeReadable, maxToRead: number) {
   // not paused from a previous `push()===false` (readStart, like net.Socket).
   // Runs even when a pull promise is outstanding so that promise can resolve.
   if (ptr) ptr.setFlowing?.(true);
-  if (this[kPendingRead]) {
+  if (this[kPendingRead] || this[kSawEof]) {
+    return;
+  }
+  const events = this[kReadEvents];
+  // The events of the last read come first, and push() has not said yet if it wants more.
+  if (events !== undefined && events.length !== 0) {
     return;
   }
   if (!ptr) {
@@ -158,17 +189,25 @@ function read(this: NativeReadable, maxToRead: number) {
       this[kHighWaterMark] = Math.min(this[kHighWaterMark], result);
     }
     if ($isTypedArrayView(result) && result.byteLength > 0) {
-      pushAndCheck(this, result);
+      deliverChunk(this, result);
     }
     const drainResult = ptr.drain();
     this[kConstructed] = true;
     $debug(`[${this.debugId}] drain result: ${drainResult?.byteLength ?? "null"}`);
     if ((drainResult?.byteLength ?? 0) > 0) {
-      pushAndCheck(this, drainResult);
+      deliverChunk(this, drainResult);
     }
   }
   const chunk = getRemainingChunk(this, maxToRead);
-  var result = ptr.pull(chunk, this[kCloseState]);
+  var result;
+  try {
+    result = ptr.pull(chunk, this[kCloseState]);
+  } catch (error) {
+    if (events === undefined) throw error;
+    // The chunks that this read already got come before the error.
+    queueReadEvent(this, events, kErrorEvent, error);
+    return;
+  }
   $assert(result !== undefined);
   $debug(
     `[${this.debugId}] pull ${chunk?.byteLength} bytes, result: ${$isPromise(result) ? "<pending>" : $isTypedArrayView(result) ? `<${result.byteLength} bytes>` : result}, closeState: ${this[kCloseState][0]}`,
@@ -184,7 +223,8 @@ function read(this: NativeReadable, maxToRead: number) {
         this[kRemainingChunk] = handleResult(this, result, chunk, this[kCloseState][0]);
       },
       reason => {
-        errorOrDestroy(this, reason);
+        if (events === undefined) errorOrDestroy(this, reason);
+        else queueReadEvent(this, events, kErrorEvent, reason);
       },
     );
   } else {
@@ -201,7 +241,7 @@ function handleResult(stream: NativeReadable, result: any, chunk: Buffer | undef
     return handleNumberResult(stream, result, chunk, isClosed);
   } else if (typeof result === "boolean") {
     $debug(`[${stream.debugId}] handleResult(${result})`, chunk, isClosed);
-    process.nextTick(pushEof, stream);
+    deliverEof(stream);
     return (chunk?.byteLength ?? 0) > 0 ? chunk : undefined;
   } else if ($isTypedArrayView(result)) {
     if (result.byteLength >= stream[kHighWaterMark] && !stream[kHasResized] && !isClosed) {
@@ -213,7 +253,7 @@ function handleResult(stream: NativeReadable, result: any, chunk: Buffer | undef
   }
 }
 
-// EOF is pushed a tick after the last chunk. After a destroy() in between, Node emits 'close' without 'end'.
+// After a destroy() before the EOF gets its turn, Node emits 'close' without 'end'.
 function pushEof(stream: NativeReadable) {
   if (!stream.destroyed) stream.push(null);
 }
@@ -228,29 +268,102 @@ function pushAndCheck(stream: NativeReadable, chunk: any) {
   }
 }
 
+function deliverChunk(stream: NativeReadable, chunk: any) {
+  const events = stream[kReadEvents];
+  // Nobody listens, so nobody sees an order, and a read() call gets the bytes now: Node has them buffered already.
+  if (
+    events === undefined ||
+    (events.length === 0 && stream.listenerCount("data") === 0 && stream.listenerCount("readable") === 0)
+  ) {
+    pushAndCheck(stream, chunk);
+  } else {
+    queueReadEvent(stream, events, kChunkEvent, chunk);
+  }
+}
+
+// Node gets the EOF in a read of its own, after the nextTicks and promise jobs of the last chunk's listeners.
+function deliverEof(stream: NativeReadable) {
+  stream[kSawEof] = true;
+  const events = stream[kReadEvents];
+  if (events === undefined) runAfterTickDrain(pushEof, stream);
+  else queueReadEvent(stream, events, kEofEvent, undefined);
+}
+
+function queueReadEvent(stream: NativeReadable, events: unknown[], kind: number, value: unknown) {
+  $arrayPush(events, kind);
+  $arrayPush(events, value);
+  queueReadEventDelivery(stream);
+}
+
+function queueReadEventDelivery(stream: NativeReadable) {
+  if (stream[kReadEventQueued]) return;
+  stream[kReadEventQueued] = true;
+  runAfterTickDrain(deliverReadEvent, stream);
+}
+
+function deliverReadEvent(stream: NativeReadable) {
+  stream[kReadEventQueued] = false;
+  const events = stream[kReadEvents]!;
+  // Node reports nothing more for a handle that is closed.
+  if (stream.destroyed) {
+    events.length = 0;
+    stream[kReadEventsHead] = 0;
+    return;
+  }
+  const head = stream[kReadEventsHead];
+  const kind = events[head];
+  const value = events[head + 1];
+  if (head + 2 === events.length) {
+    events.length = 0;
+    stream[kReadEventsHead] = 0;
+  } else {
+    stream[kReadEventsHead] = head + 2;
+    // Before the push, so that a listener that throws does not strand the events behind this one.
+    queueReadEventDelivery(stream);
+  }
+  if (kind === kEofEvent) {
+    stream.push(null);
+  } else if (kind === kErrorEvent) {
+    errorOrDestroy(stream, value);
+  } else {
+    let threw = true;
+    try {
+      pushAndCheck(stream, value);
+      threw = false;
+    } finally {
+      // The throw of a 'data' listener skipped the maybeReadMore() of push(). Node's handle reads on.
+      if (threw) process.nextTick(readAfterListenerThrew, stream);
+    }
+  }
+}
+
+function readAfterListenerThrew(stream: NativeReadable) {
+  if (!stream.destroyed) stream.read(0);
+}
+
 function handleNumberResult(stream: NativeReadable, result: number, chunk: any, isClosed: boolean) {
   if (result > 0) {
     const slice = chunk.subarray(0, result);
     chunk = slice.byteLength < chunk.byteLength ? chunk.subarray(result) : undefined;
     if (slice.byteLength > 0) {
-      pushAndCheck(stream, slice);
+      deliverChunk(stream, slice);
     }
   }
 
   if (isClosed) {
-    process.nextTick(pushEof, stream);
+    deliverEof(stream);
   }
 
   return chunk;
 }
 
 function handleArrayBufferViewResult(stream: NativeReadable, result: any, chunk: any, isClosed: boolean) {
   if (result.byteLength > 0) {
-    pushAndCheck(stream, result);
+    deliverChunk(stream, result);
   }
 
   if (isClosed) {
-    process.nextTick(pushEof, stream);
+    deliverEof(stream);
   }
 
   return chunk;
@@ -261,19 +374,29 @@ function adjustHighWaterMark(stream: NativeReadable) {
   stream[kHasResized] = true;
 }
 
-function destroy(this: NativeReadable, error: any, cb: () => void) {
+function destroy(this: NativeReadable, error: any, cb: (error?: any) => void) {
   const ptr = this.$bunNativePtr;
   if (ptr) {
     ptr.cancel(error);
   }
+  if (this[kReadEvents] !== undefined) {
+    // As net.Socket: 'close' comes from the handle's close callback. It is queued before cb(), which can throw.
+    runAfterTickDrain(emitClose, this);
+    if (cb) cb(error);
+    return;
+  }
   dropReadAhead(this);
   if (cb) {
     // `_destroy` reports its error through the callback.
     process.nextTick(cb, error);
   }
 }
 
-// `_read()` pushes synchronously, so flow() stays one chunk ahead of the 'data' listener. Node's async sources do not.
+function emitClose(stream: NativeReadable) {
+  stream.emit("close");
+}
+
+// `_read()` of a "webstream" source pushes synchronously, so flow() stays one chunk ahead of the 'data' listener. Node's async sources do not.
 function dropReadAhead(stream: NativeReadable) {
   const state = stream._readableState;
   // Paused: Node has this buffered too, and a later read() returns it.

src/js/internal/webstreams_adapters.ts

@@ -38,7 +38,7 @@ function tryTransferToNativeReadable(stream, options) {
   if (!ptr || ptr === -1) {
     return undefined;
   }
-  return require("internal/streams/native-readable").constructNativeReadable(stream, options);
+  return require("internal/streams/native-readable").constructNativeReadable(stream, options, "webstream");
 }
 
 class ReadableFromWeb extends Readable {