diff --git a/src/io/PipeReader.rs b/src/io/PipeReader.rs index 457d7f49b40e..a1d20381a972 100644 --- a/src/io/PipeReader.rs +++ b/src/io/PipeReader.rs @@ -458,7 +458,11 @@ impl PosixBufferedReader { // Exists for consistently with Windows. pub fn has_pending_read(&self) -> bool { - matches!(&self.handle, PollOrFd::Poll(poll) if poll.is_registered()) + // `is_watching()` (registered && !needs-rearm) rather than + // `is_registered()`: a one-shot poll that has fired but not been + // re-armed will not deliver another callback, so callers that skip + // `read()` on "pending" must not be told one is in flight. + matches!(&self.handle, PollOrFd::Poll(poll) if poll.is_watching()) } pub fn watch(&mut self) { @@ -672,16 +676,21 @@ impl PosixBufferedReader { if streaming { // Stream this chunk and register for next cycle - if !parent.vtable.on_read_chunk( + let keep_going = parent.vtable.on_read_chunk( &stack_buffer[..bytes_read], if received_hup && bytes_read < stack_buffer.len() { ReadState::Eof } else { ReadState::Progress }, - ) && !received_hup - && !over_budget - { + ); + // Re-entrant JS inside on_read_chunk can close the + // reader (nested on_pull -> read -> EOF); the + // captured `fd` is then stale regardless of HUP. + if parent.is_done() { + return; + } + if !keep_going && !received_hup && !over_budget { return; } } else { @@ -739,6 +748,9 @@ impl PosixBufferedReader { ReadState::Progress }, ); + if parent.is_done() { + return; + } // Closing for `over_budget` outranks the // consumer asking us to stop: it must still // happen, or nothing ever caps the pipe. @@ -900,15 +912,21 @@ impl PosixBufferedReader { // Once HUP is set the kernel // returns the remaining bytes then 0, so // draining to `bytes_read == 0` is bounded. - if !parent.vtable.on_read_chunk( + let keep_going = parent.vtable.on_read_chunk( &event_loop.pipe_read_buffer_mut()[..head_start], if received_hup { ReadState::Eof } else { ReadState::Progress }, - ) && !received_hup - { + ); + // Re-entrant close (nested on_pull -> read -> + // EOF) invalidates the captured `fd`; stop + // before the next recv regardless of HUP. + if parent.is_done() { + return; + } + if !keep_going && !received_hup { return; } head_start = 0; @@ -949,15 +967,18 @@ impl PosixBufferedReader { } if head_start > 0 { - if !parent.vtable.on_read_chunk( + let keep_going = parent.vtable.on_read_chunk( &event_loop.pipe_read_buffer_mut()[..head_start], if received_hup { ReadState::Eof } else { ReadState::Progress }, - ) && !received_hup - { + ); + if parent.is_done() { + return; + } + if !keep_going && !received_hup { return; } } @@ -1048,7 +1069,7 @@ impl PosixBufferedReader { .vtable .on_read_chunk(&parent._buffer, ReadState::Progress); parent._buffer.clear(); - if !keep_going { + if parent.is_done() || !keep_going { return; } continue; @@ -1912,7 +1933,14 @@ impl WindowsBufferedReader { // grows by `amount_result` every chunk and never resets, so a 1 GB // `cat` holds 1 GB resident instead of ~64 KB. Clear it here, after // the streaming consumer has finished with `slice`. - if should_continue && has_more != ReadState::Eof && self.vtable.is_streaming_enabled() { + // `should_continue` no longer gates the clear: FileReader may say + // stop at its highwater mark while uv keeps delivering, and leaving + // `_buffer` uncleared would double-buffer (here + FileReader.buffered). + // Parents that want the reader paused call `reader().pause()` + // themselves; stopping here could free a parent whose caller still + // holds `this` (FileResponseStream on abort). + let _ = should_continue; + if has_more != ReadState::Eof && self.vtable.is_streaming_enabled() { self._buffer.clear(); } diff --git a/src/js/internal/streams/native-readable.ts b/src/js/internal/streams/native-readable.ts index c2e773947157..65307212b821 100644 --- a/src/js/internal/streams/native-readable.ts +++ b/src/js/internal/streams/native-readable.ts @@ -29,7 +29,7 @@ let dynamicallyAdjustChunkSize = (_?) => ( type NativeReadable = typeof import("node:stream").Readable & typeof import("node:stream").Stream & { - push: (chunk: any) => void; + push: (chunk: any) => boolean; $bunNativePtr?: NativePtr; [kRefCount]: number; [kCloseState]: [boolean]; @@ -48,6 +48,7 @@ interface NativePtr { pull: (view: any, closer: any) => any; updateRef: (ref: boolean) => void; cancel: (error: any) => void; + setFlowing?: (flowing: boolean) => void; } let debugId = 0; @@ -122,10 +123,14 @@ function getRemainingChunk(stream: NativeReadable, maxToRead?: number) { function read(this: NativeReadable, maxToRead: number) { $debug(`[${this.debugId}] read${this[kPendingRead] ? ", is already pending" : ""}`); + var ptr = this.$bunNativePtr; + // Readable called `_read`, so it wants data: make sure the native reader is + // 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]) { return; } - var ptr = this.$bunNativePtr; if (!ptr) { $debug(`[${this.debugId}] read, no ptr`); this.push(null); @@ -139,13 +144,13 @@ function read(this: NativeReadable, maxToRead: number) { this[kHighWaterMark] = Math.min(this[kHighWaterMark], result); } if ($isTypedArrayView(result) && result.byteLength > 0) { - this.push(result); + pushAndCheck(this, result); } const drainResult = ptr.drain(); this[kConstructed] = true; $debug(`[${this.debugId}] drain result: ${drainResult?.byteLength ?? "null"}`); if ((drainResult?.byteLength ?? 0) > 0) { - this.push(drainResult); + pushAndCheck(this, drainResult); } } const chunk = getRemainingChunk(this, maxToRead); @@ -196,12 +201,22 @@ function handleResult(stream: NativeReadable, result: any, chunk: Buffer, isClos } } +// `push()` returning false means the Readable's buffer is at/above hwm (or +// the consumer paused); stop the native reader so kernel backpressure reaches +// the writer (readStop, like net.Socket). The next `_read()` re-enables it. +function pushAndCheck(stream: NativeReadable, chunk: any) { + if (!stream.push(chunk)) { + const ptr = stream.$bunNativePtr; + if (ptr) ptr.setFlowing?.(false); + } +} + 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) { - stream.push(slice); + pushAndCheck(stream, slice); } } @@ -216,7 +231,7 @@ function handleNumberResult(stream: NativeReadable, result: number, chunk: any, function handleArrayBufferViewResult(stream: NativeReadable, result: any, chunk: any, isClosed: boolean) { if (result.byteLength > 0) { - stream.push(result); + pushAndCheck(stream, result); } if (isClosed) { diff --git a/src/runtime/webcore/FileReader.rs b/src/runtime/webcore/FileReader.rs index b205c96b724c..c4eb44b4a0c9 100644 --- a/src/runtime/webcore/FileReader.rs +++ b/src/runtime/webcore/FileReader.rs @@ -778,8 +778,15 @@ impl FileReader { unsafe { (*parent).increment_count() }; self.pending.with_mut(|p| p.run()); close_if_needed!(); - // Re-entrant cancel closed the reader; tell the io caller to stop. - let ret = if self.done.get() { false } else { ret }; + // Re-entrant cancel (sets `done`) or a nested on_pull that read to + // EOF (sets IS_DONE via on_reader_done but not `self.done`) closed + // the reader; tell the io caller to stop so it does not re-read the + // captured fd. + let ret = if self.done.get() || self.reader().is_done() { + false + } else { + ret + }; // SAFETY: see `parent()`; the pin keeps the count >= 1, so this // never frees. `self` is not accessed after. let _ = unsafe { Source::decrement_count(parent) }; @@ -794,14 +801,19 @@ impl FileReader { } // No JS read is waiting; stop at the highwater mark. onPull restarts. + // + // `started` gates the backstop: a `from_pipe()` reader for non-lazy + // `Bun.spawn` is already reading when it arrives here, and throttling + // before any consumer has attached deadlocks a child that alternates + // stdout/stderr writes while the caller only awaits one of them. // SAFETY: see `reader_buffer` decl. let reader_buffer_len = unsafe { (*reader_buffer).len() }; - let ret = self.flowing.get() - && !matches!( - self.read_inside_on_pull.get(), - ReadDuringJSOnPullResult::Temporary(_) - ) - && self.buffered.get().len() + reader_buffer_len < self.highwater_mark; + let ret = !matches!( + self.read_inside_on_pull.get(), + ReadDuringJSOnPullResult::Temporary(_) + ) && (!self.started.get() + || (self.flowing.get() + && self.buffered.get().len() + reader_buffer_len < self.highwater_mark)); close_if_needed!(); ret } @@ -950,12 +962,6 @@ impl FileReader { self.pending_value.with_mut(|p| p.set(&global, array)); self.pending_view.set(buffer); - // `has_pending_read()` tracks the registration flag, not the kernel - // arm state: the highwater backstop leaves the one-shot poll disarmed. - if self.flowing.get() { - self.reader().watch(); - } - bun_core::scoped_log!(FileReader, "onPull({}) = pending", buffer_len); streams::Result::Pending(self.pending.as_ptr()) diff --git a/test/js/node/child_process/child_process.test.ts b/test/js/node/child_process/child_process.test.ts index 57a90aa434dd..0e7299e11e1d 100644 --- a/test/js/node/child_process/child_process.test.ts +++ b/test/js/node/child_process/child_process.test.ts @@ -992,6 +992,68 @@ describe.skipIf(!isPosix)("stdout pipe backpressure", () => { }); }); +// child.stdout.pause() must stop the native reader so the kernel pipe fills +// and the child blocks on write. Previously, once the stream had flowed even +// once the native FileReader kept the poll armed (or uv_read_start active on +// Windows) regardless of JS state, so the child wrote its entire output into +// the parent's heap and 'data' kept firing after #handleOnExit resumed it. +it("child.stdout.pause() after flowing stops native reads and blocks the child", async () => { + // 20 MB: well above any kernel socket buffer, small enough to drain fast + // once resumed on ASAN. + const SIZE = 20 * 1024 * 1024; + const writer = `const c=Buffer.alloc(1<<20,97);let w=0;(function f(){while(w<${SIZE}){const n=Math.min(c.length,${SIZE}-w);w+=n;if(!process.stdout.write(c.subarray(0,n))){process.stdout.once('drain',f);return}}})()`; + const c = spawn(bunExe(), ["-e", writer], { + stdio: ["ignore", "pipe", "ignore"], + env: bunEnv, + }); + try { + let events = 0; + let bytes = 0; + let eventsAfterPause = 0; + const { promise: firstData, resolve: gotFirst, reject: failFirst } = Promise.withResolvers(); + c.on("error", failFirst); + c.on("close", () => failFirst(new Error("child closed before first 'data'"))); + c.stdout!.on("data", (d: Buffer) => { + events++; + bytes += d.length; + if (events === 1) { + c.stdout!.pause(); + gotFirst(); + } else { + eventsAfterPause++; + } + }); + await firstData; + + expect(c.stdout!.isPaused()).toBe(true); + + // Give the eager-read path every chance to over-buffer: without the fix + // the native reader has already drained most of SIZE and the child is + // racing to exit. Deadline-polled; breaks early if the bug is present. + const deadline = Date.now() + 1000; + while (c.exitCode === null && Date.now() < deadline) { + await new Promise(r => setImmediate(r)); + } + + // Core assertion: pause() applied backpressure, so the child cannot have + // finished writing SIZE bytes and is blocked on a full pipe. + expect(c.exitCode).toBeNull(); + expect(eventsAfterPause).toBe(0); + expect(c.stdout!.isPaused()).toBe(true); + // At most a few pipe-buffer-sized chunks were delivered before pause took + // effect, never the whole output. + expect(bytes).toBeLessThan(SIZE); + + // Resuming delivers every byte and the child exits cleanly. + c.stdout!.resume(); + await once(c, "close"); + expect(bytes).toBe(SIZE); + expect(c.exitCode).toBe(0); + } finally { + c.kill(); + } +}); + // When spawn fails (ENOENT, bad cwd, etc.) the ChildProcess emits 'error' and // 'close' but never 'exit'. The abort listener on options.signal was only // removed on 'exit', so every failed spawn against a shared AbortSignal leaked