console.write: no unhandled rejections from a broken pipe; FileSink write counts

src/io/PipeWriter.rs

@@ -854,11 +854,15 @@ impl<Parent: PosixStreamingWriterParent> PosixStreamingWriter<Parent> {
             return WriteResult::Wrote(buf_len);
         }
 
-        self.try_write_newly_buffered_data()
+        self.try_write_newly_buffered_data(buf_len)
     }
 
-    fn try_write_newly_buffered_data(&mut self) -> WriteResult {
+    /// Writes `outgoing`, whose last `buf_len` bytes are the chunk being written now. The count in the result
+    /// is that chunk's: the bytes before it are earlier chunks, and each was reported when it was buffered.
+    fn try_write_newly_buffered_data(&mut self, buf_len: usize) -> WriteResult {
         debug_assert!(!self.is_done);
+        // The buffer goes out in order, so the first `earlier` bytes to reach the fd are not this chunk's.
+        let earlier = self.outgoing.size() - buf_len;
 
         // Borrow `self.outgoing` only for the syscall. `try_write` takes `&self`
         // so the shared borrow of `outgoing.slice()` is sound and ends before
@@ -873,27 +877,29 @@ impl<Parent: PosixStreamingWriterParent> PosixStreamingWriter<Parent> {
                 if amt == self.outgoing.size() {
                     self.outgoing.reset();
                     self.parent_on_write(amt, WriteStatus::Drained);
+                    WriteResult::Wrote(amt.saturating_sub(earlier))
                 } else {
                     self.outgoing.wrote(amt);
                     self.parent_on_write(amt, WriteStatus::Pending);
                     Self::register_poll(self);
-                    return WriteResult::Pending(amt);
+                    WriteResult::Pending(amt)
                 }
             }
             WriteResult::Done(amt) => {
                 self.outgoing.reset();
                 self.parent_on_write(amt, WriteStatus::EndOfFile);
+                WriteResult::Done(amt.saturating_sub(earlier))
             }
+            // Not this chunk's count: the parent works out what was accepted from the buffer's size.
             WriteResult::Pending(amt) => {
                 self.outgoing.wrote(amt);
                 self.parent_on_write(amt, WriteStatus::Pending);
                 Self::register_poll(self);
+                WriteResult::Pending(amt)
             }
 
-            WriteResult::Err(e) => return WriteResult::Err(e),
+            WriteResult::Err(e) => WriteResult::Err(e),
         }
-
-        rc
     }
 
     pub fn write(&mut self, buf: &[u8]) -> WriteResult {
@@ -924,7 +930,7 @@ impl<Parent: PosixStreamingWriterParent> PosixStreamingWriter<Parent> {
                 return WriteResult::Err(sys::Error::oom());
             }
 
-            return self.try_write_newly_buffered_data();
+            return self.try_write_newly_buffered_data(buf.len());
         }
 
         let rc = self.try_write(self.force_sync, buf);

src/js/builtins/ConsoleObject.ts

@@ -129,23 +129,54 @@ export function write(this: Console & { $writer: ConsoleWriter | undefined }, in
     $putByIdDirectPrivate(this, "writer", writer);
   }
 
+  // What is not a Promise is added up as it always was: the first result, then `+=` each later one. (A finished
+  // sink's write() is `true`, so two of them make 2.)
+  //
   // A backed-up writer (FileSink) returns a Promise instead of a count: one Promise, the same for every write made
-  // while it is backed up, of the bytes those writes added. The caller gets a Promise of the total then, because
-  // awaiting it waits for the drain and is where a write error (EPIPE from a reader that hung up) arrives.
-  var wrote = 0;
+  // while it is backed up, of the bytes those writes added. `+` made a string of it. The caller gets a Promise of
+  // the total, because awaiting it waits for the drain and is where a write error (EPIPE from a reader that hung
+  // up) arrives.
+  //
+  // A write that fails on the spot (the pipe is already broken) returns a Promise of its own instead, already
+  // rejected. The sink's Promise is never that: it is still pending whenever the sink hands it out. The caller gets
+  // the last Promise, so a failed write's is lost when a later write returns another, or when a later argument or
+  // the flush throws. Nobody can handle its rejection then, so it is marked handled. The sink's Promise never is:
+  // the sink hands it out again, to callers whose rejection has to be reported.
+  //
+  // An earlier Promise of the sink's has settled by the time a write returns a different one, since the sink has
+  // one pending write at a time. Fulfilled (a short write beside it can settle it early), its bytes join the total.
+  var wrote: number | undefined;
   var pending: Promise<number> | undefined;
+  var failedOnTheSpot = false;
   const count = $argumentCount();
   var i = 0;
-  do {
-    const result = writer.write(arguments[i]);
-    if (typeof result === "number") wrote += result;
-    else pending = result;
-  } while (++i < count);
+  try {
+    do {
+      var result: number | Promise<number> = writer.write(arguments[i]);
+      if ($isPromise<number>(result)) {
+        if (pending === result) continue;
+        const earlier = pending;
+        const earlierFailedOnTheSpot = failedOnTheSpot;
+        pending = result;
+        failedOnTheSpot = $isPromiseRejected(result);
+        if (earlier === undefined) continue;
+        if (earlierFailedOnTheSpot) $pokePromiseAsHandled(earlier);
+        if (!$isPromiseFulfilled(earlier)) continue;
+        result = $peekPromiseSettledValue(earlier)!;
+      }
+      wrote = wrote === undefined ? result : wrote + result;
+    } while (++i < count);
+
+    writer.flush(true);
+  } catch (e) {
+    if (failedOnTheSpot) $pokePromiseAsHandled(pending!);
+    throw e;
+  }
 
-  writer.flush(true);
   if (pending === undefined) return wrote;
-  if (wrote === 0) return pending;
-  return pending.then(n => wrote + n);
+  if (wrote === undefined || wrote === 0) return pending;
+  const counted = wrote;
+  return pending.$then(n => counted + n);
 }
 
 // This is the `console.Console` constructor. It is mostly copied from Node.

src/runtime/api/bun/Terminal.rs

@@ -1496,9 +1496,9 @@ impl Terminal {
 
         // StreamingWriter::write() buffers any bytes it couldn't flush
         // synchronously, so the full input has been accepted on every non-error
-        // return. The per-arm counts are sync-flushed bytes (and on a buffered
-        // writer can even exceed `input_len` when prior data drains), so
-        // returning them would make callers re-send an already-queued tail.
+        // return. A `Pending` or partial count is the bytes that reached the fd,
+        // not the bytes accepted, so returning it would make callers re-send an
+        // already-queued tail.
         match write_result {
             bun_io::WriteResult::Err(err) => {
                 Err(global_object.throw_value(err.to_js(global_object)))

test/js/bun/console/console-write.test.ts

@@ -1,5 +1,8 @@
 import { expect, test } from "bun:test";
-import { bunEnv, bunExe, isWindows } from "harness";
+import { bunEnv, bunExe, isLinux, isWindows, libcPathForDlopen, tempDir } from "harness";
+import { mkfifo } from "mkfifo";
+import { closeSync, constants, openSync, readSync } from "node:fs";
+import { join } from "node:path";
 
 test("console.write rejects a non-object this", async () => {
   await using proc = Bun.spawn({
@@ -103,6 +106,9 @@ console.error(JSON.stringify({ one, two }));
 
 // The Promise a backed-up console.write() returns is where its write error arrives: a script that
 // awaits it can handle a reader that hung up, instead of dying with an unhandled rejection.
+//
+// Not on Windows: a write to a pipe whose reader is alive is accepted whole there and returns its byte
+// count, so whether this script sees a Promise at all depends on whether the reader is already gone.
 test.skipIf(isWindows).each([
   ["one argument", ""],
   ["several arguments", ', "tail"'],
@@ -134,3 +140,199 @@ try {
   expect(stderr).toBe("caught EPIPE\n");
   expect(exitCode).toBe(0);
 });
+
+// Every write to a pipe that is already broken fails on the spot, with a rejected Promise of its own.
+// console.write() kept the last one, and lost that too when a short argument's flush threw or a later argument
+// was not something to write. A script that awaited it and caught the error still died of the unhandled rejection
+// of a Promise it was never given.
+//
+// The child blocks in a synchronous read of stdin until the parent has closed its end of stdout, and checks with
+// a writer of its own that the pipe really is broken, so the test cannot pass by the writes simply being queued.
+//
+// A later console.write() to the broken pipe fails too.
+const big = `Buffer.alloc(1024 * 1024, "a").toString()`;
+test.concurrent.each([
+  ["two large arguments", `${big}, ${big}`, "EPIPE"],
+  ["a large argument, then a short one", `${big}, "\\n"`, "EPIPE"],
+  ["a short argument, then a large one", `"\\n", ${big}`, "EPIPE"],
+  ["a large argument, then one that is not something to write", `${big}, 123`, "ERR_INVALID_ARG_TYPE"],
+])("an awaited console.write to a broken pipe fails once: %s", async (_label, args, code) => {
+  await using proc = Bun.spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `
+process.on("unhandledRejection", e => {
+  console.error("unhandledRejection " + e?.code);
+});
+require("node:fs").readSync(0, Buffer.alloc(1));
+const probe = Bun.stdout.writer().write(${big});
+console.error("pipe already broken: " + (probe instanceof Promise && Bun.peek.status(probe) === "rejected"));
+if (probe instanceof Promise) probe.catch(() => {});
+try {
+  await console.write(${args});
+  console.error("resolved");
+} catch (e) {
+  console.error("caught " + e.code);
+}
+await new Promise(resolve => setImmediate(resolve));
+try {
+  console.error("later: " + (await console.write("later")));
+} catch (e) {
+  console.error("later: caught " + e.code);
+}
+`,
+    ],
+    env: bunEnv,
+    stdin: "pipe",
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+
+  await proc.stdout.cancel();
+  proc.stdin.write("x");
+  await proc.stdin.end();
+
+  const [stderr, exitCode] = await Promise.all([proc.stderr.text(), proc.exited]);
+  expect(stderr).toBe(`pipe already broken: true\ncaught ${code}\nlater: caught EPIPE\n`);
+  expect(exitCode).toBe(0);
+});
+
+// console.write(a, "s", b): `a` overflows the pipe by a little and is pending; the short "s" beside it settles that
+// write's Promise early, and `b` then starts a pending write of its own. Only the last Promise was kept, so the
+// call resolved to a total without `a`.
+//
+// Linux, because the first argument has to be sized from the pipe's capacity (F_GETPIPE_SZ). stdout is a FIFO that
+// this test does not read until the child has made the call.
+test.concurrent.skipIf(!isLinux)("console.write counts a write that settled before the call was over", async () => {
+  using dir = tempDir("console-write-total", {});
+  const fifo = join(String(dir), "stdout.fifo");
+  mkfifo(fifo, 0o600);
+  const readEnd = openSync(fifo, constants.O_RDONLY | constants.O_NONBLOCK);
+  const writeEnd = openSync(fifo, constants.O_WRONLY);
+  try {
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `
+const { dlopen } = require("bun:ffi");
+const F_GETPIPE_SZ = 1032;
+const libc = ${JSON.stringify(libcPathForDlopen())};
+const capacity = dlopen(libc, { fcntl: { args: ["int", "int"], returns: "int" } }).symbols.fcntl(1, F_GETPIPE_SZ);
+const a = Buffer.alloc(capacity + 100, "a").toString();
+const b = Buffer.alloc(1024 * 1024, "b").toString();
+const total = console.write(a, "s", b);
+console.error("expected " + (a.length + 1 + b.length));
+console.error("resolved " + (await total));
+`,
+      ],
+      env: bunEnv,
+      stdout: writeEnd,
+      stderr: "pipe",
+    });
+    closeSync(writeEnd);
+
+    const reader = proc.stderr.getReader();
+    const decoder = new TextDecoder();
+    let stderr = "";
+    while (!stderr.includes("\n")) {
+      const { value, done } = await reader.read();
+      if (done) break;
+      stderr += decoder.decode(value, { stream: true });
+    }
+    const expected = Number(/expected (\d+)/.exec(stderr)?.[1]);
+
+    const chunk = Buffer.alloc(1024 * 1024);
+    let arrived = 0;
+    while (arrived < expected) {
+      try {
+        const n = readSync(readEnd, chunk);
+        if (n === 0) break;
+        arrived += n;
+      } catch (e: any) {
+        if (e.code !== "EAGAIN") throw e;
+        await new Promise(resolve => setImmediate(resolve));
+      }
+    }
+    for (;;) {
+      const { value, done } = await reader.read();
+      if (done) break;
+      stderr += decoder.decode(value, { stream: true });
+    }
+
+    expect({ stderr, arrived }).toEqual({ stderr: `expected ${expected}\nresolved ${expected}\n`, arrived: expected });
+    expect(await proc.exited).toBe(0);
+  } finally {
+    closeSync(readEnd);
+  }
+});
+
+// What is not a Promise is added up as released: the first write's result, then `+=` each later one. A sink that
+// has finished after a failure returns `true` from write(), so one argument gives `true` and two give 2. Not a
+// count of anything, but it is what these calls return, and scripts that keep writing after their reader has gone
+// see it.
+//
+// stdout is a FIFO whose read end this test holds open without reading, then closes. The child fills the pipe and
+// leaves one short chunk buffered, whose flush cannot finish. When the read end goes, the sink finishes, and the
+// one sign of it a script gets is the unhandled rejection of that flush's Promise, which console.write() dropped.
+// (A script that writes before the event loop has seen the hang-up gets EPIPE thrown from every call instead.)
+test.concurrent.skipIf(isWindows)("console.write to a finished sink adds up what write() returns", async () => {
+  using dir = tempDir("console-write-finished", {});
+  const fifo = join(String(dir), "stdout.fifo");
+  mkfifo(fifo, 0o600);
+  const readEnd = openSync(fifo, constants.O_RDONLY | constants.O_NONBLOCK);
+  const writeEnd = openSync(fifo, constants.O_WRONLY);
+  let readEndOpen = true;
+  try {
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `
+const fs = require("node:fs");
+const finished = new Promise(resolve => process.on("unhandledRejection", resolve));
+console.write("0");
+for (const size of [1024, 1]) {
+  try {
+    for (;;) fs.writeSync(1, Buffer.alloc(size, "f"));
+  } catch {}
+}
+console.write("short");
+console.error("READY");
+fs.readSync(0, Buffer.alloc(1));
+await finished;
+console.error(JSON.stringify([console.write("x"), console.write("x", "y"), console.write("x", "y", "z"), console.write(""), console.write("", "x")]));
+`,
+      ],
+      env: bunEnv,
+      stdin: "pipe",
+      stdout: writeEnd,
+      stderr: "pipe",
+    });
+    closeSync(writeEnd);
+
+    const reader = proc.stderr.getReader();
+    const decoder = new TextDecoder();
+    let stderr = "";
+    while (!stderr.includes("READY")) {
+      const { value, done } = await reader.read();
+      if (done) break;
+      stderr += decoder.decode(value, { stream: true });
+    }
+    closeSync(readEnd);
+    readEndOpen = false;
+    proc.stdin.write("x");
+    await proc.stdin.end();
+    for (;;) {
+      const { value, done } = await reader.read();
+      if (done) break;
+      stderr += decoder.decode(value, { stream: true });
+    }
+
+    expect(stderr).toBe("READY\n" + JSON.stringify([true, 2, 3, 0, 1]) + "\n");
+    expect(await proc.exited).toBe(0);
+  } finally {
+    if (readEndOpen) closeSync(readEnd);
+  }
+});