FileSink: resolve a backpressured write() to its chunk's byte count

src/io/PipeWriter.rs

@@ -745,6 +745,11 @@ impl<Parent: PosixStreamingWriterParent> PosixStreamingWriter<Parent> {
         self.outgoing.is_not_empty()
     }
 
+    /// Bytes accepted from callers that have not reached the fd yet.
+    pub fn buffered_len(&self) -> usize {
+        self.outgoing.size()
+    }
+
     pub fn should_buffer(&self, addition: usize) -> bool {
         !self.force_sync && self.outgoing.size() + addition < Self::CHUNK_SIZE
     }
@@ -2095,6 +2100,12 @@ impl<Parent: WindowsStreamingWriterParent> WindowsStreamingWriter<Parent> {
         self.outgoing.is_not_empty() || self.current_payload.is_not_empty()
     }
 
+    /// Bytes accepted from callers that have not reached the fd yet: queued in
+    /// `outgoing` or handed to libuv in `current_payload`.
+    pub fn buffered_len(&self) -> usize {
+        self.outgoing.size() + self.current_payload.size()
+    }
+
     fn on_write_complete(&mut self, status: uv::ReturnCode) {
         // PORT_NOTES_PLAN R-2: `&mut self` carries LLVM `noalias`, but
         // `Parent::on_write` (e.g. `FileSink::on_write`) re-enters JS via

src/runtime/webcore/FileSink.rs

@@ -418,17 +418,14 @@ impl FileSink {
 
             // if we are not done yet and has pending data we just wait so we do not runPending twice
             if status == WriteStatus::Pending && has_pending_data {
-                if (*this).pending.get().state == streams::PendingState::Pending {
-                    (*this).pending.with_mut(|p| p.consumed = amount as u64); // @truncate
-                }
                 return;
             }
 
             if (*this).pending.get().state == streams::PendingState::Pending {
-                (*this).pending.with_mut(|p| p.consumed = amount as u64); // @truncate
-
-                // when "done" is true, we will never receive more data.
+                // `consumed` was credited when the pending operation accepted its
+                // bytes; `amount` is only what this drain pushed to the fd.
                 let consumed = (*this).pending.get().consumed;
+                // when "done" is true, we will never receive more data.
                 if (*this).done.get() || status == WriteStatus::EndOfFile {
                     (*this)
                         .pending
@@ -879,21 +876,20 @@ impl FileSink {
         // SAFETY(JsCell): `IOWriter::flush` is pure I/O; no JS re-entry while
         // the `&mut IOWriter` is held.
         let rc = self.writer.with_mut(|w| w.flush());
-        match rc {
-            WriteResult::Done(written) => {
-                self.written.set(self.written.get() + written as usize); // @truncate
-            }
-            WriteResult::Pending(written) => {
-                self.written.set(self.written.get() + written as usize); // @truncate
-            }
-            WriteResult::Wrote(written) => {
+        let flushed = match rc {
+            WriteResult::Done(written)
+            | WriteResult::Pending(written)
+            | WriteResult::Wrote(written) => {
                 self.written.set(self.written.get() + written as usize); // @truncate
+                written as u64 // @truncate
             }
             WriteResult::Err(err) => {
                 return sys::Result::Err(err);
             }
-        }
-        match self.to_result(rc) {
+        };
+        // A flush takes no new chunk from the caller; a pending one reports the
+        // bytes it pushed out. It only reaches here when no write is pending.
+        match self.to_result(rc, flushed) {
             streams::Writable::Err(_) => unreachable!(),
             result => sys::Result::Ok(result.to_js(global_this)),
         }
@@ -969,9 +965,11 @@ impl FileSink {
         if self.done.get() {
             return streams::Writable::Done;
         }
+        let buffered_before = self.writer.get().buffered_len();
         // SAFETY(JsCell): `IOWriter::write` buffers/writes to fd; does not call JS.
         let rc = self.writer.with_mut(|w| w.write(data.slice()));
-        self.to_result(rc)
+        let accepted = self.bytes_accepted(buffered_before, &rc);
+        self.to_result(rc, accepted)
     }
 
     #[inline]
@@ -983,18 +981,22 @@ impl FileSink {
         if self.done.get() {
             return streams::Writable::Done;
         }
+        let buffered_before = self.writer.get().buffered_len();
         // SAFETY(JsCell): `IOWriter::write_latin1` buffers/writes; no JS.
         let rc = self.writer.with_mut(|w| w.write_latin1(data.slice()));
-        self.to_result(rc)
+        let accepted = self.bytes_accepted(buffered_before, &rc);
+        self.to_result(rc, accepted)
     }
 
     pub fn write_utf16(&self, data: &streams::Result) -> streams::Writable {
         if self.done.get() {
             return streams::Writable::Done;
         }
+        let buffered_before = self.writer.get().buffered_len();
         // SAFETY(JsCell): `IOWriter::write_utf16` buffers/writes; no JS.
         let rc = self.writer.with_mut(|w| w.write_utf16(data.slice16()));
-        self.to_result(rc)
+        let accepted = self.bytes_accepted(buffered_before, &rc);
+        self.to_result(rc, accepted)
     }
 
     pub fn end(&self, _err: Option<sys::Error>) -> sys::Result<()> {
@@ -1103,8 +1105,14 @@ impl FileSink {
                     self.ref_();
                 }
                 self.done.set(true);
-                self.pending
-                    .with_mut(|p| p.result = streams::Writable::Owned(pending_written as u64));
+                self.pending.with_mut(|p| {
+                    // A write already pending on this slot owns `consumed`; seed it
+                    // only when `end()` is the call that opens the slot.
+                    if p.state != streams::PendingState::Pending {
+                        p.consumed += pending_written as u64; // @truncate
+                    }
+                    p.result = streams::Writable::Owned(p.consumed);
+                });
 
                 // SAFETY: JsCell — `WritablePending::promise` allocates a JSPromise
                 // (may GC) but does not invoke any FileSink host-fn synchronously.
@@ -1230,7 +1238,22 @@ impl FileSink {
         }
     }
 
-    fn to_result(&self, write_result: WriteResult) -> streams::Writable {
+    /// Bytes the writer took off our hands in the `write_*` call that produced
+    /// `rc`: what reached the fd plus what it buffered for later. The writer
+    /// never takes part of a chunk, so for a `Pending` result this is the
+    /// chunk's own (encoded) byte count, not the partial `write(2)` return.
+    fn bytes_accepted(&self, buffered_before: usize, rc: &WriteResult) -> u64 {
+        let WriteResult::Pending(written) = rc else {
+            return 0;
+        };
+        let buffered_after = self.writer.get().buffered_len();
+        (buffered_after + written).saturating_sub(buffered_before) as u64 // @truncate
+    }
+
+    /// `accepted` is what the pending slot is credited with when `write_result`
+    /// is `Pending`: a write's full chunk, or the bytes a flush pushed out. It
+    /// is ignored for every other result.
+    fn to_result(&self, write_result: WriteResult, accepted: u64) -> streams::Writable {
         match write_result {
             WriteResult::Done(amt) => {
                 if amt > 0 {
@@ -1245,14 +1268,14 @@ impl FileSink {
                 streams::Writable::Temporary(amt as u64)
             }
             WriteResult::Err(err) => streams::Writable::Err(err),
-            WriteResult::Pending(pending_written) => {
+            WriteResult::Pending(_) => {
                 if !self.must_be_kept_alive_until_eof.get() {
                     self.must_be_kept_alive_until_eof.set(true);
                     self.ref_();
                 }
                 self.pending.with_mut(|p| {
-                    p.consumed += pending_written as u64; // @truncate
-                    p.result = streams::Writable::Owned(pending_written as u64);
+                    p.consumed += accepted;
+                    p.result = streams::Writable::Owned(p.consumed);
                 });
                 streams::Writable::Pending(self.pending.as_ptr())
             }

src/runtime/webcore/streams.rs

@@ -498,6 +498,9 @@ impl WritablePending {
             return;
         }
         self.state = PendingState::Used;
+        // `consumed` belongs to the operation being settled here; the next one
+        // starts from zero.
+        self.consumed = 0;
 
         match core::mem::replace(&mut self.future, WritableFuture::None) {
             WritableFuture::Promise { mut strong, global } => {

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

@@ -207,6 +207,72 @@ it("write result is not cumulative", async () => {
   await util.promisify(fs.close)(fd);
 });
 
+// A backpressured write buffers everything `write(2)` would not take, so the
+// Promise it returns has to resolve with the chunk's own byte count. It used to
+// resolve with the partial `write(2)` return instead.
+it.skipIf(!isPosix)("a backpressured write() resolves to the chunk's byte count", async () => {
+  const [readFd, writeFd] = createSocketPair();
+  const sink = Bun.file(writeFd).writer();
+  const size = 4 * 1024 * 1024;
+  const chunk = Buffer.alloc(size, 0x61);
+
+  // Nothing drains `readFd` yet, so the socket buffers fill up and only part of
+  // the chunk reaches the fd.
+  const first = sink.write(chunk);
+
+  let received = 0;
+  const reader = (async () => {
+    for await (const part of Bun.file(readFd).stream()) received += part.byteLength;
+  })();
+
+  try {
+    expect(first).toBeInstanceOf(Promise);
+    expect(await first).toBe(size);
+
+    // The next backpressured write starts its own accounting.
+    const second = sink.write(chunk);
+    expect(second).toBeInstanceOf(Promise);
+    expect(await second).toBe(size);
+  } finally {
+    await Promise.resolve(sink.end()).catch(() => {});
+    fs.closeSync(writeFd);
+    await reader;
+    fs.closeSync(readFd);
+  }
+
+  expect(received).toBe(size * 2);
+});
+
+// Strings are buffered as UTF-8, so the count the Promise reports is the
+// encoded byte count, which is what a non-pending write() returns too.
+it.skipIf(!isPosix)("a backpressured string write() resolves to its encoded byte count", async () => {
+  const [readFd, writeFd] = createSocketPair();
+  const sink = Bun.file(writeFd).writer();
+  // Latin-1 in JSC, two bytes per character once encoded.
+  const text = Buffer.alloc(2 * 1024 * 1024, "é").toString();
+  const size = Buffer.byteLength(text);
+  expect(size).toBe(text.length * 2);
+
+  const written = sink.write(text);
+
+  let received = 0;
+  const reader = (async () => {
+    for await (const part of Bun.file(readFd).stream()) received += part.byteLength;
+  })();
+
+  try {
+    expect(written).toBeInstanceOf(Promise);
+    expect(await written).toBe(size);
+  } finally {
+    await Promise.resolve(sink.end()).catch(() => {});
+    fs.closeSync(writeFd);
+    await reader;
+    fs.closeSync(readFd);
+  }
+
+  expect(received).toBe(size);
+});
+
 if (isWindows) {
   it("ENOENT, Windows", () => {
     expect(() => Bun.file("A:\\this-does-not-exist.txt").writer()).toThrow(
@@ -324,7 +390,9 @@ it("Bun.file(fd).writer() write/end under GC pressure does not crash", async ()
         const fs = require("fs");
         const fd = fs.openSync(${JSON.stringify(join(dir, "out.txt"))}, "w");
         const buf = Buffer.alloc(64 * 1024, 0x61);
-        for (let i = 0; i < 200; i++) {
+        // A synchronous Bun.gc() costs ~18ms under debug+ASAN; 50 rounds keeps
+        // this inside the default timeout there, and still reproduced the crash.
+        for (let i = 0; i < 50; i++) {
           const w = Bun.file(fd).writer();
           const p = w.write(buf);
           if (p && typeof p.then === "function") await p;