FileSink: settle the pending write()'s promise in every synchronous close()/end() flush arm

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))
             }
         }

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