diff --git a/src/io/PipeReader.rs b/src/io/PipeReader.rs index 83f8c081e15..0866e6f75af 100644 --- a/src/io/PipeReader.rs +++ b/src/io/PipeReader.rs @@ -74,6 +74,11 @@ pub trait BufferedReaderParent { /// Mirrors `@hasDecl(Type, "onReadChunk")`. const HAS_ON_READ_CHUNK: bool = true; + /// `chunk` is valid for this call only: it is either the per-loop scratch + /// buffer, refilled by the next read, or a heap buffer the read loop + /// clears or frees as soon as this returns. Copy what must outlive the + /// call (in particular anything reported back from a read that the + /// parent itself issued synchronously, e.g. `FileReader::on_pull`). unsafe fn on_read_chunk(this: *mut Self, chunk: &[u8], has_more: ReadState) -> bool { let _ = (this, chunk, has_more); // Default: should not be called when HAS_ON_READ_CHUNK == false. diff --git a/src/runtime/webcore/FileReader.rs b/src/runtime/webcore/FileReader.rs index 35754fe0caa..2534ce6f2de 100644 --- a/src/runtime/webcore/FileReader.rs +++ b/src/runtime/webcore/FileReader.rs @@ -21,9 +21,9 @@ use crate::webcore::streams; bun_core::declare_scope!(FileReader, visible); -// `pending_view` and the `Js`/`Temporary` variants below borrow into a -// JS-owned typed-array buffer kept alive by `pending_value: Strong` / `ensure_still_alive`. -// Represented as unbounded `&mut [u8]` / `&[u8]` here to keep function bodies +// `pending_view` and the `Js` variant below borrow into a JS-owned typed-array +// buffer kept alive by `pending_value: Strong` / `ensure_still_alive`. +// Represented as unbounded `&mut [u8]` here to keep function bodies // readable; TODO(refactor): replace with a proper raw-slice wrapper (BACKREF lifetime). // R-2 (host-fn re-entrancy): every JS-exposed / vtable-reachable method takes @@ -102,18 +102,23 @@ impl Default for FileReader { pub type IOReader = BufferedReader; +/// What the synchronous read issued by `on_pull` has produced so far. A chunk +/// handed to `on_read_chunk` is only valid during that call (it is the +/// reader's scratch, or a heap buffer the reader frees or reuses as soon as +/// the call returns), so every variant owns or has already copied its bytes. #[derive(strum::IntoStaticStr)] pub enum ReadDuringJSOnPullResult { None, - // TODO(refactor): `&'static mut` forge — sibling `static-widen-mut` pattern; - // see note on `FileReader::pending_view`. - Js(&'static mut [u8]), - AmountRead(usize), - /// Borrows the reader/JS buffer for the duration of one `on_pull` call - /// only. Holder-lifetime, not process-lifetime — `RawSlice` per - /// `bun_ptr::Interned` Population-B triage. - Temporary(bun_ptr::RawSlice), - UseBuffered(usize), + /// Chunks are copied straight into the pull buffer `buffer` (`filled` + /// bytes so far) while they fit. + // `&'static mut` forge — sibling `static-widen-mut` pattern; see note on + // `FileReader::pending_view`. + Js { + buffer: &'static mut [u8], + filled: usize, + }, + /// Everything read so far is in `FileReader::buffered`. + UseBuffered, } impl ReadDuringJSOnPullResult { @@ -756,29 +761,26 @@ impl FileReader { // `&self`; `self.buffered` is a disjoint `JsCell` so nested access // inside the closure is sound. self.read_inside_on_pull.with_mut(|riop| match riop { - ReadDuringJSOnPullResult::Js(in_progress) => { - if in_progress.len() >= buf.len() && !has_more { - in_progress[0..buf.len()].copy_from_slice(buf); - let remaining: *mut [u8] = &raw mut in_progress[buf.len()..]; - // SAFETY: lifetime laundering — see the `static-widen-mut` note on `ReadDuringJSOnPullResult::Js`. - let remaining = unsafe { &mut *remaining }; - *riop = ReadDuringJSOnPullResult::Js(remaining); - } else if !in_progress.is_empty() && !has_more { - // `buf` outlives the `on_pull` call that consumes this - // variant; holder-lifetime, encoded as `RawSlice`. - *riop = ReadDuringJSOnPullResult::Temporary(bun_ptr::RawSlice::new(buf)); - } else if has_more && !is_slice_in_vec_capacity(buf, self.buffered.get()) { - self.buffered.with_mut(|b| b.extend_from_slice(buf)); - *riop = ReadDuringJSOnPullResult::UseBuffered(buf.len()); + ReadDuringJSOnPullResult::Js { buffer, filled } => { + let free = &mut buffer[*filled..]; + if !has_more && free.len() >= buf.len() { + free[..buf.len()].copy_from_slice(buf); + *filled += buf.len(); + } else { + // `buf` dies with this call, and `on_pull` returns + // either the pull buffer or `buffered`, so anything + // already copied into the former moves along with it. + self.buffered.with_mut(|b| { + b.extend_from_slice(&buffer[..*filled]); + b.extend_from_slice(buf); + }); + *riop = ReadDuringJSOnPullResult::UseBuffered; } } - ReadDuringJSOnPullResult::UseBuffered(original) => { - let original = *original; + ReadDuringJSOnPullResult::UseBuffered => { self.buffered.with_mut(|b| b.extend_from_slice(buf)); - *riop = ReadDuringJSOnPullResult::UseBuffered(buf.len() + original); } ReadDuringJSOnPullResult::None => unreachable!(), - _ => panic!("Invalid state"), }); } else if self.pending.get().state == streams::PendingState::Pending { // Certain readers (such as pipes) may return 0-byte reads even when @@ -942,12 +944,9 @@ impl FileReader { // 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 = !matches!( - self.read_inside_on_pull.get(), - ReadDuringJSOnPullResult::Temporary(_) - ) && (!self.started.get() + let ret = !self.started.get() || (self.flowing.get() - && self.buffered.get().len() + reader_buffer_len < self.highwater_mark)); + && self.buffered.get().len() + reader_buffer_len < self.highwater_mark); close_if_needed!(); ret } @@ -1016,8 +1015,12 @@ impl FileReader { } let buffer_len = buffer.len(); + // `drain()` returned early otherwise. The `Js` arm of `on_read_chunk` + // copies into the pull buffer first, which would deliver ahead of + // anything still sitting in `buffered`. + debug_assert!(self.buffered.get().is_empty()); self.read_inside_on_pull - .set(ReadDuringJSOnPullResult::Js(buffer)); + .set(ReadDuringJSOnPullResult::Js { buffer, filled: 0 }); // SAFETY: the reader cell is live for `self`'s lifetime; `read` is // the raw re-entrancy-safe entry (its dispatch runs user JS). unsafe { IOReader::read(self.reader.get()) }; @@ -1027,45 +1030,33 @@ impl FileReader { .read_inside_on_pull .replace(ReadDuringJSOnPullResult::None); match pulled { - ReadDuringJSOnPullResult::Js(remaining_buf) => { - let amount_read = buffer_len - remaining_buf.len(); + ReadDuringJSOnPullResult::Js { buffer, filled } => { + bun_core::scoped_log!(FileReader, "onPull({}) = {}", buffer_len, filled); - bun_core::scoped_log!(FileReader, "onPull({}) = {}", buffer_len, amount_read); - - if amount_read > 0 { + if filled > 0 { if self.reader().is_done() { return streams::Result::IntoArrayAndDone(streams::IntoArray { value: array, - len: amount_read as u64, // @truncate + len: filled as u64, // @truncate }); } return streams::Result::IntoArray(streams::IntoArray { value: array, - len: amount_read as u64, // @truncate + len: filled as u64, // @truncate }); } if self.reader().is_done() { return streams::Result::Done; } - // fallthrough — but `buffer` was moved into read_inside_on_pull. - // Recover it from `remaining_buf` (amount_read == 0 ⇒ same slice). let global = self.parent_global(); self.pending_value.with_mut(|p| p.set(&global, array)); - self.pending_view.set(remaining_buf); + self.pending_view.set(buffer); bun_core::scoped_log!(FileReader, "onPull({}) = pending", buffer_len); return streams::Result::Pending(self.pending.as_ptr()); } - ReadDuringJSOnPullResult::Temporary(buf) => { - bun_core::scoped_log!(FileReader, "onPull({}) = {}", buffer_len, buf.len()); - if self.reader().is_done() { - return streams::Result::TemporaryAndDone(buf); - } - - return streams::Result::Temporary(buf); - } - ReadDuringJSOnPullResult::UseBuffered(_) => { + ReadDuringJSOnPullResult::UseBuffered => { bun_core::scoped_log!( FileReader, "onPull({}) = {}", @@ -1078,17 +1069,8 @@ impl FileReader { } return streams::Result::Owned(Vec::::move_from_list(buffered)); } - _ => { - // Falls through to set - // `pending_view = buffer`. The only variants reaching this arm - // are `None` (impossible — we just stored `Js(buffer)` above and - // `on_read_chunk` never sets `None`) and `AmountRead` (never - // produced by `on_read_chunk`). Unreachable in the current state - // machine; if that invariant ever changes, the buffer slice must - // be recovered from a captured raw ptr+len before the move. - unreachable!( - "on_read_chunk never yields None/AmountRead while read_inside_on_pull == Js" - ); + ReadDuringJSOnPullResult::None => { + unreachable!("on_read_chunk never resets read_inside_on_pull to None") } } } diff --git a/test/js/node/child_process/child_process.test.ts b/test/js/node/child_process/child_process.test.ts index a7bf59ba32a..eb636989347 100644 --- a/test/js/node/child_process/child_process.test.ts +++ b/test/js/node/child_process/child_process.test.ts @@ -1,7 +1,18 @@ import { semver, write } from "bun"; import { afterAll, beforeEach, describe, expect, it } from "bun:test"; import fs from "fs"; -import { bunEnv, bunExe, isLinux, isPosix, isWindows, nodeExe, runBunInstall, shellExe, tmpdirSync } from "harness"; +import { + bunEnv, + bunExe, + isLinux, + isPosix, + isWindows, + nodeExe, + runBunInstall, + shellExe, + tempDir, + tmpdirSync, +} from "harness"; import { ChildProcess, exec, execFile, execFileSync, execSync, fork, spawn, spawnSync } from "node:child_process"; import { getEventListeners, once, setMaxListeners } from "node:events"; import { promisify } from "node:util"; @@ -1148,6 +1159,111 @@ describe.skipIf(!isPosix)("stdout pipe backpressure", () => { }); }); +// A 'data' handler runs inside the native read loop that delivered its chunk, +// and node streams pull again from there, so that pull reads synchronously, +// nested in the outer loop, and can run all the way to EOF. The nested read +// collects its bytes in a heap buffer the reader frees as soon as the chunk +// has been handed over; when the chunk did not fit the pull buffer, +// FileReader kept pointing at it until the pull returned (heap-use-after-free +// under ASAN, corrupt or short output otherwise). +// +// Pinned down with two markers: the head is bigger than half of the reader's +// 256 KiB scratch buffer, so it is flushed to JS from the middle of the outer +// loop, and the 'data' handler does not return until the tail and EOF are in +// the socket. The tail is bigger than the 64 KiB pull buffer. +describe.skipIf(!isPosix)("child.stdout pull nested in a 'data' event", () => { + it("delivers a tail read to EOF that does not fit the pull buffer", async () => { + const HEAD = 136 * 1024; + const TAIL = 96 * 1024; + using dir = tempDir("child-stdout-nested-pull", { + "producer.js": ` + const fs = require("node:fs"); + const [headMarker, headDone, tailMarker, tailDone] = process.argv.slice(2); + const deadline = Date.now() + 15_000; + function waitFor(file) { + while (!fs.existsSync(file)) { + if (Date.now() > deadline) throw new Error("producer timed out waiting for " + file); + Bun.sleepSync(1); + } + } + function writeAll(buf) { + for (let off = 0; off < buf.length; ) off += fs.writeSync(1, buf, off); + } + waitFor(headMarker); + writeAll(Buffer.alloc(${HEAD}, "h")); + fs.writeFileSync(headDone, ""); + waitFor(tailMarker); + writeAll(Buffer.alloc(${TAIL}, "t")); + fs.closeSync(1); + fs.writeFileSync(tailDone, ""); + `, + "reader.js": ` + const { spawn } = require("node:child_process"); + const fs = require("node:fs"); + const path = require("node:path"); + const file = name => path.join(__dirname, name); + const deadline = Date.now() + 15_000; + function waitFor(name) { + while (!fs.existsSync(file(name))) { + if (Date.now() > deadline) throw new Error("reader timed out waiting for " + name); + Bun.sleepSync(1); + } + } + const child = spawn( + process.execPath, + [file("producer.js"), file("head"), file("head-done"), file("tail"), file("tail-done")], + { stdio: ["ignore", "pipe", "inherit"] }, + ); + const chunks = []; + child.stdout.on("data", chunk => { + chunks.push(chunk); + if (chunks.length === 1) { + fs.writeFileSync(file("tail"), ""); + waitFor("tail-done"); + } + }); + child.on("close", exitCode => { + const out = Buffer.concat(chunks); + console.log( + JSON.stringify({ + exitCode, + firstChunkOverHalfScratch: chunks[0].length > 128 * 1024, + length: out.length, + head: out.subarray(0, ${HEAD}).equals(Buffer.alloc(${HEAD}, "h")), + tail: out.subarray(${HEAD}).equals(Buffer.alloc(${TAIL}, "t")), + }), + ); + }); + // By the time this runs the stream has pulled once and found the + // socket empty. Block until the whole head is queued so that the + // read woken by it sees all of the head at once. + setImmediate(() => { + fs.writeFileSync(file("head"), ""); + waitFor("head-done"); + }); + `, + }); + await using proc = Bun.spawn({ + cmd: [bunExe(), "reader.js"], + cwd: 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, stderr, exitCode }).toEqual({ + stdout: + JSON.stringify({ exitCode: 0, firstChunkOverHalfScratch: true, length: HEAD + TAIL, head: true, tail: true }) + + "\n", + stderr: "", + exitCode: 0, + }); + // The budget is for the failure modes: a symbolized ASAN report takes + // several seconds, and the fixtures give up on their markers after 15s so + // that their own error, not a test timeout, is what gets reported. + }, 30_000); +}); + // 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