node:fs: snapshot resizable ArrayBuffer inputs for async write/writev

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<Self> {
+            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<BlobSizeType>,
@@ -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)

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<JSValue>,
+    /// 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<Box<[u8]>>,
     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);

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", {