From a684ff34052cf8a68555922e244116ef2fc7f5a1 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 24 Jul 2026 05:47:24 +0000 Subject: [PATCH] FileSink: resolve the pending write()'s promise when end()'s flush drains it When a backpressured write() leaves its promise outstanding and the caller then drains the read side and calls sink.end() before any await, end_from_js's own flush() can push the buffered remainder out synchronously and land in the Done/Wrote arm. Those arms called writer.end() and returned the drained byte count without touching the pending slot; IOWriter::flush() doesn't route through on_write, so nothing else ever settled the write()'s promise and it hung forever. Mirror the Pending/Err arms: when a write is pending, hand back that promise and schedule run_pending to resolve it with the Owned(consumed) that to_result already latched. The no-pending path is unchanged. --- src/runtime/webcore/FileSink.rs | 34 ++++++++++++ test/js/bun/util/filesink.test.ts | 88 +++++++++++++++++++++++++++++++ 2 files changed, 122 insertions(+) diff --git a/src/runtime/webcore/FileSink.rs b/src/runtime/webcore/FileSink.rs index cbc45719711c..6007dbce2869 100644 --- a/src/runtime/webcore/FileSink.rs +++ b/src/runtime/webcore/FileSink.rs @@ -1128,6 +1128,25 @@ impl FileSink { match flush_result { WriteResult::Done(written) => { + if self.pending.get().state == streams::PendingState::Pending { + // A backpressured `write()` left its promise outstanding and + // `flush()` drained the remainder just now. `IOWriter::flush` + // doesn't route through `on_write`, so nothing else will + // settle that promise: hand it back (like the Pending/Err + // arms) and schedule `run_pending`. `pending.result` already + // holds `Owned(consumed)` from `to_result`. Grab the promise + // before `writer.end()` so `on_close` re-entry runs with the + // slot already pinned. + // SAFETY: JsCell — `WritablePending::promise` allocates a + // JSPromise (may GC) but does not invoke any FileSink + // host-fn synchronously. + let promise_result = unsafe { self.pending.get_mut() }.promise(global_this); + self.update_ref(false); + self.writer.with_mut(|w| w.end()); + self.run_pending_later(); + // SAFETY: `WritablePending::promise()` never returns null. + return sys::Result::Ok(unsafe { (*promise_result).to_js() }); + } self.update_ref(false); self.writer.with_mut(|w| w.end()); sys::Result::Ok(JSValue::js_number(written as f64)) @@ -1187,6 +1206,21 @@ impl FileSink { sys::Result::Ok(unsafe { (*promise_result).to_js() }) } WriteResult::Wrote(written) => { + if self.pending.get().state == streams::PendingState::Pending { + // Same as the Done arm above: `flush()` drained the + // backpressured write's remainder without routing through + // `on_write`, so hand back the outstanding promise and + // schedule `run_pending` to resolve it with the + // `Owned(consumed)` `to_result` already latched. + // SAFETY: JsCell — `WritablePending::promise` allocates a + // JSPromise (may GC) but does not invoke any FileSink + // host-fn synchronously. + let promise_result = unsafe { self.pending.get_mut() }.promise(global_this); + self.writer.with_mut(|w| w.end()); + self.run_pending_later(); + // SAFETY: `WritablePending::promise()` never returns null. + return sys::Result::Ok(unsafe { (*promise_result).to_js() }); + } self.writer.with_mut(|w| w.end()); 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..0299cf651e0b 100644 --- a/test/js/bun/util/filesink.test.ts +++ b/test/js/bun/util/filesink.test.ts @@ -273,6 +273,94 @@ it.skipIf(!isPosix)("a backpressured string write() resolves to its encoded byte expect(received).toBe(size); }); +// end() called after a backpressured write() once the reader has drained the +// socket buffer: end_from_js's own flush() writes the remaining buffered bytes +// synchronously and lands in the Done/Wrote arm. Those arms used to return the +// drained byte count without touching the pending slot, so the write()'s +// promise was orphaned and never settled. end() now hands back that same +// promise (like the Pending/Err arms) and schedules run_pending to resolve it +// with the write's chunk size. +it.skipIf(!isPosix)( + "end() after a backpressured write() whose remainder drains synchronously returns and resolves the write's promise", + async () => { + const [readFd, writeFd] = createSocketPair(); + const sink = Bun.file(writeFd).writer(); + try { + // Write 16 KiB chunks until one backpressures. 16 KiB is at least the + // sink's CHUNK_SIZE on every target so each write reaches the fd + // immediately, and when backpressure hits the sink's buffered remainder + // is at most one chunk — small enough that end()'s flush() can push it + // out in one go after the read side has been drained. + const chunkSize = 16 * 1024; + const chunk = Buffer.alloc(chunkSize, 0x61); + let total = 0; + let writePromise!: Promise; + for (let i = 0; i < 4096; i++) { + const r = sink.write(chunk); + total += chunkSize; + if (r instanceof Promise) { + writePromise = r; + break; + } + } + expect(writePromise).toBeInstanceOf(Promise); + + // Drain the read side synchronously so the socket buffer has room before + // end()'s flush() runs. No await in between: the deferred auto-flush + // (which already handles this correctly) must not fire first. + const buf = Buffer.alloc(64 * 1024); + let drained = 0; + const drainSync = () => { + for (;;) { + try { + const n = fs.readSync(readFd, buf); + if (!n) break; + drained += n; + } catch { + break; + } + } + }; + drainSync(); + expect(drained).toBeGreaterThan(0); + + // flush() writes the buffered remainder. When it completes (Done/Wrote), + // end() hands back the write()'s promise instead of a bare number that + // strands it; when the remainder still doesn't fit (Pending) it already + // did. Either way end()'s result and write()'s promise are the same + // object and it resolves to the backpressured chunk's size. + const endResult = sink.end(); + expect(endResult).toBe(writePromise); + + // Keep the reader draining so any still-pending tail finishes. + const reader = (async () => { + while (drained < total) { + drainSync(); + if (drained >= total) break; + await new Promise(r => setImmediate(r)); + } + })(); + + expect(await writePromise).toBe(chunkSize); + await reader; + expect(drained).toBe(total); + + // Once settled a follow-up end() short-circuits to a number. + expect(typeof sink.end()).toBe("number"); + } finally { + try { + await Promise.resolve(sink.end()).catch(() => {}); + } catch {} + try { + fs.closeSync(writeFd); + } catch {} + try { + fs.closeSync(readFd); + } catch {} + } + }, +); + // end() called after a backpressured write() with the reader already gone: // end_from_js's own flush() sees EPIPE synchronously. Throwing it would leave // the write()'s outstanding promise orphaned (never settled here; in the spawn