diff --git a/src/runtime/webcore/FileSink.rs b/src/runtime/webcore/FileSink.rs index cbc45719711c..f0a207de17bf 100644 --- a/src/runtime/webcore/FileSink.rs +++ b/src/runtime/webcore/FileSink.rs @@ -1044,16 +1044,34 @@ 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`). `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 has_pending { + 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) } @@ -1066,11 +1084,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(()) - } } } @@ -1126,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 @@ -1188,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 72da12167a4f..e78190256084 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"; @@ -318,6 +318,98 @@ it.skipIf(!isPosix)( }, ); +// Sibling of the end() test above for sink.close() (js_close -> FileSink::end()). +// 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 +// 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", + async () => { + const src = ` + const { createSocketPair } = require("bun:internal-for-testing"); + const fs = require("node:fs"); + 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], + 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(""); + 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(); + 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 {} + 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