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

src/runtime/webcore/FileSink.rs

@@ -1044,16 +1044,35 @@
             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 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 +1085,6 @@
                 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(())
-            }
         }
     }
 

test/js/bun/util/filesink.test.ts

@@ -318,6 +318,84 @@
   },
 );
 
+// 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() settles the write's promise (Err + drained-Done arms)",
+  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());
+    `;
+    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);
+  },
+);
+
 // 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