Skip to content

shell: close the stdio pipes of a subprocess whose stdin writer failed to start #38367

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
robobun wants to merge 1 commit into
base: main
Choose a base branch
from
+162 −16
Open

shell: close the stdio pipes of a subprocess whose stdin writer failed to start #38367

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
41 changes: 34 additions & 7 deletions src/runtime/shell/subproc.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -451,9 +451,10 @@ impl ShellSubprocess {
}
}

/// Tear down a subprocess whose stdio start() failed. Marks pending pipe readers as
/// errored so PipeReader.deinit's done-assert passes, drops the exit handler so a
/// later onProcessExit doesn't touch the freed Subprocess, then deinits.
/// Tear down a subprocess whose stdio start() failed. Releases the buffer-stdin
/// writer, marks pending pipe readers as errored so PipeReader.deinit's done-assert
/// passes (a reader that never started closes its pipe fd there), drops the exit
/// handler so a later onProcessExit doesn't touch the freed Subprocess, then deinits.
///
/// Windows: PipeReader.deinit asserts the libuv source is closed. Whether the source
/// is uv-initialized depends on how far startWithCurrentPipe got, so a blind close or
Expand All @@ -467,6 +468,21 @@ impl ShellSubprocess {
}
#[cfg(not(windows))]
{
// Release the slot's `create()` ref ourselves: a writer whose
// `start()` failed never reaches `on_close_io`, and one that did
// start (a sibling failed) must not reach it after we free `this`.
// `close()` re-enters `on_close_io` (and releases `start()`'s ref
// if it was taken), so the slot is emptied first, through `this`
// rather than a `Box` the re-entry would alias.
// SAFETY: `this` is the live subprocess from `spawn`; the borrow of
// the slot ends with the `replace`, before `close()` re-enters.
let stdin = unsafe { core::mem::replace(&mut (*this).stdin, Writable::Ignore) };
if let Writable::Buffer(buffer) = stdin {
// SAFETY: single-threaded; the slot held the only handle to the
// writer, so no other borrow of it is live.
unsafe { buffer_mut(&buffer) }.close();
buffer.deref();
}
// SAFETY: `this` was created via `heap::alloc` in `spawn` and is
// uniquely owned here; reclaim and tear down.
let mut subproc = unsafe { bun_core::heap::take(this) };
Expand Down Expand Up @@ -1004,6 +1020,9 @@ pub enum WritableInitError {
pub enum Writable {
Pipe(FileSinkPtr),
Fd(Fd),
/// Holds `create()`'s ref (`RefPtr` has no `Drop`). Released by
/// `ShellSubprocess::on_close_io` once the writer closes, or by
/// `abort_after_failed_start` when the spawn is abandoned before then.
Buffer(RefPtr<StaticPipeWriter>),
Memfd(Fd),
Inherit,
Expand Down Expand Up @@ -1199,9 +1218,9 @@ impl Writable {
Writable::Buffer(buffer) => {
// SAFETY: single-threaded; temporary `&mut` for the call only.
unsafe { buffer_mut(buffer) }.update_ref(false);
// Intentionally does NOT reassign `*self` — the variant tag is
// left as `Writable::Buffer`. RefPtr's Drop (on
// Subprocess teardown) handles the final deref.
// A writer still here is in flight (see the variant's doc for
// who releases the ref); closing it would re-enter this
// subprocess mid-drop, so it is left alone.
}
Writable::Memfd(fd) => {
fd.close();
Expand Down Expand Up @@ -1539,6 +1558,9 @@ pub struct PipeReader {
pub(crate) process: Option<*mut ShellSubprocess>,
pub(crate) event_loop: EventLoopHandle,
pub(crate) state: PipeReaderState,
/// POSIX: our end of the child's stdio pipe, owned here until `start()`
/// hands it to `reader`. Still `Some` on a reader that never started
/// (the spawn was aborted first), in which case `drop` closes it.
#[cfg_attr(windows, allow(dead_code))]
pub(crate) stdio_result: StdioResult,
pub(crate) out_type: OutKind,
Expand Down Expand Up @@ -1868,7 +1890,7 @@ impl PipeReader {
}

#[cfg(not(windows))]
match self.reader.start(self.stdio_result.unwrap(), true) {
match self.reader.start(self.stdio_result.take().unwrap(), true) {
bun_sys::Result::Err(err) => bun_sys::Result::Err(err),
bun_sys::Result::Ok(()) => {
// `reader.start` reports a poll-registration failure through
Expand Down Expand Up @@ -2195,6 +2217,11 @@ impl Drop for PipeReader {
{
debug_assert!(self.reader.is_done() || matches!(self.state, PipeReaderState::Err(_)));
}
#[cfg(not(windows))]
if let Some(fd) = self.stdio_result.take() {
// Never started, so `reader` never took the fd over.
fd.close();
}

#[cfg(windows)]
{
Expand Down
137 changes: 128 additions & 9 deletions test/js/bun/shell/shell-pipe-read-fault.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,16 @@ const cc = Bun.which("cc") || Bun.which("gcc") || Bun.which("clang");
// recv guarantees the streaming inner loop's mid-read
// flush (`head_start` past the half-buffer cutoff)
// fires in a single poll wake.
// SHELL_FAIL_EPOLL_WRITABLE=1 every epoll_ctl ADD asking for EPOLLOUT on a
// socket this process made with socketpair() (the
// child stdio pipes) fails with ENOMEM. Only writers
// ask for EPOLLOUT, so this fails the child's
// buffer-stdin writer's start() (the only stdio
// start() whose failure is returned to the spawn
// instead of reported through a callback) and
// nothing else: the stdout/stderr readers poll for
// EPOLLIN, and the shell's writers to the fixture's
// own stdio sit on inherited fds.
// FilePoll registers the socketpair through the raw syscall(SYS_epoll_ctl,
// ...) wrapper, not the libc epoll_ctl symbol, so the epoll modes interpose
// syscall(2).
Expand All @@ -58,16 +68,19 @@ const SHIM_C = /* c */ `
static ssize_t (*real_recv)(int, void *, size_t, int);
static long (*real_syscall)(long, long, long, long, long, long, long);
static int (*real_close)(int);
static int (*real_socketpair)(int, int, int, int *);
static int fail_recv = -1;
static int fail_epoll = -1;
static int fail_epoll_after = -1;
static int recv_one_chunk = -1;
static int recv_eagain_first = -1;
static int fail_epoll_from = -1; /* 0 = off, N >= 1 = 1-based index of the first failing call */
static int recv_bulk = -1; /* 0 = off, N >= 1 = number of fabricated full-buffer recvs */
static int fail_epoll_writable = -1;
static unsigned char recv_count[MAX_FD];
static unsigned char epoll_count[MAX_FD];
static unsigned char bulk_count[MAX_FD];
static unsigned char own_socketpair[MAX_FD];

static void init_modes(void) {
if (fail_recv < 0) fail_recv = getenv("SHELL_FAIL_RECV") != NULL;
Expand All @@ -83,6 +96,7 @@ static void init_modes(void) {
const char *s = getenv("SHELL_RECV_BULK");
recv_bulk = s ? atoi(s) : 0;
}
if (fail_epoll_writable < 0) fail_epoll_writable = getenv("SHELL_FAIL_EPOLL_WRITABLE") != NULL;
}

static int is_unix_sock(int fd) {
Expand All @@ -100,6 +114,28 @@ static int is_pipe_like(int fd) {
return is_unix_sock(fd);
}

// Reset the per-fd state on close so a recycled fd number starts fresh. Bun
// closes through syscall(SYS_close), so both close entry points land here.
static void forget_fd(int fd) {
if (fd >= 0 && fd < MAX_FD) {
recv_count[fd] = 0;
epoll_count[fd] = 0;
bulk_count[fd] = 0;
own_socketpair[fd] = 0;
}
}

int socketpair(int domain, int type, int protocol, int sv[2]) {
if (!real_socketpair) real_socketpair = (int (*)(int, int, int, int *))dlsym(RTLD_NEXT, "socketpair");
int rc = real_socketpair(domain, type, protocol, sv);
if (rc == 0) {
for (int i = 0; i < 2; i++) {
if (sv[i] >= 0 && sv[i] < MAX_FD) own_socketpair[sv[i]] = 1;
}
}
return rc;
}

ssize_t recv(int fd, void *buf, size_t len, int flags) {
if (!real_recv) {
real_recv = (ssize_t (*)(int, void *, size_t, int))dlsym(RTLD_NEXT, "recv");
Expand Down Expand Up @@ -145,9 +181,16 @@ long syscall(long number, ...) {
real_syscall = (long (*)(long, long, long, long, long, long, long))dlsym(RTLD_NEXT, "syscall");
init_modes();
}
if (number == SYS_close) forget_fd((int)a);
if (number == SYS_epoll_ctl) {
int op = (int)b;
int target = (int)c;
const struct epoll_event *event = (const struct epoll_event *)d;
if (fail_epoll_writable && op == EPOLL_CTL_ADD && target >= 0 && target < MAX_FD && own_socketpair[target] &&
event && (event->events & EPOLLOUT)) {
errno = ENOMEM;
return -1;
}
if (op == EPOLL_CTL_ADD || op == EPOLL_CTL_MOD) {
if (fail_epoll && is_pipe_like(target)) {
errno = ENOMEM;
Expand All @@ -170,14 +213,9 @@ long syscall(long number, ...) {
return real_syscall(number, a, b, c, d, e, f);
}

// Reset the per-fd counters on close so a recycled fd number starts fresh.
int close(int fd) {
if (!real_close) real_close = (int (*)(int))dlsym(RTLD_NEXT, "close");
if (fd >= 0 && fd < MAX_FD) {
recv_count[fd] = 0;
epoll_count[fd] = 0;
bulk_count[fd] = 0;
}
forget_fd(fd);
return real_close(fd);
}
`;
Expand Down Expand Up @@ -234,6 +272,58 @@ const r = await $\`sh -c 'printf AAAA; exec sleep 5' 2> /dev/null\`.nothrow();
console.log(JSON.stringify({ exitCode: r.exitCode }));
`;

// For SHELL_FAIL_EPOLL_WRITABLE: `< ${blob}` / `< ${bytes}` give the child a
// buffer-stdin writer, which spawn_async starts before the stdout/stderr
// readers. Its failed start() aborts the spawn; the command reports the errno
// and everything the spawn had set up must be closed again: the two pipe ends
// the never-started readers were holding and the writer's own stdin pipe end.
// The fds are listed with their inode so a reused number cannot hide a leak.
const STDIN_START_FAULT_FIXTURE = /* js */ `
import { readdirSync, readlinkSync } from "node:fs";
import { $ } from "bun";

function openFds() {
const fds = [];
for (const fd of readdirSync("/proc/self/fd")) {
// The readdir handle itself is already closed again by the time we get here.
try {
fds.push(fd + ":" + readlinkSync("/proc/self/fd/" + fd));
} catch {}
}
return fds;
}

// Whatever a spawn sets up once (work pool, closer threads, the writers for
// this process's own stdio) must not count against the failed commands, so
// take the baseline after a successful command of each kind below.
await $\`head -c 0 /dev/zero\`;
await $\`head -c 0 /dev/zero\`.quiet();
const before = openFds();

const results = [];
async function run(cmd) {
const r = await cmd.nothrow();
results.push({ exitCode: r.exitCode, stderr: r.stderr.toString() });
}
// .quiet() gives the readers plain capture pipes; without it they also tee
// into this process's stdio, the shell's other reader setup.
await run($\`cat < \${new Blob(["blob stdin"])}\`.quiet());
await run($\`cat < \${new TextEncoder().encode("buffer stdin")}\`.quiet());
await run($\`cat < \${new Blob(["blob stdin"])}\`);

// Pipe ends held by a poll are closed on the work pool, so wait for the set to
// drain; a real leak never drains and the deadline (well inside the test's
// own timeout, so the list below is what gets reported) shows what is left.
let leaked;
const deadline = Date.now() + 2000;
while (true) {
leaked = openFds().filter(fd => !before.includes(fd));
if (leaked.length === 0 || Date.now() > deadline) break;
await Bun.sleep(10);
}
console.log(JSON.stringify({ results, leaked }));
`;

let shimPath: string;
let dir: ReturnType<typeof tempDir> | undefined;

Expand All @@ -246,6 +336,7 @@ beforeAll(async () => {
"quiet-chunk.js": QUIET_CHUNK_FIXTURE,
"poll-chunk.js": POLL_CHUNK_FIXTURE,
"tee-chunk.js": TEE_CHUNK_FIXTURE,
"stdin-start-fault.js": STDIN_START_FAULT_FIXTURE,
});
shimPath = join(String(dir), "shim.so");
await using ccProc = Bun.spawn({
Expand Down Expand Up @@ -273,6 +364,7 @@ const MODES = [
"SHELL_FAIL_EPOLL_AFTER",
"SHELL_RECV_ONE_CHUNK",
"SHELL_RECV_EAGAIN_FIRST",
"SHELL_FAIL_EPOLL_WRITABLE",
] as const;
// Integer-valued fault knobs; cleared alongside MODES and set through `extraEnv`.
const VALUE_MODES = ["SHELL_FAIL_EPOLL_FROM", "SHELL_RECV_BULK"] as const;
Expand All @@ -295,7 +387,10 @@ function shimEnv(
return env;
}

async function expectShellFault(
// Runs a fixture under the shim and returns its last stdout line (the JSON it
// prints) along with stderr and the exit code, so one combined assertion can
// surface a crash's stderr and exit code in the diff.
async function runShellFaultFixture(
script: string,
modes: (typeof MODES)[number][],
extraEnv: Partial<Record<(typeof VALUE_MODES)[number], string>> = {},
Expand All @@ -319,8 +414,15 @@ async function expectShellFault(
} catch {
parsed = line;
}
// One combined assertion so a crash surfaces stderr and the exit code in the diff.
expect({ parsed, stderr, exitCode }).toEqual({
return { parsed, stderr, exitCode };
}

async function expectShellFault(
script: string,
modes: (typeof MODES)[number][],
extraEnv: Partial<Record<(typeof VALUE_MODES)[number], string>> = {},
) {
expect(await runShellFaultFixture(script, modes, extraEnv)).toEqual({
parsed: { exitCode: ENOMEM },
stderr: expect.any(String),
exitCode: 0,
Expand Down Expand Up @@ -355,6 +457,23 @@ test.concurrent.skipIf(!isLinux || !cc)(
},
);

// Regression: abort_after_failed_start dropped the stdout/stderr PipeReaders
// without closing the pipe ends they had not handed to their readers yet, and
// never released the stdin writer, so every failed command leaked three
// sockets (the writer's fd went with it).
test.concurrent.skipIf(!isLinux || !cc)(
"shell closes every stdio pipe of a command whose buffer stdin writer failed to start",
async () => {
const failed = { exitCode: 1, stderr: expect.stringContaining("Cannot allocate memory") };
expect(await runShellFaultFixture("stdin-start-fault.js", ["SHELL_FAIL_EPOLL_WRITABLE"])).toEqual({
parsed: { results: [failed, failed, failed], leaked: [] },
// The un-quieted command also tees its error message here.
stderr: expect.stringContaining("Cannot allocate memory"),
exitCode: 0,
});
},
);

// Regression: the re-registration failing AFTER the eager read made progress
// closes the stream and detaches the PipeReader (process = None), and the
// reader then still delivers the drained chunk and retries the poll, so the
Expand Down
Loading