Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
89 changes: 75 additions & 14 deletions src/spawn_sys/spawn_process.rs
Original file line number Diff line number Diff line change
Expand Up @@ -596,6 +596,19 @@ struct PosixSpawnFdGuard {
on_error: bool,
}

#[cfg(unix)]
impl PosixSpawnFdGuard {
/// File actions run in slot order, so an earlier slot's `close`/`dup2` would hit a `dup2` source numbered at or below `max_slot`; dup it above every slot first, like libuv.
fn source_above_slots(&mut self, max_slot: i32, src: Fd) -> bun_sys::Result<Fd> {
if src.native() > max_slot {
return Ok(src);
}
let moved = bun_sys::dup_at_least(src, max_slot + 1)?;
self.to_close_at_end.push(moved);
Ok(moved)
}
}

#[cfg(unix)]
impl Drop for PosixSpawnFdGuard {
fn drop(&mut self) {
Expand Down Expand Up @@ -716,12 +729,37 @@ pub unsafe fn spawn_process_posix(
let _ = attr.set(flags as _);
let _ = attr.reset_signals();

// Highest child fd that a file action targets: stderr, or the last extra slot.
let Ok(max_slot) = FdT::try_from(2 + options.extra_fds.len()) else {
return Ok(Err(bun_sys::Error::from_code(
bun_sys::E::EMFILE,
bun_sys::Tag::posix_spawn,
)));
};

if let Some(ipc) = options.ipc {
actions.inherit(ipc)?;
spawned.ipc = Some(ipc);
}

let stdio_options: [&PosixStdio; 3] = [&options.stdin, &options.stdout, &options.stderr];
// Probed before this function creates any fd: a socketpair end can land on a closed slot number and make that slot look open.
let inherits_closed_fd = |stdio: &PosixStdio, slot: usize| {
matches!(stdio, PosixStdio::Inherit)
&& bun_sys::get_fcntl_flags(Fd::from_native(slot as FdT)).is_err()
};
let closed_stdio: [bool; 3] = core::array::from_fn(|i| inherits_closed_fd(stdio_options[i], i));
Comment thread
robobun marked this conversation as resolved.
Resolved
if options
.extra_fds
.iter()
.enumerate()
.any(|(i, stdio)| inherits_closed_fd(stdio, 3 + i))
{
return Ok(Err(bun_sys::Error::from_code(
bun_sys::E::EBADF,
bun_sys::Tag::posix_spawn,
)));
}
// Reshaped for borrowck: we
// index spawned.{stdin,stdout,stderr} via a helper closure.
let mut dup_stdout_to_stderr: bool = false;
Expand Down Expand Up @@ -756,9 +794,11 @@ pub unsafe fn spawn_process_posix(
}
PosixStdio::Inherit => {
// A closed slot would inherit whatever fd is created later at that number (e.g. the ipc socketpair); libuv gives it /dev/null.
if bun_sys::get_fcntl_flags(fileno).is_err() {
if closed_stdio[i] {
actions.open_z(fileno, c"/dev/null", flag | bun_sys::O::CREAT as u32, 0o664)?;
Comment thread
robobun marked this conversation as resolved.
Resolved
} else {
// O_NONBLOCK is on the shared open file description: left set (by `process.stdout` on a pipe), the child's plain write(2) fails with EAGAIN. libuv clears it too.
let _ = bun_sys::update_nonblocking(fileno, false);
actions.inherit(fileno)?;
}
}
Expand Down Expand Up @@ -788,7 +828,11 @@ pub unsafe fn spawn_process_posix(

cleanup.to_close_on_error.push(fd);
cleanup.to_set_cloexec.push(fd);
actions.dup2(fd, fileno)?;
let src = match cleanup.source_above_slots(max_slot, fd) {
Ok(src) => src,
Err(e) => return Ok(Err(e)),
};
actions.dup2(src, fileno)?;
set_spawned_stdio(&mut spawned, i, fd);
spawned.memfds[i] = true;
continue 'stdio;
Expand Down Expand Up @@ -869,15 +913,24 @@ pub unsafe fn spawn_process_posix(
}
}

actions.dup2(fds[1], fileno)?;
if fds[1] != fileno {
let src = match cleanup.source_above_slots(max_slot, fds[1]) {
Ok(src) => src,
Err(e) => return Ok(Err(e)),
};
actions.dup2(src, fileno)?;
if src == fds[1] {
actions.close(fds[1])?;
}
Comment thread
robobun marked this conversation as resolved.
Resolved

set_spawned_stdio(&mut spawned, i, fds[0]);
}
PosixStdio::Pipe(fd) => {
actions.dup2(*fd, fileno)?;
let _ = bun_sys::update_nonblocking(*fd, false);
let src = match cleanup.source_above_slots(max_slot, *fd) {
Ok(src) => src,
Err(e) => return Ok(Err(e)),
};
actions.dup2(src, fileno)?;
set_spawned_stdio(&mut spawned, i, *fd);
}
PosixStdio::SocketFd => {
Expand Down Expand Up @@ -919,24 +972,28 @@ pub unsafe fn spawn_process_posix(
extra_fds.push(ExtraPipe::Unavailable);
}
PosixStdio::Ipc | PosixStdio::Buffer | PosixStdio::SocketFd => {
let is_ipc = matches!(ipc, PosixStdio::Ipc);
// Only the parent's end goes nonblocking: a child that is not bun or node does a plain write(2) on its end, and libuv hands out a blocking one too.
let fds: [Fd; 2] =
match bun_sys::socketpair(libc::AF_UNIX, libc::SOCK_STREAM, 0, is_ipc) {
match bun_sys::socketpair(libc::AF_UNIX, libc::SOCK_STREAM, 0, false) {
Ok(p) => p,
Err(e) => return Ok(Err(e)),
};

if !options.sync && !is_ipc {
cleanup.to_close_at_end.push(fds[1]);
cleanup.to_close_on_error.push(fds[0]);

if !options.sync {
if let Err(e) = bun_sys::set_nonblocking(fds[0]) {
return Ok(Err(e));
}
}

cleanup.to_close_at_end.push(fds[1]);
cleanup.to_close_on_error.push(fds[0]);

actions.dup2(fds[1], fileno)?;
if fds[1] != fileno {
let src = match cleanup.source_above_slots(max_slot, fds[1]) {
Ok(src) => src,
Err(e) => return Ok(Err(e)),
};
actions.dup2(src, fileno)?;
if src == fds[1] {
actions.close(fds[1])?;
}
// SocketFd: push as OwnedFd here so every error path between
Expand All @@ -949,7 +1006,11 @@ pub unsafe fn spawn_process_posix(
extra_fds.push(ExtraPipe::OwnedFd(fds[0]));
}
PosixStdio::Pipe(fd) => {
actions.dup2(*fd, fileno)?;
let src = match cleanup.source_above_slots(max_slot, *fd) {
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
Ok(src) => src,
Err(e) => return Ok(Err(e)),
};
actions.dup2(src, fileno)?;
// The fd was supplied by the caller (a number in the stdio array) and is
// not owned by us. Record it so `stdio[N]` returns the caller's fd, but
// mark it unowned so finalizeStreams leaves it open.
Expand Down
28 changes: 28 additions & 0 deletions test/js/bun/spawn/fixtures/fd-nonblock-probe.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
// Prints "<fd>:blocking" or "<fd>:nonblocking" for each fd number in argv.
Comment thread
robobun marked this conversation as resolved.
Resolved
// The flag is read before anything writes, so this process cannot change it.
// Must not touch process.stdout/process.stderr/process.send: those set
// O_NONBLOCK on their own fds.
import { dlopen } from "bun:ffi";
import { existsSync, readFileSync } from "node:fs";

let isNonblocking;
if (existsSync("/proc/self/fdinfo")) {
isNonblocking = fd => {
const flags = readFileSync(`/proc/self/fdinfo/${fd}`, "utf8").match(/^flags:\s*([0-7]+)/m)[1];
return (parseInt(flags, 8) & 0o4000) !== 0;
};
} else {
// F_GETFL is 3 on every platform here. O_NONBLOCK is 0o4000 on Linux and 4 on macOS and the BSDs.
const libc = { darwin: "libSystem.B.dylib", freebsd: "libc.so.7" }[process.platform] ?? "libc.so.6";
const O_NONBLOCK = process.platform === "linux" ? 0o4000 : 4;
const { fcntl } = dlopen(libc, {
fcntl: { args: ["i32", "i32", "i32"], returns: "i32" },
}).symbols;
isNonblocking = fd => (fcntl(fd, 3, 0) & O_NONBLOCK) !== 0;
}

const out = process.argv
.slice(2)
.map(arg => `${arg}:${isNonblocking(Number(arg)) ? "nonblocking" : "blocking"}`)
.join(" ");
console.log(out);
203 changes: 203 additions & 0 deletions test/js/bun/spawn/spawn.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1841,3 +1841,206 @@ it.if(parentThp() === "1")("spawned children keep the system THP policy", async
expect(thpEnabled(readFileSync("/proc/self/status", "utf8"))).toBe("1");
expect(exitCode).toBe(0);
});

// Runs a script in a fresh bun process, so its fd numbers are low and known and a blocked spawnSync cannot block this runner.
async function runInFreshProcess(script: string) {
await using proc = spawn({
cmd: [bunExe(), "-e", script],
env: bunEnv,
stdio: ["ignore", "pipe", "pipe"],
});
const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
return { stdout, stderr, exitCode };
}

// O_NONBLOCK lives on the open file description, so a flag bun sets on its
// own stdout or on a socketpair end reaches any child that shares it. A child
// that is not bun or node then fails a plain write(2) with EAGAIN.
describe.skipIf(!isPosix)("file status flags handed to the child", () => {
const probe = join(import.meta.dir, "fixtures", "fd-nonblock-probe.js");

it.concurrent("stdout and stderr inherited after the parent used process.stdout/stderr are blocking", async () => {
await using proc = spawn({
cmd: [
bunExe(),
"-e",
`process.stdout.write("out\\n");
process.stderr.write("err\\n");
const r = Bun.spawnSync([process.execPath, ${JSON.stringify(probe)}, "1", "2"], { stdout: "inherit", stderr: "inherit" });
process.exit(r.exitCode);`,
],
env: bunEnv,
stdout: "pipe",
stderr: "pipe",
});
const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
expect(stdout).toBe("out\n1:blocking 2:blocking\n");
expect(stderr).toBe("err\n");
expect(exitCode).toBe(0);
});

it.concurrent("the ipc fd is blocking", async () => {
await using proc = spawn({
cmd: [bunExe(), probe, "3"],
env: bunEnv,
stdout: "pipe",
stderr: "pipe",
ipc() {},
});
const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
expect(stdout).toBe("3:blocking\n");
expect(stderr).toBe("");
expect(exitCode).toBe(0);
});

it.concurrent("a plain tool writing more than the pipe holds succeeds on inherited stdout", async () => {
// The user-visible shape of the test above: with fd 1 nonblocking, head
// gets EAGAIN as soon as it outruns this reader and exits 1 with a short
// count. (If the reader keeps up, head never sees a full pipe, so the
// flag probe above is the deterministic check.)
await using proc = spawn({
cmd: [
bunExe(),
"-e",
`process.stdout.write("start\\n");
const r = Bun.spawnSync(["head", "-c", "1000000", "/dev/zero"], { stdout: "inherit", stderr: "inherit" });
process.stderr.write("rc=" + r.exitCode + "\\n");`,
],
env: bunEnv,
stdout: "pipe",
stderr: "pipe",
});
const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.bytes(), proc.stderr.text(), proc.exited]);
expect(stderr).toBe("rc=0\n");
expect(stdout.byteLength).toBe("start\n".length + 1000000);
expect(exitCode).toBe(0);
});
});

describe.skipIf(!isPosix)("stdio source fds that are also slot numbers", () => {
// File actions run in slot order in the child. The source fd of a later
// slot must survive the actions of the slots before it (the close of an
// "ignore" slot, the dup2 onto another slot) even when its fd number is one
// of those slots. Each case runs in a fresh process so the source fds are
// low, known numbers.
it.concurrent("a pipe after many ignore slots", async () => {
const slots = 60;
const result = await runInFreshProcess(`
const fs = require("fs");
const proc = Bun.spawn({
cmd: [process.execPath, "-e", "require('fs').writeSync(${slots}, 'hi')"],
stdio: ["ignore", "inherit", "inherit", ...Array(${slots - 3}).fill("ignore"), "pipe"],
});
const fd = proc.stdio[${slots}];
const exitCode = await proc.exited;
const buf = Buffer.alloc(16);
const n = fs.readSync(fd, buf);
console.log(JSON.stringify({ sourceBelowSlot: fd < ${slots}, exitCode, data: buf.toString("utf8", 0, n) }));
`);
expect(result.stderr).toBe("");
expect(JSON.parse(result.stdout)).toEqual({ sourceBelowSlot: true, exitCode: 0, data: "hi" });
expect(result.exitCode).toBe(0);
});

it.concurrent("a caller fd after many ignore slots", async () => {
using dir = tempDir("spawn-high-slot", {});
const slots = 60;
const file = join(String(dir), "out.txt");
const result = await runInFreshProcess(`
const fs = require("fs");
const fd = fs.openSync(${JSON.stringify(file)}, "w");
const proc = Bun.spawn({
cmd: [process.execPath, "-e", "require('fs').writeSync(${slots}, 'hi')"],
stdio: ["ignore", "inherit", "inherit", ...Array(${slots - 3}).fill("ignore"), fd],
});
console.log(JSON.stringify({ sourceBelowSlot: fd < ${slots}, exitCode: await proc.exited }));
`);
expect(result.stderr).toBe("");
expect(JSON.parse(result.stdout)).toEqual({ sourceBelowSlot: true, exitCode: 0 });
expect(readFileSync(file, "utf8")).toBe("hi");
expect(result.exitCode).toBe(0);
});

it.concurrent("fd 2 at stdout and fd 1 at stderr swap them", async () => {
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
const result = await runInFreshProcess(`
const proc = Bun.spawn({
cmd: [process.execPath, "-e", "require('fs').writeSync(1, 'to-fd-1,'); require('fs').writeSync(2, 'to-fd-2,')"],
stdio: ["ignore", 2, 1],
});
await proc.exited;
`);
expect({ stdout: result.stdout, stderr: result.stderr }).toEqual({ stdout: "to-fd-2,", stderr: "to-fd-1," });
expect(result.exitCode).toBe(0);
});

// The child reports which of its fds 1 and 2 is /dev/null. It writes the report to the other one.
const reportDevNull = (reportFd: 1 | 2) => `
const fs = require("fs");
const devNull = fs.statSync("/dev/null").rdev;
const isDevNull = fd => { const s = fs.fstatSync(fd); return s.isCharacterDevice() && s.rdev === devNull; };
fs.writeSync(${reportFd}, JSON.stringify({ stdoutIsDevNull: isDevNull(1), stderrIsDevNull: isDevNull(2) }));
`;

it.concurrent(
"a closed stdout slot that is inherited gets /dev/null, not the stdin pipe end that took fd 1",
async () => {
const result = await runInFreshProcess(`
require("fs").closeSync(1);
// stdin "pipe" makes a socketpair, and its child end lands on the free fd 1.
const proc = Bun.spawn({
cmd: [process.execPath, "-e", ${JSON.stringify(reportDevNull(2))}],
stdio: ["pipe", "inherit", "inherit"],
});
process.exitCode = await proc.exited;
`);
expect(JSON.parse(result.stderr)).toEqual({ stdoutIsDevNull: true, stderrIsDevNull: false });
expect(result.exitCode).toBe(0);
},
);

it.concurrent(
"a closed stderr slot that is inherited gets /dev/null, not the stdout pipe end that took fd 2",
async () => {
const result = await runInFreshProcess(`
require("fs").closeSync(2);
// stdout "pipe" makes a socketpair, and the parent's end lands on the free fd 2.
const proc = Bun.spawn({
cmd: [process.execPath, "-e", ${JSON.stringify(reportDevNull(1))}],
stdio: ["ignore", "pipe", "inherit"],
});
const [report, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
console.log(report);
process.exitCode = exitCode;
`);
expect(JSON.parse(result.stdout)).toEqual({ stdoutIsDevNull: false, stderrIsDevNull: true });
expect(result.exitCode).toBe(0);
},
);

it.concurrent(
"an extra inherit slot that is closed in the parent fails with EBADF, not with the pipe end that took its number",
async () => {
const result = await runInFreshProcess(`
const fs = require("fs");
const free = fs.openSync("/dev/null", "r"); // the lowest free fd number
fs.closeSync(free);
let outcome;
try {
// stdin "pipe" makes a socketpair, and its child end lands on that free number.
const proc = Bun.spawn({
cmd: ["true"],
stdio: ["pipe", "inherit", "inherit", ...Array(free - 3).fill("ignore"), "inherit"],
});
await proc.exited;
outcome = "spawned";
} catch (e) {
outcome = e.code;
}
console.log(JSON.stringify({ freeIsExtraSlot: free >= 3, outcome }));
`);
expect(result.stderr).toBe("");
expect(JSON.parse(result.stdout)).toEqual({ freeIsExtraSlot: true, outcome: "EBADF" });
expect(result.exitCode).toBe(0);
},
);
});
Loading
Loading