process.stdout: an empty write() calls back after the queued chunks

src/js/internal/fs/streams.ts

@@ -33,6 +33,7 @@ type FSStream = Omit<import("node:fs").ReadStream & import("node:fs").WriteStrea
    * sink = FileSink
    */
   [kWriteStreamFastPath]?: undefined | true | FileSink;
+  [kDrainedWrite]?: Promise<unknown>;
 };
 type FD = number;
 
@@ -43,6 +44,7 @@ const kIoDone = Symbol("kIoDone");
 // Bun supports a fast path for `createWriteStream("path.txt")` where instead of
 // using `node:fs`, `Bun.file(...).writer()` is used instead.
 const kWriteStreamFastPath = Symbol("kWriteStreamFastPath");
+const kDrainedWrite = Symbol("kDrainedWrite");
 const kFs = Symbol("kFs");
 
 const {
@@ -451,6 +453,7 @@ function WriteStream(this: FSStream, path: string | null | undefined, options?:
   this.pos = undefined;
   this.bytesWritten = 0;
   this[kIsPerformingIO] = false;
+  this[kDrainedWrite] = undefined;
 
   if (start !== undefined) {
     validateInteger(start, "start", 0);
@@ -560,7 +563,10 @@ function writevAll(chunks, size, pos, cb, retries = 0) {
 }
 
 // Like Writable's afterWrite: no 'drain' once the stream is ending or destroyed.
-function emitDrain(stream: FSStream) {
+// Chunks written while the sink is backed up share `write`, its promise: one 'drain' for all of them.
+function emitDrain(stream: FSStream, write: Promise<unknown>) {
+  if (stream[kDrainedWrite] === write) return;
+  stream[kDrainedWrite] = write;
   const state = stream._writableState;
   if (state === undefined || !(state.ending || state.destroyed)) {
     stream.emit("drain");
@@ -575,7 +581,7 @@ function _write(data, encoding, cb) {
     if ($isPromise(maybePromise)) {
       maybePromise
         .then(() => {
-          emitDrain(this);
+          emitDrain(this, maybePromise);
           cb(null);
         })
         .catch(cb);
@@ -621,7 +627,7 @@ function underscoreWriteFast(this: FSStream, data: any, encoding: any, cb: any)
       maybePromise.then(
         () => {
           if (cb) cb(null);
-          emitDrain(this);
+          emitDrain(this, maybePromise);
         },
         err => {
           if (cb) cb(err);
@@ -688,7 +694,7 @@ function writeFast(this: FSStream, data: any, encoding: any, cb: any) {
       // mistaken for a write failure.
       maybePromise.then(
         () => {
-          emitDrain(this);
+          emitDrain(this, maybePromise);
           cb(null);
         },
         err => {
@@ -731,7 +737,7 @@ writeStreamPrototype._writev = function (data, cb) {
     if ($isPromise(maybePromise)) {
       maybePromise
         .then(() => {
-          emitDrain(this);
+          emitDrain(this, maybePromise);
           cb(null);
         })
         .catch(cb);

src/runtime/webcore/FileSink.rs

@@ -1498,6 +1498,13 @@ impl crate::webcore::sink::JsSinkType for FileSink {
         // the C++ `JSFileSink` wrapper `js_construct` is about to create.
         this.write(Self::construct());
     }
+    /// Behind the chunks the fd refused, like any other chunk written now.
+    fn write_empty(&mut self) -> streams::Writable {
+        if self.done.get() || !self.writer.get().is_backed_up() {
+            return streams::Writable::Temporary(0);
+        }
+        self.to_result(WriteResult::Pending(0), 0)
+    }
     fn end_from_js(&mut self, cx: &bun_jsc::JsThread<'_>) -> sys::Result<JSValue> {
         Self::end_from_js(self, cx)
     }

src/runtime/webcore/Sink.rs

@@ -371,6 +371,10 @@ pub(crate) trait JsSinkType: Sized + JsSinkAbi {
     fn write_bytes(&mut self, data: &streams::Result) -> streams::result::Writable;
     fn write_utf16(&mut self, data: &streams::Result) -> streams::result::Writable;
     fn write_latin1(&mut self, data: &streams::Result) -> streams::result::Writable;
+    /// `write()` of an empty chunk, which never reaches the `write_*` methods.
+    fn write_empty(&mut self) -> streams::result::Writable {
+        streams::result::Writable::Temporary(0)
+    }
     fn end(&mut self, err: Option<SysError>) -> sys::Result<()>;
     fn end_from_js(&mut self, cx: &bun_jsc::JsThread<'_>) -> sys::Result<JSValue>;
     fn flush(&mut self) -> sys::Result<()>;
@@ -544,7 +548,7 @@ impl<T: JsSinkType> JSSink<T> {
         if let Some(buffer) = arg.as_array_buffer(global) {
             let slice = buffer.slice();
             if slice.is_empty() {
-                return Ok(JSValue::js_number(0.0));
+                return Ok(this.sink.write_empty().to_js(&cx));
             }
             // Borrowed view over GC-kept buffer for the duration of the call.
             let data = bun_ptr::RawSlice::new(slice);
@@ -563,7 +567,7 @@ impl<T: JsSinkType> JSSink<T> {
 
         let view = arg.to_js_string_view(global)?;
         if view.is_empty() {
-            return Ok(JSValue::js_number(0.0));
+            return Ok(this.sink.write_empty().to_js(&cx));
         }
 
         if view.is_utf16() {

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

@@ -273,6 +273,40 @@ it.skipIf(!isPosix)("a backpressured string write() resolves to its encoded byte
   expect(received).toBe(size);
 });
 
+// An empty chunk is behind the chunks queued before it like any other; a number
+// would tell the caller that nothing is waiting.
+it.skipIf(!isPosix).each([
+  ["string", ""],
+  ["Uint8Array", new Uint8Array(0)],
+])("an empty %s write() behind a backpressured write() returns that write's promise", async (_, empty) => {
+  const [readFd, writeFd] = createSocketPair();
+  const sink = Bun.file(writeFd).writer();
+  const size = 4 * 1024 * 1024;
+
+  expect(sink.write(empty)).toBe(0);
+  const first = sink.write(Buffer.alloc(size, 0x61));
+  const second = sink.write(empty);
+
+  let received = 0;
+  const reader = (async () => {
+    for await (const part of Bun.file(readFd).stream()) received += part.byteLength;
+  })();
+
+  try {
+    expect(first).toBeInstanceOf(Promise);
+    expect(second).toBe(first);
+    expect(await second).toBe(size);
+    expect(sink.write(empty)).toBe(0);
+  } finally {
+    await Promise.resolve(sink.end()).catch(() => {});
+    fs.closeSync(writeFd);
+    await reader;
+    fs.closeSync(readFd);
+  }
+
+  expect(received).toBe(size);
+});
+
 // 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
@@ -1234,6 +1268,39 @@ describe("a stream piped into a FileSink on a pipe is pumped to its end", () =>
     }
   });
 
+  // The flush at the end of the tick takes the tail of the short write and resumes the pump from
+  // inside the flush, with the parked write's promise not settled yet. The empty chunk is what
+  // the pump writes then, and nothing would resume a pump that parked on it.
+  it.skipIf(!isPosix).each([
+    ["string", ""],
+    ["ArrayBuffer", new ArrayBuffer(0)],
+  ])("an empty %s chunk behind a short write does not park the pump again", async (_, empty) => {
+    using dir = tempDir("filesink-piped-empty", {});
+    const path = join(String(dir), "piped.fifo");
+    mkfifo(path, 0o666);
+    const fd = fs.openSync(path, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK);
+    try {
+      // A little more than a pipe holds, so the tail fits once the pipe is empty.
+      const short = Buffer.alloc(64 * 1024 + 1000, 97);
+      const source = new ReadableStream({
+        start(c) {
+          c.enqueue(short);
+          c.enqueue(empty);
+          c.enqueue("end");
+          c.close();
+        },
+      });
+      const written = Bun.write(Bun.file(path), new Response(source));
+      const head = fs.readSync(fd, Buffer.alloc(short.length));
+      expect(head).toBeLessThan(short.length);
+      const [tail, total] = await Promise.all([drain(fd), written]);
+      expect(head + tail.length).toBe(short.length + 3);
+      expect(total).toBe(short.length + 3);
+    } finally {
+      fs.closeSync(fd);
+    }
+  });
+
   // With a listener the stream arrives on unfixed code too: the loop goes back to work each
   // time the listener has run. It comes to the edge of its exit after each drain, though,
   // and the listener hears of each one.

test/js/node/process/process-stdio.test.ts

@@ -324,6 +324,44 @@ describe.concurrent.skipIf(isWindows)(
       await expectDrainOverSocket("", 1 << 20);
     });
 
+    // A write's callback says the chunks before it are out, so exiting from it loses nothing. As in Node, the chunks queued
+    // while the pipe is full get one 'drain' between them.
+    test.each([
+      [`""`, 0],
+      [`Buffer.alloc(0)`, 0],
+      [`"B"`, 1],
+    ])(
+      "process.stdout.write(%s, cb) behind a write to a full pipe calls back once both are out",
+      async (chunk, length) => {
+        await using proc = spawn({
+          cmd: [
+            bunExe(),
+            "-e",
+            `const order = [];
+           const report = v => require("fs").writeSync(2, JSON.stringify(v) + "\\n");
+           process.stdout.on("drain", () => order.push("drain"));
+           const first = process.stdout.write(Buffer.alloc(1 << 20, "A"), () => order.push("first"));
+           const second = process.stdout.write(${chunk}, () => {
+             order.push("second");
+             report({ order });
+             process.exit(0);
+           });
+           report({ first, second });`,
+          ],
+          env: bunEnv,
+          stdout: "pipe",
+          stderr: "pipe",
+          lazy: true,
+        });
+        const events = jsonLines(proc.stderr);
+        expect(await events.next()).toEqual({ first: false, second: false });
+        const [stdout, order, exitCode] = await Promise.all([proc.stdout.bytes(), events.next(), proc.exited]);
+        expect(stdout.byteLength).toBe((1 << 20) + length);
+        expect(order).toEqual({ order: ["drain", "first", "second"] });
+        expect(exitCode).toBe(0);
+      },
+    );
+
     // Bun.spawn's "pipe" is a socketpair; these go through sh so fd 1 is a real pipe(2).
     test("console.log and process.stdout.write are not truncated over a real pipe", async () => {
       await using proc = spawn({