console: output goes behind what process.stdout/stderr have buffered, as in Node

src/jsc/ConsoleObject.rs

@@ -270,6 +270,13 @@ unsafe fn vm_console_mut<'a>(global: &JSGlobalObject) -> &'a mut ConsoleObject {
     unsafe { &mut *vm_console(global) }
 }
 
+unsafe extern "Rust" {
+    /// Defined in `bun_runtime::webcore::file_sink`: the sink behind `process.stdout` (1) / `process.stderr` (2) holds bytes a full pipe or socket refused.
+    safe fn __bun_stdio_sink_is_backed_up(global: &JSGlobalObject, fd: i32) -> bool;
+    /// Appends `message` behind those bytes; `false`, with nothing appended, if there are none any more.
+    safe fn __bun_stdio_sink_append(global: &JSGlobalObject, fd: i32, message: &[u8]) -> bool;
+}
+
 static STDERR_MUTEX: Mutex = Mutex::new();
 static STDOUT_MUTEX: Mutex = Mutex::new();
 
@@ -445,19 +452,27 @@ fn message_with_type_and_level_(
     let default_indent = unsafe { vm_console_mut(global) }.default_indent;
 
     // SAFETY: see [`vm_console`] — `console` points at the live boxed
-    // `ConsoleObject` for this VM; JS-thread-only. Kept as a raw deref (not
-    // `vm_console_mut`) so the resulting `writer` borrow does not pin a
-    // long-lived `&mut ConsoleObject` across the re-derive in the empty-`Log`
-    // arm below.
-    let raw_writer: &mut bun_core::io::Writer = unsafe {
+    // `ConsoleObject` for this VM; JS-thread-only.
+    let (raw_writer, fd): (&mut bun_core::io::Writer, i32) = unsafe {
         if matches!(level, MessageLevel::Warning | MessageLevel::Error) {
-            (*console).error_writer()
+            ((*console).error_writer(), 2)
         } else {
-            (*console).writer()
+            ((*console).writer(), 1)
         }
     };
+    // As in Node, a message for a stream that is backed up is formatted first and then goes behind what the stream buffered.
+    let mut behind_sink = None;
     // `bun_core::io::Writer: bun_io::Write` — `&mut Writer` unsize-coerces directly.
-    let writer: &mut dyn bun_io::Write = raw_writer;
+    let writer: &mut dyn bun_io::Write = if __bun_stdio_sink_is_backed_up(global, fd) {
+        &mut **behind_sink.insert(scopeguard::guard(Vec::new(), move |message: Vec<u8>| {
+            if !__bun_stdio_sink_append(global, fd, &message) {
+                let _ = raw_writer.write_all(&message);
+                let _ = raw_writer.flush();
+            }
+        }))
+    } else {
+        raw_writer
+    };
 
     // LAYERING: `Jest::runner()` lives in `bun_runtime::test_runner` (forward
     // dep on the high tier). Dispatch through `RuntimeHooks` instead — the
@@ -541,12 +556,8 @@ fn message_with_type_and_level_(
             print_options,
         )?;
     } else if message_type == MessageType::Log {
-        // SAFETY: see [`vm_console`]. `writer` (above) is dead in this arm —
-        // the only later uses are in the mutually-exclusive `Trace` block, and
-        // `message_type == Log` here.
-        let w = unsafe { (*console).writer() };
-        let _ = w.write_all(b"\n");
-        let _ = w.flush();
+        let _ = writer.write_all(b"\n");
+        let _ = writer.flush();
     } else if message_type != MessageType::Trace {
         let _ = writer.write_all(b"undefined\n");
     }

src/jsc/bindings/BunProcess.cpp

@@ -3030,21 +3030,30 @@ static JSValue constructStdioWriteStream(JSC::JSGlobalObject* globalObject, JSC:
     // Note: files are always sync anyway.
     // forceSync = fdType == BunProcessStdinFdType::file || bun_stdio_tty[fd];
 
-    // TODO: once console.* is wired up to write/read through the same buffering mechanism as FileSink for process.stdout, process.stderr, we can make this non-blocking for sockets on POSIX.
-    // Until then, we have to force it to be sync EVEN for sockets or else console.log() may flush at a different time than process.stdout.write.
+    // Turns off FileSink's coalescing of small writes. What a full pipe or socket refuses is still buffered, and console.* output goes behind it (m_stdioSinks).
     forceSync = true;
 #endif
     if (forceSync) {
         JSValue sink = resultObject->getIndex(globalObject, 1);
         RETURN_IF_EXCEPTION(scope, jsUndefined());
         Bun__ForceFileSinkToBeSynchronousForProcessObjectStdio(globalObject, JSValue::encode(sink));
+        if (auto* sinkObject = sink.getObject())
+            uncheckedDowncast<Process>(processObject)->m_stdioSinks[fd - 1].set(vm, processObject, sinkObject);
     }
 
     JSValue stream = resultObject->getIndex(globalObject, 0);
     RETURN_IF_EXCEPTION(scope, jsUndefined());
     return stream;
 }
 
+extern "C" JSC::EncodedJSValue Bun__Process__getStdioSink(JSC::JSGlobalObject* lexicalGlobalObject, int fd)
+{
+    ASSERT(fd == 1 || fd == 2);
+    auto* globalObject = defaultGlobalObject(lexicalGlobalObject);
+    JSObject* sink = globalObject->hasProcessObject() ? globalObject->processObject()->m_stdioSinks[fd - 1].get() : nullptr;
+    return JSValue::encode(sink ? JSValue(sink) : jsUndefined());
+}
+
 static JSValue constructStdout(VM& vm, JSObject* processObject)
 {
     return constructStdioWriteStream(processObject->globalObject(), processObject, 1);
@@ -3782,6 +3791,8 @@ void Process::visitChildrenImpl(JSCell* cell, Visitor& visitor)
     visitor.append(thisObject->m_argv);
     visitor.append(thisObject->m_execArgv);
     visitor.append(thisObject->m_onWarning);
+    for (auto& sink : thisObject->m_stdioSinks)
+        visitor.append(sink);
 
     thisObject->m_cpuUsageStructure.visit(visitor);
     thisObject->m_resourceUsageStructure.visit(visitor);

src/jsc/bindings/BunProcess.h

@@ -47,6 +47,8 @@ class Process : public WebCore::JSEventEmitter {
 
     DECLARE_EXPORT_INFO;
     bool m_reportOnUncaughtException = false;
+    // The FileSinks behind process.stdout ([0]) and process.stderr ([1]): console.* output goes behind what they have buffered.
+    WriteBarrier<JSObject> m_stdioSinks[2];
 
     static void destroy(JSC::JSCell* cell)
     {

src/runtime/webcore/FileSink.rs

@@ -210,6 +210,34 @@ impl FileSink {
     }
 }
 
+unsafe extern "C" {
+    /// The `FileSink` wrapper behind `process.stdout` (1) / `process.stderr` (2), or `undefined` before that stream exists.
+    safe fn Bun__Process__getStdioSink(global: &JSGlobalObject, fd: i32) -> JSValue;
+}
+
+/// The writer of that sink, while it holds bytes a full pipe or socket refused.
+fn backed_up_stdio_writer(global: &JSGlobalObject, fd: i32) -> Option<&JsCell<IOWriter>> {
+    let this_ptr = JSSink::from_js(Bun__Process__getStdioSink(global, fd))?;
+    // SAFETY: `from_js` returned the wrapper's live `m_ctx`; `global`'s `Process` keeps the wrapper alive.
+    let writer = unsafe { &(*this_ptr).sink.writer };
+    // A writer whose poll failed to register is closed with bytes left, and nothing writes them.
+    #[cfg(unix)]
+    writer.get().get_poll()?;
+    writer.get().has_pending_data().then_some(writer)
+}
+
+#[unsafe(no_mangle)]
+fn __bun_stdio_sink_is_backed_up(global: &JSGlobalObject, fd: i32) -> bool {
+    backed_up_stdio_writer(global, fd).is_some()
+}
+
+/// The poll that is armed for the bytes already there writes `message` after them.
+#[unsafe(no_mangle)]
+fn __bun_stdio_sink_append(global: &JSGlobalObject, fd: i32, message: &[u8]) -> bool {
+    backed_up_stdio_writer(global, fd)
+        .is_some_and(|writer| writer.with_mut(|w| w.outgoing.write(message)).is_ok())
+}
+
 #[unsafe(no_mangle)]
 pub(crate) extern "C" fn Bun__ForceFileSinkToBeSynchronousForProcessObjectStdio(
     _global: *mut JSGlobalObject,

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

@@ -163,9 +163,9 @@ describe.concurrent("process-stdio", () => {
 
 // Like Node, process.stdout/stderr put O_NONBLOCK on fd 1/2 when they are pipes or sockets. The flag lives on the shared open
 // file description, so spawn clears it again for a child's fds 0-2 (libuv does the same), and the native console writer must
-// wait on EAGAIN instead of dropping the rest of the line.
+// wait on EAGAIN instead of dropping the rest of the line, and its output goes behind what the stream has buffered.
 describe.concurrent.skipIf(isWindows)(
-  "O_NONBLOCK on process.stdout/stderr does not leak to children or truncate console output",
+  "process.stdout/stderr on a pipe or socket: O_NONBLOCK does not leak to children, console output is whole and in call order",
   () => {
     const probe = path.join(import.meta.dir, "fd-nonblock-fixture.js");
 
@@ -227,6 +227,17 @@ describe.concurrent.skipIf(isWindows)(
       expect(exitCode).toBe(0);
     }
 
+    async function signalReaderOnceFull(events: ReturnType<typeof jsonLines>) {
+      let readerPid: number | undefined, ret: boolean | undefined;
+      while (readerPid === undefined || ret === undefined) {
+        const v = await events.next();
+        if ("reader" in v) readerPid = v.reader;
+        if ("ret" in v) ret = v.ret;
+      }
+      expect(ret).toBe(false);
+      process.kill(readerPid, "SIGUSR1");
+    }
+
     // Real pipe(2) via sh, reader gated on SIGUSR1.
     async function expectDrainOverPipe(
       prelude: string,
@@ -241,14 +252,7 @@ describe.concurrent.skipIf(isWindows)(
         stderr: "pipe",
       });
       const events = jsonLines(proc.stderr);
-      let readerPid: number | undefined, ret: boolean | undefined;
-      while (readerPid === undefined || ret === undefined) {
-        const v = await events.next();
-        if ("reader" in v) readerPid = v.reader;
-        if ("ret" in v) ret = v.ret;
-      }
-      expect(ret).toBe(false);
-      process.kill(readerPid!, "SIGUSR1");
+      await signalReaderOnceFull(events);
       expect(await events.next()).toEqual(expectedDrain);
       const [stdout, exitCode] = await Promise.all([proc.stdout.bytes(), proc.exited]);
       expect(stdout.byteLength).toBe(expectedBytes);
@@ -371,5 +375,118 @@ describe.concurrent.skipIf(isWindows)(
         await expectDrainOverPipe(afterInheritSpawn, "{}", { drained: true }, 1 + (1 << 20));
       },
     );
+
+    // Each console call and what it prints. Reports go to the other fd; nothing reads the fd under test until the first write() returned false.
+    const stdoutCalls = [
+      [`log("<log>")`, "<log>"],
+      [`info("<info>")`, "<info>"],
+      [`debug("<debug>")`, "<debug>"],
+      [`log()`, ""],
+    ] as const;
+    const stderrCalls = [
+      [`warn("<warn>")`, "<warn>"],
+      [`error("<error>")`, "<error>"],
+      [`error()`, ""],
+    ] as const;
+    type Calls = typeof stdoutCalls | typeof stderrCalls;
+    const callOrderScript = (stream: "stdout" | "stderr", calls: Calls) => `
+      const report = v => require("fs").writeSync(${stream === "stdout" ? 2 : 1}, JSON.stringify(v) + "\\n");
+      [${calls.map(([call]) => `() => console.${call}`).join(", ")}].forEach((call, i) => {
+        const ret = process.${stream}.write(Buffer.alloc(1 << 20, String(i)));
+        if (i === 0) report({ ret });
+        call();
+      });`;
+    const inCallOrder = (calls: Calls) => calls.map(([, printed], i) => `[${i}*${1 << 20}]${printed}\n`).join("");
+    const squash = (bytes: Uint8Array) =>
+      Buffer.from(bytes)
+        .toString()
+        .replace(/(\d)\1+/g, m => `[${m[0]}*${m.length}]`);
+
+    // A pipeline's status is its last command's, so the writer's own goes out with the reports.
+    const stdoutPipeline = `( "$0" -e "$1"; echo "{\\"writer\\":$?}" >&2 ) | "$0" -e "$2"`;
+    const stderrPipeline = `{ ( "$0" -e "$1"; echo "{\\"writer\\":$?}" ) 2>&1 >&3 | "$0" -e "$2"; } 3>&2`;
+    async function expectWriterExitedCleanly(events: ReturnType<typeof jsonLines>) {
+      expect(await events.next()).toEqual({ writer: 0 });
+      expect(await events.rest()).toBe("");
+    }
+
+    test.each([
+      ["stdout", stdoutCalls, stdoutPipeline],
+      ["stderr", stderrCalls, stderrPipeline],
+    ] as const)(
+      "console output on %s comes after what the stream buffered on a full pipe",
+      async (stream, calls, pipeline) => {
+        await using proc = spawn({
+          cmd: ["sh", "-c", pipeline, bunExe(), callOrderScript(stream, calls), gatedReader],
+          env: bunEnv,
+          stdout: "pipe",
+          stderr: "pipe",
+        });
+        const events = jsonLines(proc.stderr);
+        await signalReaderOnceFull(events);
+        const [stdout, exitCode] = await Promise.all([proc.stdout.bytes(), proc.exited]);
+        expect(squash(stdout)).toBe(inCallOrder(calls));
+        await expectWriterExitedCleanly(events);
+        expect(exitCode).toBe(0);
+      },
+    );
+
+    // process.exit() drops what is still buffered, so everything written before the callback has to be out by then.
+    test("the callback of a write() after console output runs once all of it is out", async () => {
+      const script = `
+        require("fs").writeSync(2, JSON.stringify({ ret: process.stdout.write(Buffer.alloc(1 << 20, "0")) }) + "\\n");
+        console.log("<log>");
+        process.stdout.write(Buffer.alloc(1 << 20, "1"), () => process.exit(0));`;
+      await using proc = spawn({
+        cmd: ["sh", "-c", stdoutPipeline, bunExe(), script, gatedReader],
+        env: bunEnv,
+        stdout: "pipe",
+        stderr: "pipe",
+      });
+      const events = jsonLines(proc.stderr);
+      await signalReaderOnceFull(events);
+      const [stdout, exitCode] = await Promise.all([proc.stdout.bytes(), proc.exited]);
+      expect(squash(stdout)).toBe(`[0*${1 << 20}]<log>\n[1*${1 << 20}]`);
+      await expectWriterExitedCleanly(events);
+      expect(exitCode).toBe(0);
+    });
+
+    // The result on main too. write() returns true once nothing is buffered, so by the end of formatting there is nothing to go behind.
+    test("a message whose formatting outlasts the backlog is still printed", async () => {
+      const script = `
+        require("fs").writeSync(2, JSON.stringify({ ret: process.stdout.write(Buffer.alloc(1 << 20, "0")) }) + "\\n");
+        console.log({
+          [Symbol.for("nodejs.util.inspect.custom")]() {
+            while (!process.stdout.write("1"));
+            return "<log>";
+          },
+        });`;
+      await using proc = spawn({
+        cmd: ["sh", "-c", stdoutPipeline, bunExe(), script, gatedReader],
+        env: bunEnv,
+        stdout: "pipe",
+        stderr: "pipe",
+      });
+      const events = jsonLines(proc.stderr);
+      await signalReaderOnceFull(events);
+      const [stdout, exitCode] = await Promise.all([proc.stdout.bytes(), proc.exited]);
+      expect(squash(stdout)).toMatch(new RegExp(String.raw`^\[0\*${1 << 20}\](1|\[1\*\d+\])<log>\n$`));
+      await expectWriterExitedCleanly(events);
+      expect(exitCode).toBe(0);
+    });
+
+    test("console output on stdout comes after what the stream buffered on a full socket", async () => {
+      await using proc = spawn({
+        cmd: [bunExe(), "-e", callOrderScript("stdout", stdoutCalls)],
+        env: bunEnv,
+        stdout: "pipe",
+        stderr: "pipe",
+        lazy: true,
+      });
+      expect(await jsonLines(proc.stderr).next()).toEqual({ ret: false });
+      const [stdout, exitCode] = await Promise.all([proc.stdout.bytes(), proc.exited]);
+      expect(squash(stdout)).toBe(inCallOrder(stdoutCalls));
+      expect(exitCode).toBe(0);
+    });
   },
 );

test/js/web/console/console-log.test.ts

@@ -157,6 +157,18 @@ NamedError: console.error a named error
 `);
 });
 
+it("console.error() and console.warn() with no arguments print their newline on stderr", async () => {
+  await using proc = spawn({
+    cmd: [bunExe(), "-e", "console.error(); console.warn(); console.log();"],
+    env: bunEnv,
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  expect({ stdout, stderr }).toEqual({ stdout: "\n", stderr: "\n\n" });
+  expect(exitCode).toBe(0);
+});
+
 it("console.log with SharedArrayBuffer", () => {
   // console.log(x) === Bun.inspect(x) + "\n" written to stdout.
   expect(Bun.inspect(new ArrayBuffer(0))).toBe("ArrayBuffer(0) []");