From 6ed950a654cb262ea8f9d245510b8da0b0355c53 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 24 Jul 2026 03:35:12 +0000 Subject: [PATCH 1/4] FileSink: settle the pending write()'s promise when close()'s flush fails FileSink::end() (the js_close path behind sink.close() on the FileSink prototype and the piped-ReadableStream controller's close()) had the same WriteResult::Err arm that #35344 fixed in end_from_js(): it set done=true, tore down the writer, and returned Err without touching the pending slot. A backpressured write()'s promise was left pending forever while close() threw EPIPE at its caller. Mirror end_from_js(): when a backpressured write()'s promise is outstanding, latch the error into it, schedule run_pending, and return Ok so js_close doesn't also throw. When nothing is pending the Err return (and close()'s throw) is unchanged. --- src/runtime/webcore/FileSink.rs | 14 +++++++++++ test/js/bun/util/filesink.test.ts | 39 +++++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+) diff --git a/src/runtime/webcore/FileSink.rs b/src/runtime/webcore/FileSink.rs index cbc45719711c..526322dc9fa0 100644 --- a/src/runtime/webcore/FileSink.rs +++ b/src/runtime/webcore/FileSink.rs @@ -1054,6 +1054,20 @@ impl FileSink { } WriteResult::Err(e) => { self.done.set(true); + if self.pending.get().state == streams::PendingState::Pending { + // A backpressured `write()` left its promise outstanding. + // `js_close` can't hand that promise back the way + // `end_from_js` does, but the error still goes to it + // exactly once: latch, tear down, schedule `run_pending`, + // return `Ok` so `close()` doesn't also throw. Latch before + // `writer.end()`; its `on_close` re-entry runs with the + // slot already holding the error. + self.pending + .with_mut(|p| p.result = streams::Writable::Err(e)); + self.writer.with_mut(|w| w.end()); + self.run_pending_later(); + return sys::Result::Ok(()); + } self.writer.with_mut(|w| w.end()); sys::Result::Err(e) } diff --git a/test/js/bun/util/filesink.test.ts b/test/js/bun/util/filesink.test.ts index 72da12167a4f..60df6af31d02 100644 --- a/test/js/bun/util/filesink.test.ts +++ b/test/js/bun/util/filesink.test.ts @@ -318,6 +318,45 @@ it.skipIf(!isPosix)( }, ); +// Sibling of the test above for sink.close() (js_close -> FileSink::end()). +// close()'s flush() sees EPIPE synchronously; before the fix end()'s Err arm +// set done=true, tore down the writer, and returned Err without touching the +// pending write()'s promise, so that promise was left pending forever. close() +// can't hand the promise back, but it latches the error into it and returns +// normally so the failure is reported once, via the promise. +it.skipIf(!isPosix)( + "close() after a backpressured write() with the reader gone rejects the write's promise with EPIPE", + async () => { + const [readFd, writeFd] = createSocketPair(); + let readFdOpen = true; + const sink = Bun.file(writeFd).writer(); + try { + const writePromise = sink.write(Buffer.alloc(4 * 1024 * 1024, 0x61)); + expect(writePromise).toBeInstanceOf(Promise); + + fs.closeSync(readFd); + readFdOpen = false; + + // close()'s flush hits EPIPE. The error is routed to writePromise; the + // close() call itself doesn't throw (no double reporting). + expect(sink.close()).toBeUndefined(); + + let caught: any; + try { + await writePromise; + } catch (e) { + caught = e; + } + expect(caught?.code).toBe("EPIPE"); + } finally { + try { + fs.closeSync(writeFd); + } catch {} + if (readFdOpen) fs.closeSync(readFd); + } + }, +); + // The deferred auto-flush microtask runs at the first microtask checkpoint // after write() backpressures. If its flush() hit EPIPE, it discarded the // error and then let `run_pending_later()` resolve the pending write() promise From dd00437301c647dc02377285b1d6042e385265ac Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 24 Jul 2026 04:38:16 +0000 Subject: [PATCH 2/4] Settle the pending write() in end()'s Done/Wrote arms too; move test to a subprocess The Err arm wasn't the whole bug: end()'s Done and Wrote arms (reached when close()'s flush drains the buffer synchronously because the reader pulled between write() and close()) call writer.end() without scheduling run_pending either, so the write()'s promise hangs the same way. writer.end() only re-enters on_close, which never touches the pending slot; flush() doesn't route through on_write. Schedule run_pending_later() in those arms too (pending.result already holds Owned(consumed) from to_result). The test now runs in a subprocess with detect_leaks=0 in its env: sink.close() on a Blob-created FileSink strands the JS wrapper's +1 ref on main (doClose nulls m_sinkPtr so ~JSFileSink skips finalize), which the x64-asan lane flags. That leak is pre-existing and tracked separately. --- src/runtime/webcore/FileSink.rs | 28 ++++---- test/js/bun/util/filesink.test.ts | 103 ++++++++++++++++++++---------- 2 files changed, 85 insertions(+), 46 deletions(-) diff --git a/src/runtime/webcore/FileSink.rs b/src/runtime/webcore/FileSink.rs index 526322dc9fa0..504818daa302 100644 --- a/src/runtime/webcore/FileSink.rs +++ b/src/runtime/webcore/FileSink.rs @@ -1044,24 +1044,29 @@ impl FileSink { return sys::Result::Ok(()); } + // A backpressured `write()` may have left its promise in `self.pending`; + // `writer.end()` only re-enters `on_close`, which never touches it, so + // every synchronous arm that tears the writer down here must settle it + // (mirrors `on_auto_flush` / `end_from_js`). `js_close` can't hand the + // promise back, so the outcome is delivered via `run_pending` and the + // call returns `Ok`. + let has_pending = self.pending.get().state == streams::PendingState::Pending; + // SAFETY(JsCell): `IOWriter::flush` is pure I/O; any callback re-entry // goes via the stored `*mut FileSink` backref, not this borrow. match self.writer.with_mut(|w| w.flush()) { - WriteResult::Done(written) => { + WriteResult::Done(written) | WriteResult::Wrote(written) => { self.written.set(self.written.get() + written as usize); // @truncate self.writer.with_mut(|w| w.end()); + if has_pending { + // `to_result` already seeded `Owned(consumed)`; just deliver it. + self.run_pending_later(); + } sys::Result::Ok(()) } WriteResult::Err(e) => { self.done.set(true); - if self.pending.get().state == streams::PendingState::Pending { - // A backpressured `write()` left its promise outstanding. - // `js_close` can't hand that promise back the way - // `end_from_js` does, but the error still goes to it - // exactly once: latch, tear down, schedule `run_pending`, - // return `Ok` so `close()` doesn't also throw. Latch before - // `writer.end()`; its `on_close` re-entry runs with the - // slot already holding the error. + if has_pending { self.pending .with_mut(|p| p.result = streams::Writable::Err(e)); self.writer.with_mut(|w| w.end()); @@ -1080,11 +1085,6 @@ impl FileSink { self.done.set(true); sys::Result::Ok(()) } - WriteResult::Wrote(written) => { - self.written.set(self.written.get() + written as usize); // @truncate - self.writer.with_mut(|w| w.end()); - sys::Result::Ok(()) - } } } diff --git a/test/js/bun/util/filesink.test.ts b/test/js/bun/util/filesink.test.ts index 60df6af31d02..6007af60e772 100644 --- a/test/js/bun/util/filesink.test.ts +++ b/test/js/bun/util/filesink.test.ts @@ -318,42 +318,81 @@ it.skipIf(!isPosix)( }, ); -// Sibling of the test above for sink.close() (js_close -> FileSink::end()). -// close()'s flush() sees EPIPE synchronously; before the fix end()'s Err arm -// set done=true, tore down the writer, and returned Err without touching the -// pending write()'s promise, so that promise was left pending forever. close() -// can't hand the promise back, but it latches the error into it and returns -// normally so the failure is reported once, via the promise. +// Sibling of the end() test above for sink.close() (js_close -> FileSink::end()). +// end()'s Err/Done/Wrote arms set done=true / tore down the writer without +// ever scheduling run_pending, so a backpressured write()'s promise was left +// pending forever. close() can't hand the promise back, so the outcome goes to +// it via run_pending and close() returns undefined. +// +// Runs in a subprocess because sink.close() on a Blob-created FileSink +// currently leaks the native FileSink (doClose detaches m_sinkPtr so the +// wrapper's +1 never reaches finalize); running it in-process would abort the +// whole file under detect_leaks=1. That leak is pre-existing on main and +// tracked separately. it.skipIf(!isPosix)( - "close() after a backpressured write() with the reader gone rejects the write's promise with EPIPE", + "close() after a backpressured write() settles the write's promise (Err + drained-Done arms)", async () => { - const [readFd, writeFd] = createSocketPair(); - let readFdOpen = true; - const sink = Bun.file(writeFd).writer(); - try { - const writePromise = sink.write(Buffer.alloc(4 * 1024 * 1024, 0x61)); - expect(writePromise).toBeInstanceOf(Promise); - - fs.closeSync(readFd); - readFdOpen = false; - - // close()'s flush hits EPIPE. The error is routed to writePromise; the - // close() call itself doesn't throw (no double reporting). - expect(sink.close()).toBeUndefined(); + const src = ` + const { createSocketPair } = require("bun:internal-for-testing"); + const fs = require("node:fs"); + + async function errArm() { + const [readFd, writeFd] = createSocketPair(); + const sink = Bun.file(writeFd).writer(); + const p = sink.write(Buffer.alloc(4 * 1024 * 1024, 0x61)); + if (!(p instanceof Promise)) return "err-arm:not-backpressured"; + fs.closeSync(readFd); + let threw = false; + try { sink.close(); } catch { threw = true; } + if (threw) return "err-arm:close-threw"; + try { await p; return "err-arm:resolved"; } + catch (e) { return "err-arm:" + (e?.code ?? "unknown"); } + } - let caught: any; - try { - await writePromise; - } catch (e) { - caught = e; + async function drainedArm() { + const [readFd, writeFd] = createSocketPair(); + const sink = Bun.file(writeFd).writer(); + const size = 300 * 1024; + const p = sink.write(Buffer.alloc(size, 0x61)); + if (!(p instanceof Promise)) return "drained-arm:not-backpressured"; + const buf = Buffer.alloc(64 * 1024); + while (true) { try { if (!fs.readSync(readFd, buf)) break; } catch { break; } } + sink.close(); + // Drain any remainder close()'s flush pushed to the socket. + while (true) { try { if (!fs.readSync(readFd, buf)) break; } catch { break; } } + let settled = "timeout"; + await Promise.race([ + p.then(v => { settled = String(v); }, e => { settled = "rejected:" + (e?.code ?? e); }), + // run_pending is an enqueued task, not a timing race; the fixed build + // settles within one tick. The bound only exists for the fail-before. + new Promise(r => setTimeout(r, 500)), + ]); + try { fs.closeSync(readFd); } catch {} + try { fs.closeSync(writeFd); } catch {} + return "drained-arm:" + settled + ":" + size; } - expect(caught?.code).toBe("EPIPE"); - } finally { - try { - fs.closeSync(writeFd); - } catch {} - if (readFdOpen) fs.closeSync(readFd); - } + + console.log(await errArm()); + console.log(await drainedArm()); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", src], + env: { + ...bunEnv, + // Pre-existing leak in sink.close() (see comment above); don't let the + // child's LSAN abort hide the actual assertion we're testing. + ASAN_OPTIONS: "allow_user_segv_handler=1:disable_coredump=0:detect_leaks=0", + }, + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + const lines = stdout.trim().split("\n"); + expect(lines[0]).toBe("err-arm:EPIPE"); + // The drained arm resolves with the bytes write() accepted. + const [, settled, size] = lines[1].split(":"); + expect(settled).toBe(size); + expect(exitCode).toBe(0); }, ); From 159dbd592c5470455e9c89919ad5d4198c7910e5 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 24 Jul 2026 05:11:42 +0000 Subject: [PATCH 3/4] Close the class: fix end_from_js()'s Done/Wrote arms; drop the platform-fragile drained close() sub-test end_from_js()'s Done/Wrote arms had the identical orphan (#35344 only fixed its Err arm). When a backpressured write()'s promise is outstanding they now hand that promise back (like the Err/Pending arms already do) and schedule run_pending to resolve it, instead of returning a bare byte count while the promise hangs. The close() subprocess test's drained-arm check depended on Linux's ~200KB AF_UNIX send buffer letting close()'s flush drain the remainder in one go; on macOS (~8KB) flush() returns Pending and the test would have timed out. Drop it and cover the Done/Wrote path via an in-process sink.end() test instead, which is Linux-only for the same reason (on macOS the Pending arm was already correct) and doesn't hit the pre-existing close() leak. --- src/runtime/webcore/FileSink.rs | 29 ++++++-- test/js/bun/util/filesink.test.ts | 115 +++++++++++++++++------------- 2 files changed, 90 insertions(+), 54 deletions(-) diff --git a/src/runtime/webcore/FileSink.rs b/src/runtime/webcore/FileSink.rs index 504818daa302..f0a207de17bf 100644 --- a/src/runtime/webcore/FileSink.rs +++ b/src/runtime/webcore/FileSink.rs @@ -1047,9 +1047,8 @@ impl FileSink { // A backpressured `write()` may have left its promise in `self.pending`; // `writer.end()` only re-enters `on_close`, which never touches it, so // every synchronous arm that tears the writer down here must settle it - // (mirrors `on_auto_flush` / `end_from_js`). `js_close` can't hand the - // promise back, so the outcome is delivered via `run_pending` and the - // call returns `Ok`. + // (mirrors `on_auto_flush`). `js_close` can't hand the promise back, so + // the outcome is delivered via `run_pending` and the call returns `Ok`. let has_pending = self.pending.get().state == streams::PendingState::Pending; // SAFETY(JsCell): `IOWriter::flush` is pure I/O; any callback re-entry @@ -1140,15 +1139,30 @@ impl FileSink { // SAFETY(JsCell): `IOWriter::flush` is pure I/O; no JS while held. let flush_result = self.writer.with_mut(|w| w.flush()); + // `writer.end()` only re-enters `on_close`, which never touches + // `self.pending`; every arm that tears the writer down here with a + // backpressured `write()` outstanding must hand that promise back and + // schedule `run_pending` to settle it. + let has_pending = self.pending.get().state == streams::PendingState::Pending; + match flush_result { WriteResult::Done(written) => { self.update_ref(false); self.writer.with_mut(|w| w.end()); + if has_pending { + // `to_result` already seeded `Owned(consumed)`. + // SAFETY: JsCell — `WritablePending::promise` allocates a + // JSPromise (may GC) but invokes no FileSink host-fn. + let promise = unsafe { self.pending.get_mut() }.promise(global_this); + self.run_pending_later(); + // SAFETY: `WritablePending::promise()` never returns null. + return sys::Result::Ok(unsafe { (*promise).to_js() }); + } sys::Result::Ok(JSValue::js_number(written as f64)) } WriteResult::Err(err) => { self.done.set(true); - if self.pending.get().state == streams::PendingState::Pending { + if has_pending { // A backpressured write() left its promise outstanding. // Throwing here would report the failure to the caller and // then let the auto-flush/error path reject that promise a @@ -1202,6 +1216,13 @@ impl FileSink { } WriteResult::Wrote(written) => { self.writer.with_mut(|w| w.end()); + if has_pending { + // SAFETY: JsCell — see the `Done` arm above. + let promise = unsafe { self.pending.get_mut() }.promise(global_this); + self.run_pending_later(); + // SAFETY: `WritablePending::promise()` never returns null. + return sys::Result::Ok(unsafe { (*promise).to_js() }); + } sys::Result::Ok(JSValue::js_number(written as f64)) } } diff --git a/test/js/bun/util/filesink.test.ts b/test/js/bun/util/filesink.test.ts index 6007af60e772..c3a7ca4205da 100644 --- a/test/js/bun/util/filesink.test.ts +++ b/test/js/bun/util/filesink.test.ts @@ -1,6 +1,6 @@ import { createSocketPair, fileSinkInternals } from "bun:internal-for-testing"; import { describe, expect, it } from "bun:test"; -import { bunEnv, bunExe, fileDescriptorLeakChecker, isPosix, isWindows, tmpdirSync } from "harness"; +import { bunEnv, bunExe, fileDescriptorLeakChecker, isLinux, isPosix, isWindows, tmpdirSync } from "harness"; import { mkfifo } from "mkfifo"; import { join } from "node:path"; @@ -319,10 +319,10 @@ it.skipIf(!isPosix)( ); // Sibling of the end() test above for sink.close() (js_close -> FileSink::end()). -// end()'s Err/Done/Wrote arms set done=true / tore down the writer without -// ever scheduling run_pending, so a backpressured write()'s promise was left -// pending forever. close() can't hand the promise back, so the outcome goes to -// it via run_pending and close() returns undefined. +// end()'s Err arm set done=true and tore down the writer without scheduling +// run_pending, so a backpressured write()'s promise was left pending forever +// while close() threw. Now close() routes the error to that promise and +// returns undefined. // // Runs in a subprocess because sink.close() on a Blob-created FileSink // currently leaks the native FileSink (doClose detaches m_sinkPtr so the @@ -330,50 +330,21 @@ it.skipIf(!isPosix)( // whole file under detect_leaks=1. That leak is pre-existing on main and // tracked separately. it.skipIf(!isPosix)( - "close() after a backpressured write() settles the write's promise (Err + drained-Done arms)", + "close() after a backpressured write() with the reader gone rejects the write's promise with EPIPE", async () => { const src = ` const { createSocketPair } = require("bun:internal-for-testing"); const fs = require("node:fs"); - - async function errArm() { - const [readFd, writeFd] = createSocketPair(); - const sink = Bun.file(writeFd).writer(); - const p = sink.write(Buffer.alloc(4 * 1024 * 1024, 0x61)); - if (!(p instanceof Promise)) return "err-arm:not-backpressured"; - fs.closeSync(readFd); - let threw = false; - try { sink.close(); } catch { threw = true; } - if (threw) return "err-arm:close-threw"; - try { await p; return "err-arm:resolved"; } - catch (e) { return "err-arm:" + (e?.code ?? "unknown"); } - } - - async function drainedArm() { - const [readFd, writeFd] = createSocketPair(); - const sink = Bun.file(writeFd).writer(); - const size = 300 * 1024; - const p = sink.write(Buffer.alloc(size, 0x61)); - if (!(p instanceof Promise)) return "drained-arm:not-backpressured"; - const buf = Buffer.alloc(64 * 1024); - while (true) { try { if (!fs.readSync(readFd, buf)) break; } catch { break; } } - sink.close(); - // Drain any remainder close()'s flush pushed to the socket. - while (true) { try { if (!fs.readSync(readFd, buf)) break; } catch { break; } } - let settled = "timeout"; - await Promise.race([ - p.then(v => { settled = String(v); }, e => { settled = "rejected:" + (e?.code ?? e); }), - // run_pending is an enqueued task, not a timing race; the fixed build - // settles within one tick. The bound only exists for the fail-before. - new Promise(r => setTimeout(r, 500)), - ]); - try { fs.closeSync(readFd); } catch {} - try { fs.closeSync(writeFd); } catch {} - return "drained-arm:" + settled + ":" + size; - } - - console.log(await errArm()); - console.log(await drainedArm()); + const [readFd, writeFd] = createSocketPair(); + const sink = Bun.file(writeFd).writer(); + const p = sink.write(Buffer.alloc(4 * 1024 * 1024, 0x61)); + if (!(p instanceof Promise)) { console.log("not-backpressured"); process.exit(0); } + fs.closeSync(readFd); + let threw = false; + try { sink.close(); } catch { threw = true; } + if (threw) { console.log("close-threw"); process.exit(0); } + try { await p; console.log("resolved"); } + catch (e) { console.log(e?.code ?? "unknown"); } `; await using proc = Bun.spawn({ cmd: [bunExe(), "-e", src], @@ -387,15 +358,59 @@ it.skipIf(!isPosix)( }); const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); expect(stderr).toBe(""); - const lines = stdout.trim().split("\n"); - expect(lines[0]).toBe("err-arm:EPIPE"); - // The drained arm resolves with the bytes write() accepted. - const [, settled, size] = lines[1].split(":"); - expect(settled).toBe(size); + expect(stdout.trim()).toBe("EPIPE"); expect(exitCode).toBe(0); }, ); +// end()'s and end_from_js()'s Done/Wrote arms had the same orphan: when the +// reader drains between write() and end(), flush() pushes the whole remaining +// buffer through in one shot and returns Done/Wrote, and writer.end() only +// fires on_close (which never touches pending). Linux-only because the +// drain-flush-drain shape needs the AF_UNIX send buffer to hold the remainder +// (Linux default ~200KB; macOS is ~8KB, so flush() returns Pending there and +// the promise was already settled via on_write). +it.skipIf(!isLinux)( + "end() after a backpressured write() with the reader drained returns the write's promise and resolves it", + async () => { + const [readFd, writeFd] = createSocketPair(); + let readFdOpen = true; + const sink = Bun.file(writeFd).writer(); + const size = 300 * 1024; + try { + const writePromise = sink.write(Buffer.alloc(size, 0x61)); + expect(writePromise).toBeInstanceOf(Promise); + + const buf = Buffer.alloc(64 * 1024); + const drain = () => { + while (true) + try { + if (!fs.readSync(readFd, buf)) break; + } catch { + break; + } + }; + drain(); + + // flush() now drains the sink's remaining buffer in one write; the + // Done/Wrote arm hands back the write()'s promise and schedules + // run_pending to resolve it with the bytes write() accepted. + const endResult = sink.end(); + expect(endResult).toBe(writePromise); + drain(); + expect(await writePromise).toBe(size); + } finally { + try { + await Promise.resolve(sink.end()).catch(() => {}); + } catch {} + try { + fs.closeSync(writeFd); + } catch {} + if (readFdOpen) fs.closeSync(readFd); + } + }, +); + // The deferred auto-flush microtask runs at the first microtask checkpoint // after write() backpressures. If its flush() hit EPIPE, it discarded the // error and then let `run_pending_later()` resolve the pending write() promise From ca307169dd3052120b139ceff03ad39d327356e5 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 24 Jul 2026 05:40:08 +0000 Subject: [PATCH 4/4] test: drop dead readFdOpen flag in the drained-end() test --- test/js/bun/util/filesink.test.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/test/js/bun/util/filesink.test.ts b/test/js/bun/util/filesink.test.ts index c3a7ca4205da..e78190256084 100644 --- a/test/js/bun/util/filesink.test.ts +++ b/test/js/bun/util/filesink.test.ts @@ -374,7 +374,6 @@ it.skipIf(!isLinux)( "end() after a backpressured write() with the reader drained returns the write's promise and resolves it", async () => { const [readFd, writeFd] = createSocketPair(); - let readFdOpen = true; const sink = Bun.file(writeFd).writer(); const size = 300 * 1024; try { @@ -406,7 +405,7 @@ it.skipIf(!isLinux)( try { fs.closeSync(writeFd); } catch {} - if (readFdOpen) fs.closeSync(readFd); + fs.closeSync(readFd); } }, );