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

src/codegen/generate-js2native.ts

@@ -46,6 +46,7 @@ const sourceFiles = readdirRecursiveWithExclusionsAndExtensionsSync(
 const rustIdentifierPaths: Record<string, string> = {
   "bun.rs": "bun.rs",
   "ipc.rs": "runtime/ipc_host.rs",
+  "FileSink.rs": "runtime/webcore/FileSink.rs",
   "Counters.rs": "jsc/Counters.rs",
   "FrameworkRouter.rs": "runtime/bake/FrameworkRouter.rs",
   "JSBundler.rs": "runtime/api/JSBundler.rs",

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/BunBuiltinNames.h

@@ -197,6 +197,7 @@ using namespace JSC;
     macro(writable) \
     macro(writableType) \
     macro(write) \
+    macro(writeAll) \
     macro(writer) \
     macro(written) \
     BUN_ADDITIONAL_BUILTIN_NAMES(macro)

src/js/builtins/ConsoleObject.ts

@@ -119,33 +119,26 @@ interface ConsoleWriter extends Bun.FileSink {
   flush(wait?: boolean): number | Promise<number>;
 }
 
-export function write(this: Console & { $writer: ConsoleWriter | undefined }, input) {
+// The signature of a builtin stays on one line: bundle-functions.ts does not parse a parameter list with braces in it.
+interface ConsoleWithWriter {
+  $writer: ConsoleWriter | undefined;
+  $writeAll: (this: ConsoleWriter, ...chunks: unknown[]) => number | Promise<number>;
+}
+
+export function write(this: ConsoleWithWriter, input) {
   if (!$isObject(this)) throw $ERR_INVALID_THIS("Console");
 
   var writer = $getByIdDirectPrivate(this, "writer");
   if (!writer) {
     var length = $toLength(input?.length ?? 0);
     writer = Bun.stdout.writer({ highWaterMark: length > 65536 ? length : 65536 });
     $putByIdDirectPrivate(this, "writer", writer);
+    // Writes every argument and flushes once, natively: there a failed write is an error value and a backed-up one
+    // is the sink's one pending operation, where here they would be a Promise each to tell apart and to handle.
+    $putByIdDirectPrivate(this, "writeAll", $newRustFunction("FileSink.rs", "consoleWrite", 1));
   }
 
-  // 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;
-  var pending: Promise<number> | undefined;
-  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);
-
-  writer.flush(true);
-  if (pending === undefined) return wrote;
-  if (wrote === 0) return pending;
-  return pending.then(n => wrote + n);
+  return $getByIdDirectPrivate(this, "writeAll").$apply(writer, arguments);
 }
 
 // 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)))

src/runtime/webcore/FileSink.rs

@@ -1412,6 +1412,12 @@ impl FileSink {
 // `Sink.JSSink(@This(), "FileSink")` — generic-fn-returning-type → monomorphized type alias.
 pub(crate) type JSSink = crate::webcore::sink::JSSink<FileSink>;
 
+/// `console.write()` (src/js/builtins/ConsoleObject.ts): `this` is the console's writer, a `Bun.stdout.writer()`.
+#[bun_jsc::host_fn]
+pub(crate) fn console_write(global: &JSGlobalObject, frame: &CallFrame) -> JsResult<JSValue> {
+    JSSink::js_write_all(global, frame)
+}
+
 crate::impl_js_sink_abi!(FileSink, "FileSink");
 
 // `JsSinkType` impl: routes the codegen `FileSink__*` thunks (via
@@ -1464,6 +1470,10 @@ impl crate::webcore::sink::JsSinkType for FileSink {
     fn pending_state_is_pending(&self) -> bool {
         self.pending.get().state == streams::PendingState::Pending
     }
+    fn pending_operation(&mut self) -> Option<*mut streams::WritablePending> {
+        self.pending_state_is_pending()
+            .then(|| self.pending.as_ptr())
+    }
     fn protect_js_wrapper(&mut self, global: &JSGlobalObject, this_value: JSValue) {
         Self::protect_js_wrapper(self, global, this_value)
     }

src/runtime/webcore/Sink.rs

@@ -396,6 +396,10 @@ pub(crate) trait JsSinkType: Sized + JsSinkAbi {
     fn get_pending_error(&mut self) -> Option<JSValue> {
         None
     }
+    /// The operation whose Promise is outstanding, for a sink that settles writes through one.
+    fn pending_operation(&mut self) -> Option<*mut streams::WritablePending> {
+        None
+    }
     fn source(&mut self) -> Option<&mut SourceHandle> {
         None
     }
@@ -449,6 +453,19 @@ pub(crate) trait JsSinkType: Sized + JsSinkAbi {
 // no lut entry and no C++ caller.
 // ──────────────────────────────────────────────────────────────────────────
 
+/// What a flush gave back, for JS: its value, or its error thrown. Outside `JSSink<T>`, which is compiled once
+/// per sink type, because none of this depends on the type.
+fn flushed_to_js(
+    global: &crate::webcore::jsc::JSGlobalObject,
+    flushed: sys::Result<crate::webcore::jsc::JSValue>,
+) -> crate::webcore::jsc::JsResult<crate::webcore::jsc::JSValue> {
+    use bun_sys_jsc::ErrorJsc;
+    match flushed {
+        sys::Result::Ok(value) => Ok(value),
+        sys::Result::Err(err) => Err(global.throw_value(err.to_js(global)?)),
+    }
+}
+
 impl<T: JsSinkType> JSSink<T> {
     /// `JSSink.getThis` — recover `&mut JSSink<T>` from `callframe.this()` or
     /// throw the appropriate detached/cast-failed error.
@@ -520,7 +537,16 @@ impl<T: JsSinkType> JSSink<T> {
             )));
         }
 
-        let arg = frame.argument(0);
+        Ok(Self::write_value(this, global, frame.argument(0))?.to_js(&cx))
+    }
+
+    /// One chunk from JS into the sink: bytes as they are, a string by its encoding.
+    fn write_value(
+        this: &mut JSSink<T>,
+        global: &crate::webcore::jsc::JSGlobalObject,
+        arg: crate::webcore::jsc::JSValue,
+    ) -> crate::webcore::jsc::JsResult<streams::result::Writable> {
+        use streams::result::Writable;
         arg.ensure_still_alive();
         let _keep = bun_jsc::EnsureStillAlive(arg);
 
@@ -534,14 +560,11 @@ 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(Writable::Owned(0));
             }
             // Borrowed view over GC-kept buffer for the duration of the call.
             let data = bun_ptr::RawSlice::new(slice);
-            return Ok(this
-                .sink
-                .write_bytes(&streams::Result::Temporary(data))
-                .to_js(&cx));
+            return Ok(this.sink.write_bytes(&streams::Result::Temporary(data)));
         }
 
         if !arg.is_string() {
@@ -553,24 +576,111 @@ 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(Writable::Owned(0));
         }
 
         if view.is_utf16() {
             let utf16 = view.utf16();
             let bytes: &[u8] = bytemuck::cast_slice(utf16);
             let data = bun_ptr::RawSlice::new(bytes);
-            return Ok(this
-                .sink
-                .write_utf16(&streams::Result::Temporary(data))
-                .to_js(&cx));
+            return Ok(this.sink.write_utf16(&streams::Result::Temporary(data)));
         }
 
         let data = bun_ptr::RawSlice::new(view.latin1());
-        Ok(this
-            .sink
-            .write_latin1(&streams::Result::Temporary(data))
-            .to_js(&cx))
+        Ok(this.sink.write_latin1(&streams::Result::Temporary(data)))
+    }
+
+    /// `console.write(...chunks)`: every argument written in order, one flush, one result.
+    ///
+    /// The sink's results are combined here, where a failed write is still an error value and a backed-up write
+    /// is still the sink's one pending operation. Turned into JS values one write at a time they are a rejected
+    /// Promise each, and a Promise the caller has to recognise as one it already holds.
+    pub(crate) fn js_write_all(
+        global: &crate::webcore::jsc::JSGlobalObject,
+        frame: &crate::webcore::jsc::CallFrame,
+    ) -> crate::webcore::jsc::JsResult<crate::webcore::jsc::JSValue> {
+        use crate::webcore::jsc::JSValue;
+        use streams::result::Writable;
+        let cx = global.js_thread_of_caller(frame);
+        bun_core::mark_binding!();
+        let Some(this) = Self::get_this(global, frame)? else {
+            return Ok(JSValue::js_number(0.0));
+        };
+
+        if let Some(err) = this.sink.get_pending_error() {
+            return Err(global.throw_value(err));
+        }
+
+        let mut wrote: u64 = 0;
+        // The sink's pending operation, its Promise, and what that Promise resolves to as of the last write.
+        let mut pending: Option<(*mut streams::WritablePending, JSValue, u64)> = None;
+        // No argument is one `undefined` chunk, which write_value() rejects as write() does.
+        for i in 0..frame.arguments_count().max(1) as usize {
+            let result = Self::write_value(this, global, frame.argument(i))?;
+            // A write can settle the pending operation on the spot: the reader made room, and this chunk went out
+            // with everything buffered before it. Its Promise has resolved to `consumed`, which this call's total
+            // carries on from as a count. Checked before `to_js` below, which arms the slot for the next operation.
+            if let Some((operation, _, consumed)) = pending {
+                // SAFETY: `operation` is the sink's pending slot, which the sink owns and `this` keeps alive.
+                if unsafe { (*operation).state } != streams::PendingState::Pending {
+                    wrote += consumed;
+                    pending = None;
+                }
+            }
+            match result {
+                // Nothing after it can succeed either. The pending operation, if there is one, is given the error
+                // by the sink itself: this is only reached with nothing pending.
+                Writable::Err(err) => {
+                    // A short chunk before this one is still buffered. Left there, the auto flusher finds the
+                    // broken pipe later and marks the sink finished, and every later write reports nothing at
+                    // all. Flushed here it fails as this write did, which is the error the caller already has.
+                    let _ = Self::flush_sink(this, &cx, true);
+                    return Ok(Writable::Err(err).to_js(&cx));
+                }
+                Writable::Pending(operation) => {
+                    let promise = Writable::Pending(operation).to_js(&cx);
+                    // SAFETY: as above.
+                    pending = Some((operation, promise, unsafe { (*operation).consumed }));
+                }
+                Writable::Owned(n)
+                | Writable::OwnedAndDone(n)
+                | Writable::Temporary(n)
+                | Writable::Backpressure(n) => wrote += n,
+                // A finished sink: nothing written.
+                Writable::Done => {}
+            }
+        }
+
+        let Some((operation, promise, _)) = pending else {
+            let was_pending = this.sink.pending_operation().is_some();
+            let flushed = flushed_to_js(global, Self::flush_sink(this, &cx, true))?;
+            let Some(operation) = this.sink.pending_operation().filter(|_| !was_pending) else {
+                return Ok(JSValue::from(wrote));
+            };
+            // Every chunk was buffered, and the flush could not push the buffer out: the sink is backed up after
+            // all, and `flushed` is the Promise of the operation the flush left pending. The caller gets it. With
+            // a count instead, nobody holds that Promise when the write fails, and the failure is an unhandled
+            // rejection. It resolves to this call's total, which what the flush did push out is already part of.
+            // SAFETY: `operation` is the sink's pending slot, which the sink owns and `this` keeps alive.
+            let operation = unsafe { &mut *operation };
+            operation.consumed = wrote;
+            if let Writable::Owned(total) = &mut operation.result {
+                *total = wrote;
+            }
+            return Ok(flushed);
+        };
+        // The Promise resolves to what the pending operation consumed. The counts are part of this call's total
+        // too: a chunk written before the sink backed up, or a small one buffered beside the operation.
+        // SAFETY: `operation` is the sink's pending slot, which the sink owns and `this` keeps alive. It was
+        // still pending after the last write, and nothing has run since.
+        let operation = unsafe { &mut *operation };
+        operation.consumed += wrote;
+        // `result` is the running total, unless a write failed beside the operation: then it is that error.
+        if let Writable::Owned(total) = &mut operation.result {
+            *total = operation.consumed;
+        }
+        flushed_to_js(global, Self::flush_sink(this, &cx, true))?;
+        Ok(promise)
     }
 
     /// `${abi_name}__flush` host-fn body.
@@ -579,8 +689,6 @@ impl<T: JsSinkType> JSSink<T> {
         frame: &crate::webcore::jsc::CallFrame,
     ) -> crate::webcore::jsc::JsResult<crate::webcore::jsc::JSValue> {
         let cx = global.js_thread_of_caller(frame);
-        use crate::webcore::jsc::JSValue;
-        use bun_sys_jsc::ErrorJsc;
         bun_core::mark_binding!();
 
         let Some(this) = Self::get_this(global, frame)? else {
@@ -591,20 +699,23 @@ impl<T: JsSinkType> JSSink<T> {
             return Err(global.throw_value(err));
         }
 
-        if T::HAS_FLUSH_FROM_JS {
-            let wait = frame.arguments_count() > 0
-                && frame.argument(0).is_boolean()
-                && frame.argument(0).as_boolean();
-            return match this.sink.flush_from_js(&cx, wait) {
-                sys::Result::Ok(value) => Ok(value),
-                sys::Result::Err(err) => Err(global.throw_value(err.to_js(global)?)),
-            };
-        }
+        let wait = frame.arguments_count() > 0
+            && frame.argument(0).is_boolean()
+            && frame.argument(0).as_boolean();
+        flushed_to_js(global, Self::flush_sink(this, &cx, wait))
+    }
 
-        match this.sink.flush() {
-            sys::Result::Ok(()) => Ok(JSValue::UNDEFINED),
-            sys::Result::Err(err) => Err(global.throw_value(err.to_js(global)?)),
+    fn flush_sink(
+        this: &mut JSSink<T>,
+        cx: &bun_jsc::JsThread<'_>,
+        wait: bool,
+    ) -> sys::Result<crate::webcore::jsc::JSValue> {
+        if T::HAS_FLUSH_FROM_JS {
+            return this.sink.flush_from_js(cx, wait);
         }
+        this.sink
+            .flush()
+            .map(|()| crate::webcore::jsc::JSValue::UNDEFINED)
     }
 
     /// `${abi_name}__start` host-fn body.