diff --git a/src/runtime/node/node_fs.rs b/src/runtime/node/node_fs.rs index 86c0e9fd6c71..7614e5075090 100644 --- a/src/runtime/node/node_fs.rs +++ b/src/runtime/node/node_fs.rs @@ -10,6 +10,7 @@ use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use crate::api::bun::process::event_loop_handle_to_ctx; use crate::webcore; use bun_core::Environment; +use bun_core::zig_string::Slice as ZigStringSlice; use bun_core::{String as BunString, ZStr, ZigString}; use bun_event_loop::AnyTaskWithExtraContext::AnyTaskWithExtraContext; use bun_event_loop::MiniEventLoop::MiniEventLoop; @@ -1087,6 +1088,7 @@ mod _async_tasks { args::Rename, args::Truncate, args::FdVectorIo, + args::Writev, args::FTruncate, args::Chown, args::Lutimes, @@ -2817,11 +2819,10 @@ pub mod args { } } - /// Shared layout for `fs.writev` / `fs.readv` arguments. One concrete - /// struct; we re-export both - /// names as type aliases so every `args::Writev` / `args::Readv` caller - /// (UVFSRequest params, `readv`/`writev`/`preadv_inner`/`pwritev_inner`, - /// uv dispatch arms) is untouched. + /// Shared layout for `fs.writev` / `fs.readv` arguments. `args::Readv` is + /// a direct alias; `args::Writev` newtypes it so its `from_js` can snapshot + /// resizable inputs on the async path while every + /// `.fd`/`.buffers`/`.position` consumer stays unchanged via `Deref`. pub struct FdVectorIo { pub fd: FD, pub buffers: VectorArrayBuffer, @@ -2873,9 +2874,45 @@ pub mod args { }) } } - pub type Writev = FdVectorIo; pub type Readv = FdVectorIo; + /// Newtype over [`FdVectorIo`] so writev's `from_js` can snapshot resizable + /// inputs on the async path; `Readv` stays the plain alias. + #[repr(transparent)] + pub struct Writev(pub FdVectorIo); + impl core::ops::Deref for Writev { + type Target = FdVectorIo; + #[inline] + fn deref(&self) -> &FdVectorIo { + &self.0 + } + } + impl core::ops::DerefMut for Writev { + #[inline] + fn deref_mut(&mut self) -> &mut FdVectorIo { + &mut self.0 + } + } + impl Unprotect for Writev { + #[inline] + fn unprotect(&mut self) { + self.0.unprotect(); + } + } + impl Writev { + #[inline] + pub fn to_thread_safe(&mut self) { + self.0.to_thread_safe(); + } + pub fn from_js(ctx: &JSGlobalObject, arguments: &mut ArgumentsSlice) -> JsResult { + let mut inner = FdVectorIo::from_js(ctx, arguments)?; + if arguments.will_be_async { + inner.buffers.snapshot_resizable_inputs(ctx); + } + Ok(Self(inner)) + } + } + pub struct FTruncate { pub fd: FD, pub len: Option, @@ -3885,11 +3922,27 @@ pub mod args { } if arguments.will_be_async && matches!(args.buffer, StringOrBuffer::Buffer(_)) { if let Some(pinned) = bv.as_pinned_arraybuffer(ctx) { - args.buffer = StringOrBuffer::Buffer(Buffer { - buffer: pinned, - owns_buffer: false, - pinned: true, - }); + if pinned.resizable && !pinned.shared { + // pin() blocks transfer(), not resize(); a shrink + // decommits pages the threadpool hands to write(2) and + // returns EFAULT. Snapshot the call-time bytes instead. + let view = pinned.byte_slice(); + let off = (args.offset as usize).min(view.len()); + let len = (args.length as usize).min(view.len() - off); + let owned = view[off..off + len].to_vec(); + pinned.unpin(); + ctx.vm().report_extra_memory(owned.len()); + args.offset = 0; + args.length = owned.len() as u64; + args.buffer = + StringOrBuffer::EncodedSlice(ZigStringSlice::init_owned(owned)); + } else { + args.buffer = StringOrBuffer::Buffer(Buffer { + buffer: pinned, + owns_buffer: false, + pinned: true, + }); + } } } Ok(args) diff --git a/src/runtime/node/types.rs b/src/runtime/node/types.rs index 475ef78c71b9..79937d6aee1d 100644 --- a/src/runtime/node/types.rs +++ b/src/runtime/node/types.rs @@ -1441,6 +1441,10 @@ pub struct VectorArrayBuffer { /// The collected elements, in order. Rooted (and their backing stores /// pinned) for the lifetime of an async operation; see [`Self::release`]. pub views: Vec, + /// Owned snapshots of any resizable-backed inputs; the matching + /// `buffers[i]` points in here. Populated by + /// [`Self::snapshot_resizable_inputs`]; freed by `Drop`. + owned: Vec>, pinned: bool, } @@ -1461,6 +1465,33 @@ impl VectorArrayBuffer { view.unprotect(); } } + + /// For async writev: copy any element backed by a resizable non-shared + /// ArrayBuffer into owned storage and repoint its iovec. `pin()` blocks + /// `transfer()` but not `ArrayBuffer.prototype.resize`; a shrink decommits + /// pages the threadpool hands to `pwritev(2)`, which then returns EFAULT. + /// Fixed-length and growable-shared backings stay zero-copy. + pub fn snapshot_resizable_inputs(&mut self, global: &JSGlobalObject) { + if !self.pinned { + return; + } + debug_assert_eq!(self.views.len(), self.buffers.len()); + let mut extra: usize = 0; + for (i, view) in self.views.iter().enumerate() { + let Some(buf) = view.as_array_buffer(global) else { + continue; + }; + if buf.resizable && !buf.shared { + let mut owned: Box<[u8]> = Box::from(buf.byte_slice()); + extra += owned.len(); + self.buffers[i] = bun_sys::platform_iovec_create(&mut owned[..]); + self.owned.push(owned); + } + } + if extra > 0 { + global.vm().report_extra_memory(extra); + } + } } unsafe extern "C" { @@ -1517,6 +1548,7 @@ impl VectorArrayBuffer { value: val, buffers: Vec::new(), views: Vec::new(), + owned: Vec::new(), pinned: false, }; bun_jsc::validation_scope!(scope, global_object); diff --git a/test/js/node/fs/fs.test.ts b/test/js/node/fs/fs.test.ts index b701f181be71..8bbbee34f6e6 100644 --- a/test/js/node/fs/fs.test.ts +++ b/test/js/node/fs/fs.test.ts @@ -5512,6 +5512,111 @@ it("fs.promises.writeFile keeps a buffer path argument attached while options ar expect(readFileSync(file, "utf8")).toBe("hello world"); }); +// pin() blocks transfer(), not ArrayBuffer.prototype.resize: a shrink decommits +// pages the threadpool hands to write(2)/pwritev(2). Async write/writev now +// snapshot resizable non-shared inputs at call time. +describe.concurrent.each(["write", "writev"] as const)( + "async fs.%s snapshots a resizable ArrayBuffer input at call time", + op => { + const fixture = /* js */ ` + import fs from "node:fs"; + import { open, readFile, writeFile } from "node:fs/promises"; + import { join } from "node:path"; + + const dir = process.env.FIXTURE_DIR; + const SIZE = 256 * 1024; + const heavy = join(dir, "heavy"); + await writeFile(heavy, Buffer.alloc(2 * 1024 * 1024)); + const saturate = () => + Promise.all(Array.from({ length: 32 }, () => fs.promises.readFile(heavy).catch(() => {}))); + + const enqueue = ${ + op === "write" + ? `(fd, ab) => new Promise(r => fs.write(fd, new Uint8Array(ab), 0, SIZE, 0, (e, n) => r({ e, n })))` + : `(fd, ab) => new Promise(r => fs.writev(fd, [new Uint8Array(ab, 0, SIZE / 2), new Uint8Array(ab, SIZE / 2)], 0, (e, n) => r({ e, n })))` + }; + + const out = join(dir, "out"); + const failures = []; + for (let i = 0; i < 6; i++) { + const fh = await open(out, "w"); + const ab = new ArrayBuffer(SIZE, { maxByteLength: SIZE * 2 }); + new Uint8Array(ab).fill(0x41); + const blockers = saturate(); + const p = enqueue(fh.fd, ab); + // Shrink then regrow: zeroes the bytes without leaving PROT_NONE pages, + // so a write that races the resize succeeds with wrong content instead + // of EFAULT. The snapshot taken at call time must write the 0x41 bytes. + ab.resize(0); + ab.resize(SIZE); + const { e, n } = await p; + await blockers; + await fh.close(); + if (e) { failures.push(e.code); continue; } + if (n !== SIZE) { failures.push("short=" + n); continue; } + const content = await readFile(out); + if (content.indexOf(0) !== -1) failures.push("zeroed"); + } + if (failures.length) { + console.error(JSON.stringify(failures)); + process.exit(1); + } + console.log("ok"); + `; + + it("writes the call-time bytes", async () => { + using dir = tempDir(`fs-${op}-resizable-ab`, {}); + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", fixture], + env: { ...bunEnv, FIXTURE_DIR: String(dir) }, + 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: "ok", + stderr: expect.any(String), + exitCode: 0, + }); + }); + + if (op === "write") { + it("snapshots only the [offset, offset+length) window", async () => { + using dir = tempDir("fs-write-resizable-offset", {}); + const out = join(String(dir), "out"); + const fh = await fs.promises.open(out, "w"); + const ab = new ArrayBuffer(64, { maxByteLength: 128 }); + const view = new Uint8Array(ab); + view.fill(0x42); + view.fill(0x41, 16, 48); + const p = new Promise<{ e: any; n: number }>(r => fs.write(fh.fd, view, 16, 32, 0, (e, n) => r({ e, n }))); + ab.resize(0); + const { e, n } = await p; + await fh.close(); + expect({ e, n, out: readFileSync(out) }).toEqual({ e: null, n: 32, out: Buffer.alloc(32, 0x41) }); + }); + } + + it("accepts a growable SharedArrayBuffer input", async () => { + using dir = tempDir(`fs-${op}-growable-sab`, {}); + const out = join(String(dir), "out"); + const fh = await fs.promises.open(out, "w"); + const sab = new SharedArrayBuffer(4096, { maxByteLength: 8192 }); + new Uint8Array(sab).fill(0x41); + const { e, n } = + op === "write" + ? await new Promise<{ e: any; n: number }>(r => + fs.write(fh.fd, new Uint8Array(sab), 0, 4096, 0, (e, n) => r({ e, n })), + ) + : await new Promise<{ e: any; n: number }>(r => + fs.writev(fh.fd, [new Uint8Array(sab, 0, 2048), new Uint8Array(sab, 2048)], 0, (e, n) => r({ e, n })), + ); + await fh.close(); + expect({ e, n, out: readFileSync(out) }).toEqual({ e: null, n: 4096, out: Buffer.alloc(4096, 0x41) }); + }); + }, +); + describe("fs.close on stdio descriptors", () => { it.skipIf(isWindows)("closeSync(2) actually closes fd 2 and allows redirect", async () => { using dir = tempDir("fs-close-stdio", {