diff --git a/src/js/builtins/ProcessObjectInternals.ts b/src/js/builtins/ProcessObjectInternals.ts index 8ea675755e15..cb5de79816a7 100644 --- a/src/js/builtins/ProcessObjectInternals.ts +++ b/src/js/builtins/ProcessObjectInternals.ts @@ -328,12 +328,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; @@ -373,7 +374,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?.()); // Without this, every checkpoint after the first tick calls this function to find an empty queue. $putInternalField(nextTickQueue, 0, 0); } @@ -384,6 +386,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) { @@ -402,7 +413,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 diff --git a/src/js/internal/async_hooks_tick.ts b/src/js/internal/async_hooks_tick.ts index ce97c1cd6a81..e597368dce07 100644 --- a/src/js/internal/async_hooks_tick.ts +++ b/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, }; diff --git a/src/js/internal/process/after_tick_drain.ts b/src/js/internal/process/after_tick_drain.ts new file mode 100644 index 000000000000..a9208aae2abe --- /dev/null +++ b/src/js/internal/process/after_tick_drain.ts @@ -0,0 +1,41 @@ +// 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"); +const Dequeue = require("internal/fifo"); + +type AfterTickDrainCallback = { + callback: (arg: any) => void; + arg: unknown; + frame: import("../../node/async_hooks").Frame | undefined; +}; + +const entries = new Dequeue(); + +// 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(); + entries.push({ callback, arg, frame: $getInternalField($asyncContext, 0) }); +} + +// The tick loop calls this each time both queues are empty. False: nothing waited. +function runAfterTickDrainCallback() { + const entry = entries.shift(); + if (entry === undefined) return false; + const { callback, arg, frame } = entry; + 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 }; diff --git a/src/js/internal/streams/native-readable.ts b/src/js/internal/streams/native-readable.ts index 3d62e724416f..1831b17872d3 100644 --- a/src/js/internal/streams/native-readable.ts +++ b/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) ? "" : $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,17 +268,90 @@ 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; @@ -246,11 +359,11 @@ function handleNumberResult(stream: NativeReadable, result: number, chunk: any, 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,11 +374,17 @@ 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(error ? emitCloseAfterError : emitClose, this); + if (cb) cb(error); + return; + } dropReadAhead(this); if (cb) { // `_destroy` reports its error through the callback. @@ -273,7 +392,16 @@ function destroy(this: NativeReadable, error: any, cb: () => void) { } } -// `_read()` pushes synchronously, so flow() stays one chunk ahead of the 'data' listener. Node's async sources do not. +// The argument is net.Socket's `hadError`: https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L346-L358 +function emitClose(stream: NativeReadable) { + stream.emit("close", false); +} + +function emitCloseAfterError(stream: NativeReadable) { + stream.emit("close", true); +} + +// `_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. diff --git a/src/js/internal/webstreams_adapters.ts b/src/js/internal/webstreams_adapters.ts index a59bea657e37..29c62c98b58c 100644 --- a/src/js/internal/webstreams_adapters.ts +++ b/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 { diff --git a/src/js/node/child_process.ts b/src/js/node/child_process.ts index feb593027e59..cd2d9e84928f 100644 --- a/src/js/node/child_process.ts +++ b/src/js/node/child_process.ts @@ -1297,7 +1297,7 @@ class ChildProcess extends EventEmitter { return stream; } - const pipe = require("internal/streams/native-readable").constructNativeReadable(value, {}); + const pipe = require("internal/streams/native-readable").constructNativeReadable(value, {}, "libuv-handle"); this.#closesNeeded++; pipe.once("close", () => this.#maybeClose()); if (autoResume) pipe.resume(); diff --git a/test/js/bun/spawn/spawn-stdio-syscall-error.test.ts b/test/js/bun/spawn/spawn-stdio-syscall-error.test.ts index ef6eba9bc715..504041c57a32 100644 --- a/test/js/bun/spawn/spawn-stdio-syscall-error.test.ts +++ b/test/js/bun/spawn/spawn-stdio-syscall-error.test.ts @@ -293,6 +293,32 @@ child.on("close", () => { }); `; +// node:child_process: the first read gets bytes and then fails. Both come from one native read, and the +// bytes reach 'data' before 'error'. The child says when it has written, and the parent blocks until then. +const CHILD_PROCESS_FIRST_READ_FIXTURE = /* js */ ` +import { spawn } from "node:child_process"; +import { existsSync } from "node:fs"; +const marker = process.env.SPAWN_FAULT_MARKER; +const events = []; +const child = spawn("sh", ["-c", 'printf AAAABB; : > "$SPAWN_FAULT_MARKER"; exec sleep 60'], { stdio: ["ignore", "pipe", "ignore"] }); +child.stdout.on("data", chunk => events.push("stdout.data:" + chunk)); +child.stdout.on("error", e => events.push("stdout.error:" + e.code)); +child.stdout.on("close", hadError => { + events.push("stdout.close(" + hadError + ")"); + child.kill(); +}); +child.on("close", () => { + events.push("close"); + console.log(JSON.stringify({ events })); +}); +const sleeper = new Int32Array(new SharedArrayBuffer(4)); +const deadline = Date.now() + 60_000; +while (!existsSync(marker)) { + if (Date.now() > deadline) throw new Error("the child did not write"); + Atomics.wait(sleeper, 0, 0, 5); +} +`; + // node:child_process: the same for a consumer that reads on 'readable'. Once the buffer is at the highWaterMark the // reader stops, and the consumer's read() starts it again: a read the reader begins itself, not one it does for a pull. const CHILD_PROCESS_READABLE_FIXTURE = /* js */ ` @@ -387,6 +413,7 @@ beforeAll(async () => { "stdout-write.mjs": STDOUT_WRITE_FIXTURE, "child-process.mjs": CHILD_PROCESS_FIXTURE, "child-process-bytes.mjs": CHILD_PROCESS_BYTES_FIXTURE, + "child-process-first-read.mjs": CHILD_PROCESS_FIRST_READ_FIXTURE, "child-process-readable.mjs": CHILD_PROCESS_READABLE_FIXTURE, "stdout-stream-mid-fill.mjs": STDOUT_STREAM_MID_FILL_FIXTURE, "spawn-sync.mjs": SPAWN_SYNC_FIXTURE, @@ -501,8 +528,7 @@ describe.skipIf(!isLinux || !cc)("subprocess stdio syscall errors", () => { }); }); - // child.stdout reads one chunk ahead of its 'data' listener, so once the stream flows a chunk is still buffered - // when a later read fails. The stream is destroyed with the error, and that chunk has to reach 'data' first. + // The stream is destroyed with the error of a later read. Every chunk that was read before it has to reach 'data' first. test.concurrent("node:child_process: stdout delivers every byte read before the error", async () => { const report = join(String(dir), "recv-report.txt"); expect( @@ -514,6 +540,19 @@ describe.skipIf(!isLinux || !cc)("subprocess stdio syscall errors", () => { }); }); + test.concurrent("node:child_process: stdout delivers the bytes of the read that fails before the error", async () => { + expect( + await runWithFault("child-process-first-read.mjs", { + SPAWN_FAULT_RECV_AT: "2", + SPAWN_FAULT_MARKER: join(String(dir), "first-read-written"), + }), + ).toEqual({ + parsed: { events: ["stdout.data:AAAABB", "stdout.error:EIO", "stdout.close(true)", "close"] }, + stderr: "", + exitCode: 0, + }); + }); + // One wakeup can receive bytes and then fail. The reader delivers the bytes, then the error, and the consumer asks // for more from inside that delivery. That request must not read the socket past the error: with a failure that // does not repeat, the error would come after bytes read later, or never if those reads reach EOF first. diff --git a/test/js/node/child_process/child-process-stdio.test.js b/test/js/node/child_process/child-process-stdio.test.js index 3771ddacfbab..523dcf7c64fc 100644 --- a/test/js/node/child_process/child-process-stdio.test.js +++ b/test/js/node/child_process/child-process-stdio.test.js @@ -1,5 +1,5 @@ import { describe, expect, it } from "bun:test"; -import { bunEnv, bunExe, isPosix } from "harness"; +import { bunEnv, bunExe, isPosix, isWindows, tempDir } from "harness"; import { execSync, spawn } from "node:child_process"; import { once } from "node:events"; import { join } from "node:path"; @@ -220,3 +220,104 @@ describe.skipIf(!isPosix)("stdio handed to the child", () => { expect(exitCode).toBe(0); }); }); + +// Node reads a child's stdio from a libuv handle. It reports each chunk, the EOF and the close from +// a libuv callback of its own, and it runs every nextTick and promise job between two callbacks. +// So the nextTicks and the promise jobs that the listeners of an event queue, and all that those +// queue, run before the next event. The fixture prints the same lines under Node. +describe.concurrent("child.stdout and child.stderr event order", () => { + const fixture = import.meta.dir + "/fixtures/child-process-stdio-event-order.js"; + const chain = label => `tick(${label}) job(${label}) job2(${label}) job3(${label})`; + const end = `end ${chain("end")} close(false)`; + // The consumer awaits the end of the stream. Then it listens for 'close' and awaits the child's 'close': both still come. + const awaited = "data(AAAABB) consumed close(false) late-close(false) child-close-settled"; + const expected = { + data: `data(AAAABB) ${chain("AAAABB")} ${end}`, + readable: `read(AAAABB) read(null) ${chain("AAAABB")} read(null) ${end}`, + // The listener destroys the stream from a promise job: no 'end' follows, and the rest of that job comes before 'close'. + destroy: "data(AAAABB) destroy job(destroy) tick(destroy) job2(destroy) job3(destroy) close(false)", + // destroy() swallows the throw of its callback, and 'close' still comes. + "destroy-callback-throws": "data(AAAABB) destroy-callback close(false)", + // The throw is an uncaught exception, and the stream reads on. + throw: `data(AAAABB) uncaughtException(uncaughtException) ${chain("AAAABB")} ${end}`, + // Nobody listens: a read() call gets the bytes at once. + "read-on-exit": `read(AAAABB) ${end}`, + both: `data(OUT) ${chain("OUT")} data(ERR) ${chain("ERR")}`, + "two-chunks": `data(AAAA) ${chain("AAAA")} data(BBBB) ${chain("BBBB")} ${end}`, + "for-await": awaited, + "await-end": awaited, + "held-open": `exit ${awaited}`, + }; + + it.each([ + // first-read: the bytes and the EOF are in the pipe before the parent reads for the first time. + ["stdout", "first-read", "data"], + ["stdout", "first-read", "readable"], + ["stdout", "first-read", "destroy"], + ["stdout", "first-read", "for-await"], + ["stdout", "first-read", "await-end"], + ["stderr", "first-read", "data"], + ["both", "first-read", "data"], + // waiting-read: the parent's read already waits when the child writes. + ["stdout", "waiting-read", "data"], + ["stdout", "waiting-read", "readable"], + ["stdout", "waiting-read", "destroy"], + ["stdout", "waiting-read", "throw"], + ["stdout", "waiting-read", "destroy-callback-throws"], + // two-chunks: the second chunk and the EOF are in the pipe while the first 'data' listener runs. + ["stdout", "two-chunks", "data"], + ...(isWindows + ? [] + : [ + // Not on Windows. There Node runs the promise jobs of a listener that threw after 'end', when + // the EOF is already in the pipe. And Bun reads a pipe asynchronously there, so a read() call + // in 'exit' finds no bytes yet. + ["stdout", "first-read", "throw"], + ["stdout", "first-read", "read-on-exit"], + // held-open: a grandchild holds the pipe until the parent has seen the child's 'exit', so the + // 'close' of the pipe is also the 'close' of the child. Not checked on Windows. + ["stdout", "held-open", "for-await"], + ["stdout", "held-open", "await-end"], + ]), + ])("%s, %s, %s", async (which, mode, consumer) => { + using dir = tempDir("child-stdio-event-order", {}); + await using proc = Bun.spawn({ + cmd: [bunExe(), fixture, which, mode, consumer, String(dir)], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout: stdout.trim(), stderr, exitCode }).toEqual({ + stdout: expected[which === "both" ? which : mode === "two-chunks" || mode === "held-open" ? mode : consumer], + stderr: "", + exitCode: 0, + }); + }); + + // The events above wait for the tick loop, not for process.nextTick(), which fake timers replace + // (they read it first). The loop that flushes the held ticks stands in for the fake clock. + it.each([ + ["before its first read", "process.nextTick = held;"], + ["after a read", "const real = process.nextTick; process.nextTick = held;"], + ])("child.stdout delivers when process.nextTick is replaced %s", async (_, replace) => { + const script = ` + const ticks = []; + const held = (fn, ...args) => void ticks.push(() => fn(...args)); + ${replace} + const flush = setInterval(() => { while (ticks.length) ticks.shift()(); }, 1); + const { spawn } = require("node:child_process"); + const events = []; + const child = spawn(process.execPath, ["-e", "process.stdout.write('hi')"], { stdio: ["ignore", "pipe", "ignore"] }); + child.stdout.on("data", chunk => events.push("data(" + chunk + ")")); + child.stdout.on("end", () => events.push("end")); + child.on("close", () => { + clearInterval(flush); + console.log(events.join(" ")); + }); + `; + await using proc = Bun.spawn({ cmd: [bunExe(), "-e", script], env: bunEnv, stdout: "pipe", stderr: "pipe" }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout: stdout.trim(), stderr, exitCode }).toEqual({ stdout: "data(hi) end", stderr: "", exitCode: 0 }); + }); +}); diff --git a/test/js/node/child_process/child_process.test.ts b/test/js/node/child_process/child_process.test.ts index 8cd1bb81e8ae..3eccc4d2ea3a 100644 --- a/test/js/node/child_process/child_process.test.ts +++ b/test/js/node/child_process/child_process.test.ts @@ -1534,12 +1534,13 @@ it("child.stdout.pause() after flowing stops native reads and blocks the child", } }); -// child.stdout and child.stderr read ahead: `_read()` pushes each native pull -// result synchronously, so while the stream flows Readable holds the next chunk -// in its buffer when a 'data' listener runs. destroy() left that chunk there and -// flow() emitted it after destroy() returned, with `destroyed === true`. Node's +// child.stdout and child.stderr used to read ahead: `_read()` pushed each native +// pull result synchronously, so while the stream flowed Readable held the next +// chunk in its buffer when a 'data' listener ran. destroy() left that chunk there +// and flow() emitted it after destroy() returned, with `destroyed === true`. Node's // child stdio is a net.Socket that pushes asynchronously, so no 'data' follows -// destroy() there. +// destroy() there. Now each read result waits for the listeners of the one +// before, and destroy() drops what waits. it.concurrent.each(["stdout", "stderr"] as const)( "child.%s.destroy() inside a 'data' listener stops 'data' and 'end'", async name => { diff --git a/test/js/node/child_process/fixtures/child-process-stdio-event-order.js b/test/js/node/child_process/fixtures/child-process-stdio-event-order.js new file mode 100644 index 000000000000..b995ac8a27c9 --- /dev/null +++ b/test/js/node/child_process/fixtures/child-process-stdio-event-order.js @@ -0,0 +1,177 @@ +// Prints the order of a child's stdio stream events, together with the nextTicks and the promise +// jobs that the stream's listeners queue. Node prints the same line. +// +// usage: child-process-stdio-event-order.js +// +// mode: +// first-read the bytes and the EOF are both in the pipe before the parent reads for the first time +// waiting-read the parent's read already waits when the child writes +// two-chunks the second chunk and the EOF are in the pipe while the 'data' listener of the first chunk runs +// held-open the child exits at once, and a grandchild holds the pipe: it writes when the parent has seen 'exit' +// consumer: +// data a 'data' listener +// readable a 'readable' listener that calls read() +// destroy a 'data' listener that destroys the stream from a promise job +// throw a 'data' listener that throws +// destroy-callback-throws a 'data' listener that calls destroy() with a callback that throws +// read-on-exit no listener, and a read() call when the child exits +// for-await a `for await` loop, and after it a 'close' listener and `await once(child, "close")` +// await-end a 'data' listener and `await once(stream, "end")`, and after it the same two +// "both" reads stdout and stderr, and prints only their 'data' events. +const { spawn } = require("node:child_process"); +const { once } = require("node:events"); +const fs = require("node:fs"); +const path = require("node:path"); + +const [which, mode, consumer, dir] = process.argv.slice(2); +const fds = which === "both" ? [1, 2] : which === "stdout" ? [1] : [2]; +// The parent creates `go`. The child creates `written` after it wrote everything and closed the fds. +const go = path.join(dir, "go"); +const written = path.join(dir, "written"); + +const sleeper = new Int32Array(new SharedArrayBuffer(4)); +// Blocks the event loop, so that nothing is read or delivered until `file` exists. +function blockUntilExists(file) { + const deadline = Date.now() + 60_000; + while (!fs.existsSync(file)) { + // The test is over and has removed its directory. + if (!fs.existsSync(dir)) process.exit(1); + if (Date.now() > deadline) throw new Error(`${file} was not created`); + Atomics.wait(sleeper, 0, 0, 5); + } +} + +// The child closes the fd itself, so the EOF is in the pipe before `written` exists. (On Windows +// closeSync() of fd 1 or 2 does nothing, and the EOF follows when the child exits.) +const writerSource = ` + const fs = require("node:fs"); + const sleeper = new Int32Array(new SharedArrayBuffer(4)); + const dir = ${JSON.stringify(dir)}; + ${blockUntilExists} + const mode = ${JSON.stringify(mode)}; + const fds = ${JSON.stringify(fds)}; + if (mode === "two-chunks") fs.writeSync(fds[0], "AAAA"); + if (mode !== "first-read") blockUntilExists(${JSON.stringify(go)}); + for (const fd of fds) fs.writeSync(fd, fds.length === 2 ? (fd === 1 ? "OUT" : "ERR") : mode === "two-chunks" ? "BBBB" : "AAAABB"); + for (const fd of fds) fs.closeSync(fd); + fs.writeFileSync(${JSON.stringify(written)}, ""); +`; +// held-open: the writer is a grandchild that inherits the pipe, and the child exits without a wait for it. +const childSource = + mode === "held-open" + ? `require("node:child_process").spawn(process.execPath, ["-e", ${JSON.stringify(writerSource)}], { stdio: "inherit" }).unref();` + : writerSource; + +const stdio = ["ignore", "ignore", "ignore"]; +for (const fd of fds) stdio[fd] = "pipe"; +const env = { ...process.env }; +// The test runner sets this for ASAN builds. With it a Bun child ends its own children when it exits, and the +// grandchild has to outlive the child here. +if (mode === "held-open") delete env.BUN_FEATURE_FLAG_NO_ORPHANS; +const child = spawn(process.execPath, ["-e", childSource], { stdio, env }); + +const events = []; +// The chain goes from the promise job queue to the nextTick queue and back, twice. +async function queueTickAndJobs(label) { + process.nextTick(() => events.push(`tick(${label})`)); + await null; + events.push(`job(${label})`); + await new Promise(resolve => process.nextTick(resolve)); + events.push(`job2(${label})`); + await new Promise(resolve => process.nextTick(resolve)); + events.push(`job3(${label})`); +} + +const awaitsEnd = consumer === "for-await" || consumer === "await-end"; +for (const fd of fds) { + const stream = child.stdio[fd]; + if (consumer === "data") { + stream.on("data", chunk => { + events.push(`data(${chunk})`); + queueTickAndJobs(chunk); + if (mode === "two-chunks" && !fs.existsSync(go)) { + fs.writeFileSync(go, ""); + blockUntilExists(written); + } + }); + } else if (consumer === "readable") { + stream.on("readable", () => { + let chunk; + while ((chunk = stream.read()) !== null) { + events.push(`read(${chunk})`); + queueTickAndJobs(chunk); + } + events.push("read(null)"); + }); + } else if (consumer === "destroy") { + stream.on("data", async chunk => { + events.push(`data(${chunk})`); + await null; + stream.destroy(); + events.push("destroy"); + queueTickAndJobs("destroy"); + }); + } else if (consumer === "destroy-callback-throws") { + // destroy() swallows what its callback throws, and 'close' still comes. + process.on("uncaughtException", () => events.push("uncaughtException")); + stream.on("data", chunk => { + events.push(`data(${chunk})`); + stream.destroy(undefined, () => { + events.push("destroy-callback"); + throw new Error("the callback threw"); + }); + }); + } else if (consumer === "throw") { + // Node calls the listener from a libuv callback, so the throw is an uncaught exception, and the handle reads on. + process.on("uncaughtException", (error, origin) => events.push(`uncaughtException(${origin})`)); + process.on("unhandledRejection", () => events.push("unhandledRejection")); + stream.on("data", chunk => { + events.push(`data(${chunk})`); + queueTickAndJobs(chunk); + throw new Error("the listener threw"); + }); + } else if (awaitsEnd) { + // Node emits 'close' from the close callback of the handle, after the promise jobs that 'end' started. + (async () => { + if (consumer === "for-await") { + for await (const chunk of stream) events.push(`data(${chunk})`); + } else { + stream.on("data", chunk => events.push(`data(${chunk})`)); + await once(stream, "end"); + } + events.push("consumed"); + stream.on("close", hadError => events.push(`late-close(${hadError})`)); + await once(child, "close"); + events.push("child-close-settled"); + })(); + } else { + // Node has read the pipe from the start, so the bytes are in the stream's buffer by now. + child.on("exit", () => events.push(`read(${stream.read()})`)); + } + if (which === "both") continue; + // Not for a consumer that awaits: how many promise jobs an await takes is not what the line is about. + if (!awaitsEnd) { + stream.on("end", () => { + events.push("end"); + queueTickAndJobs("end"); + }); + } + stream.on("close", hadError => events.push(`close(${hadError})`)); +} +if (mode === "held-open") { + child.on("exit", () => { + events.push("exit"); + fs.writeFileSync(go, ""); + }); +} +child.on("close", () => { + // A job that runs after 'close' still gets into the line. + setImmediate(() => console.log(events.join(" "))); +}); + +if (mode === "first-read") { + blockUntilExists(written); +} else if (mode === "waiting-read") { + // The stream started to read in a nextTick, so the read waits by now. + setImmediate(() => fs.writeFileSync(go, "")); +} diff --git a/test/js/node/stream/node-stream.test.js b/test/js/node/stream/node-stream.test.js index 782ed9039c62..f0ad3e2cd76a 100644 --- a/test/js/node/stream/node-stream.test.js +++ b/test/js/node/stream/node-stream.test.js @@ -1,6 +1,7 @@ import { exposedInternals } from "bun:internal-for-testing"; import { describe, expect, it, jest } from "bun:test"; import { bunEnv, bunExe, bunRun, isGlibcVersionAtLeast, isMacOS, tempDir, tmpdirSync } from "harness"; +import { once } from "node:events"; import { createReadStream, mkdirSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { Duplex, duplexPair, finished, PassThrough, Readable, Stream, Transform, Writable } from "node:stream"; @@ -388,6 +389,56 @@ it("Readable.fromWeb", async () => { expect(Buffer.concat(chunks).toString()).toBe("Hello World!\n"); }); +// These web streams have a native source, which gives the last bytes and the EOF in one read. +// A nextTick of the listener is not in the order: Node pushes every fromWeb chunk from a promise +// job, so its promise jobs run first. Bun pushes a synchronous read from the nextTick that reads. +describe("Readable.fromWeb of a native stream", () => { + const sources = [ + ["Blob", () => new Blob(["AAAABB"]).stream()], + ["Response body", () => new Response("AAAABB").body], + ["Bun.file", dir => Bun.file(join(dir, "file.txt")).stream()], + ]; + + it.each(sources)("%s: 'end' comes after the promise jobs of the last 'data' listeners", async (_, makeWebStream) => { + using dir = tempDir("fromweb-end-order", { "file.txt": "AAAABB" }); + const events = []; + const readable = Readable.fromWeb(makeWebStream(String(dir))); + readable.on("data", async chunk => { + events.push(`data(${chunk})`); + await null; + events.push("job"); + await null; + events.push("job2"); + }); + readable.on("end", () => events.push("end")); + await once(readable, "close"); + expect(events).toEqual(["data(AAAABB)", "job", "job2", "end"]); + }); + + it.each(sources)( + "%s: no 'end' after a promise job of the last 'data' listener destroys it", + async (_, makeWebStream) => { + using dir = tempDir("fromweb-end-order", { "file.txt": "AAAABB" }); + const events = []; + const readable = Readable.fromWeb(makeWebStream(String(dir))); + readable.on("data", async chunk => { + events.push(`data(${chunk})`); + await null; + readable.destroy(); + events.push("destroy"); + }); + readable.on("end", () => events.push("end")); + const { promise, resolve } = Promise.withResolvers(); + finished(readable, resolve); + const [error] = await Promise.all([promise, once(readable, "close")]); + expect({ events, code: error?.code }).toEqual({ + events: ["data(AAAABB)", "destroy"], + code: "ERR_STREAM_PREMATURE_CLOSE", + }); + }, + ); +}); + // fromWeb assigns stream.$bunNativePtr on the node Readable. When user code grafts // ReadableStream.prototype into the node stream prototype chain, that put used to // reach ReadableStream's private custom setter with the Readable as the receiver,