FileSink: reject the pending write() when the deferred auto-flush hits EPIPE

src/runtime/webcore/FileSink.rs

@@ -829,7 +829,26 @@ impl FileSink {
             // SAFETY(JsCell): `IOWriter::flush` is pure I/O; the `on_write`
             // callback it may trigger goes via the stored `*mut FileSink` backref.
             match (*this).writer.with_mut(|w| w.flush()) {
-                WriteResult::Err(_) | WriteResult::Done(_) => {
+                WriteResult::Err(err) => {
+                    (*this).update_ref(false);
+                    // `flush()` returns a write error without routing through the
+                    // writer's `_on_error`, so the pending slot still holds the
+                    // `Owned(consumed)` result `to_result` seeded and
+                    // `run_pending_later()` alone would resolve it as if every
+                    // buffered byte had reached the reader. Latch the error and
+                    // move the sink to its terminal state (mirrors `end_from_js`).
+                    (*this).done.set(true);
+                    if (*this).pending.get().state == streams::PendingState::Pending {
+                        (*this)
+                            .pending
+                            .with_mut(|p| p.result = streams::Writable::Err(err));
+                    }
+                    (*this).writer.with_mut(|w| w.end());
+                    (*this).run_pending_later();
+                    (*this).auto_flusher.with_mut(|a| a.registered.set(false));
+                    return false;
+                }
+                WriteResult::Done(_) => {
                     (*this).update_ref(false);
                     (*this).run_pending_later();
                 }

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

@@ -273,6 +273,58 @@ it.skipIf(!isPosix)("a backpressured string write() resolves to its encoded byte
   expect(received).toBe(size);
 });
 
+// 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
+// with the `Owned(consumed)` result `to_result` had seeded, so `await write()`
+// + `await end()` both succeeded even though the reader was already gone and
+// nearly the whole chunk was still sitting in the sink's buffer.
+it.skipIf(!isPosix)(
+  "a backpressured write() rejects with EPIPE when the reader closes before the deferred flush",
+  async () => {
+    const [readFd, writeFd] = createSocketPair();
+    let readFdOpen = true;
+    const sink = Bun.file(writeFd).writer();
+    const size = 4 * 1024 * 1024;
+
+    try {
+      const writePromise = sink.write(Buffer.alloc(size, 0x61));
+      expect(writePromise).toBeInstanceOf(Promise);
+
+      // Close the reader before the first await: the deferred auto-flush fires
+      // as part of that await's microtask drain, and its flush() now sees EPIPE.
+      fs.closeSync(readFd);
+      readFdOpen = false;
+
+      let caught: any;
+      try {
+        await writePromise;
+      } catch (e) {
+        caught = e;
+      }
+      expect(caught?.code).toBe("EPIPE");
+
+      // The Err arm also moves the sink to its terminal state: further writes
+      // short-circuit to Writable::Done (=> true).
+      expect(sink.write("x")).toBe(true);
+
+      // end() after the error reports the bytes that actually reached the fd;
+      // the point is it doesn't claim the full chunk was delivered.
+      const endRes = await sink.end();
+      expect(typeof endRes).toBe("number");
+      expect(endRes).toBeLessThan(size);
+    } finally {
+      try {
+        await sink.end();
+      } catch {}
+      try {
+        fs.closeSync(writeFd);
+      } catch {}
+      if (readFdOpen) fs.closeSync(readFd);
+    }
+  },
+);
+
 if (isWindows) {
   it("ENOENT, Windows", () => {
     expect(() => Bun.file("A:\\this-does-not-exist.txt").writer()).toThrow(