bun test --parallel: print a test's error output before its result line

src/runtime/cli/test/parallel/Coordinator.rs

@@ -56,7 +56,6 @@ pub struct Coordinator<'a> {
     pub(crate) parallel_limit: u32,
     pub(crate) scale_up_after_ms: i64,
     pub(crate) bail: u32,
-    pub(crate) dots: bool,
     pub(crate) files_done: u32,
     pub(crate) spawned_count: u32,
     pub(crate) live_workers: u32,
@@ -408,19 +407,114 @@ impl<'a> Coordinator<'a> {
         }
     }
 
+    /// Prints `captured[..end]` under the in-flight file's header.
+    /// `dot_len` is the length of a trailing dot (no newline) that joins the
+    /// dots line instead of opening a header of its own.
+    fn print_captured(&mut self, w: &mut Worker, end: usize, dot_len: usize) {
+        if end == 0 {
+            return;
+        }
+        if end > dot_len {
+            // A worker that printed dots itself ends its dots line before
+            // other output, as the serial reporter does.
+            if w.captured[0] == b'\n' {
+                self.last_printed_dot = false;
+            }
+            self.break_dots();
+            if let Some(idx) = w.inflight {
+                self.ensure_header(idx);
+            }
+        }
+        let _ = Output::error_writer().write_all(&w.captured[..end]);
+        w.captured.drain(..end);
+        self.last_printed_dot = dot_len > 0;
+        Output::flush();
+    }
+
+    /// Reads what the worker's pipes hold right now, then prints captured
+    /// output through each result line that has arrived, in frame order.
+    /// Bytes past the last printed line stay captured: they belong to a test
+    /// whose result line is not here yet.
+    fn flush_captured_lines(&mut self, w: &mut Worker) {
+        // The worker writes a test's output and its result line before it
+        // sends the frame, so on POSIX a synchronous drain sees all of it.
+        // Windows reads complete through libuv only; `on_worker_output`
+        // prints the lines when the chunk that carries them arrives.
+        #[cfg(unix)]
+        {
+            let wp: *mut Worker = w;
+            // SAFETY: `wp` is the live worker slot. `read` dispatches
+            // `on_read_chunk`, which appends to `captured` through the pipe's
+            // worker backref; `w` is not touched until it returns (the same
+            // backref caveat as the `Worker::coord` field doc). `draining`
+            // keeps that dispatch from calling back into the coordinator.
+            unsafe {
+                (*wp).draining = true;
+                for pipe in [&raw mut (*wp).out, &raw mut (*wp).err] {
+                    if !(*pipe).done && (*pipe).reader.get_fd() != bun_sys::Fd::INVALID {
+                        bun_io::BufferedReader::read(&raw mut (*pipe).reader);
+                    }
+                }
+                (*wp).draining = false;
+            }
+        }
+        self.on_worker_output(w);
+    }
+
+    /// A chunk of worker stdout/stderr arrived: print through any result
+    /// line it completed.
+    pub(crate) fn on_worker_output(&mut self, w: &mut Worker) {
+        while let Some(line) = w.pending_lines.front() {
+            let Some(pos) = strings::index_of(&w.captured, line) else {
+                break;
+            };
+            let end = pos + line.len();
+            let dot_len = if strings::ends_with_char(line, b'\n') {
+                0
+            } else {
+                line.len()
+            };
+            w.pending_lines.pop_front();
+            self.print_captured(w, end, dot_len);
+        }
+    }
+
+    /// File end: prints through the last result line and what follows it
+    /// (hook output). On POSIX the drain in `flush_captured_lines` has read
+    /// everything the worker wrote for this file, so a line still pending
+    /// never reached the pipe and prints from the frame. On Windows reads
+    /// complete through libuv only, so such a line stays pending and prints
+    /// with the chunk that carries it.
+    fn flush_file_end(&mut self, w: &mut Worker) {
+        if cfg!(unix) {
+            self.flush_captured(w);
+            return;
+        }
+        self.flush_captured_lines(w);
+        if w.pending_lines.is_empty() {
+            self.flush_rest(w);
+        }
+    }
+
+    /// Prints everything the worker wrote. Result lines that never showed up
+    /// in the stream are printed from the frames so no result is lost.
     fn flush_captured(&mut self, w: &mut Worker) {
+        self.flush_captured_lines(w);
+        for line in w.pending_lines.drain(..) {
+            w.captured.extend_from_slice(&line);
+        }
+        self.flush_rest(w);
+    }
+
+    fn flush_rest(&mut self, w: &mut Worker) {
         if w.captured.is_empty() {
             return;
         }
-        self.break_dots();
-        if let Some(idx) = w.inflight {
-            self.ensure_header(idx);
-        }
-        let _ = Output::error_writer().write_all(&w.captured);
         if !strings::ends_with_char(&w.captured, b'\n') {
-            let _ = Output::error_writer().write_all(b"\n");
+            w.captured.push(b'\n');
         }
-        w.captured.clear();
+        let end = w.captured.len();
+        self.print_captured(w, end, 0);
     }
 
     pub(crate) fn on_frame(&mut self, w: &mut Worker, kind: frame::Kind, rd: &mut frame::Reader) {
@@ -442,20 +536,13 @@ impl<'a> Coordinator<'a> {
                 if let Some(file) = self.test_records.get_mut(idx as usize) {
                     file.tests.push(Box::from(rd.p));
                 }
-                self.flush_captured(w);
-                if formatted.is_empty() {
-                    return; // e.g. pass under --only-failures
-                }
-                // dots-mode failures print a full line (writeTestStatusLine);
-                // dots themselves are unterminated.
-                let is_dot = self.dots && !strings::ends_with_char(formatted, b'\n');
-                if !is_dot {
-                    self.break_dots();
-                    self.ensure_header(idx);
+                // The worker wrote `formatted` to its stderr before it sent
+                // this frame. Empty under --only-failures for a pass: that
+                // test's output waits for the next line or the file end.
+                if !formatted.is_empty() {
+                    w.pending_lines.push_back(Box::from(formatted));
                 }
-                let _ = Output::error_writer().write_all(formatted);
-                self.last_printed_dot = is_dot;
-                Output::flush();
+                self.flush_captured_lines(w);
             }
             frame::Kind::FileDone => {
                 let mut nums = [0u32; 9];
@@ -477,7 +564,7 @@ impl<'a> Coordinator<'a> {
                     file.elapsed_ns = rd.u64();
                 }
 
-                self.flush_captured(w);
+                self.flush_file_end(w);
                 if self.last_header_idx == Some(idx) {
                     self.end_group();
                 }
@@ -728,6 +815,7 @@ impl<'a> Coordinator<'a> {
             w.out = WorkerPipe::new(core::ptr::null());
             w.err = WorkerPipe::new(core::ptr::null());
             let _ = core::mem::take(&mut w.captured);
+            w.pending_lines.clear();
         }
     }
 

src/runtime/cli/test/parallel/Worker.rs

@@ -4,6 +4,7 @@
 //! spawn/dispatch/shutdown mechanics.
 
 use core::ffi::c_void;
+use std::collections::VecDeque;
 
 #[cfg(unix)]
 use crate::api::bun::process::PosixStdio as Stdio;
@@ -61,9 +62,17 @@ pub struct Worker {
     /// Millisecond timestamp at the most recent dispatch; drives lazy
     /// scale-up.
     pub(crate) dispatched_at: i64,
-    /// Worker stdout+stderr since the last `test_done`. Flushed atomically
-    /// under the right file header so concurrent files don't interleave.
+    /// Worker stdout+stderr not yet printed. Printed under the right file
+    /// header so concurrent files don't interleave.
     pub(crate) captured: Vec<u8>,
+    /// Result lines from `test_done` frames whose copy in `captured` has not
+    /// been printed yet, in frame order. The worker writes each line to its
+    /// stderr right before it sends the frame, so the line marks where the
+    /// test's output ends in the byte stream.
+    pub(crate) pending_lines: VecDeque<Box<[u8]>>,
+    /// Set while the coordinator drains the pipes itself, so the read
+    /// callback does not re-enter it.
+    pub(crate) draining: bool,
     pub(crate) alive: bool,
     /// Set when the process-exit notification arrives. Reaping waits for both
     /// this and `ipc.done` so trailing IPC frames are decoded first.
@@ -406,17 +415,30 @@ impl WorkerPipe {
         }
     }
 
-    pub(crate) fn on_read_chunk(&mut self, chunk: &[u8], _: bun_io::ReadState) -> bool {
+    /// Raw receiver: the coordinator call below forms a `&mut Worker`, which
+    /// contains this pipe, so no `&mut WorkerPipe` may be live across it.
+    ///
+    /// # Safety
+    /// `this` is the live pipe, embedded in its worker.
+    pub(crate) unsafe fn on_read_chunk(this: *mut Self, chunk: &[u8], _: bun_io::ReadState) -> bool {
         // SAFETY: worker backref valid while WorkerPipe is embedded in Worker.
-        // Mutating `captured` through cast_mut requires write provenance on
-        // the stored pointer; all backref creation sites (the runner.rs
-        // coord_ptr, the Worker.rs start() errdefer guard, and the
-        // Coordinator.rs spawn_worker/respawn sites via
-        // `std::ptr::from_mut(..).cast_const()`) now establish it. The
-        // residual `&mut Coordinator`-during-drive aliasing caveat described
-        // in the `Worker::coord` field doc applies to this backref too. No
-        // other reference to `captured` is live during the read callback.
-        unsafe { (*self.worker.cast_mut()).captured.extend_from_slice(chunk) };
+        // Mutating through cast_mut requires write provenance on the stored
+        // pointer; all backref creation sites (the runner.rs coord_ptr, the
+        // Worker.rs start() errdefer guard, and the Coordinator.rs
+        // spawn_worker/respawn sites via `std::ptr::from_mut(..).cast_const()`)
+        // establish it. The residual `&mut Coordinator`-during-drive aliasing
+        // caveat described in the `Worker::coord` field doc applies to this
+        // backref too. No other reference to the worker is live during the
+        // read callback, except while the coordinator drains the pipe itself
+        // from `flush_captured_lines`: `draining` is set then and the call
+        // into the coordinator is skipped.
+        unsafe {
+            let w = (*this).worker.cast_mut();
+            (*w).captured.extend_from_slice(chunk);
+            if !(*w).draining && !(*w).pending_lines.is_empty() {
+                (*(*w).coord.cast_mut()).on_worker_output(&mut *w);
+            }
+        }
         true
     }
     pub(crate) fn on_reader_done(&mut self) {
@@ -433,7 +455,7 @@ impl WorkerPipe {
 bun_io::impl_buffered_reader_parent! {
     TestParallelWorkerPipe for WorkerPipe;
     has_on_read_chunk = true;
-    on_read_chunk   = |this, chunk, state| (*this).on_read_chunk(&chunk, state);
+    on_read_chunk   = |this, chunk, state| WorkerPipe::on_read_chunk(this, &chunk, state);
     on_reader_done  = |this| (*this).on_reader_done();
     on_reader_error = |this, err| (*this).on_reader_error(err);
     // `vm.uv_loop()` is `*mut bun_io::Loop` on every target.

src/runtime/cli/test/parallel/runner.rs

@@ -129,6 +129,8 @@ pub(crate) fn run_as_coordinator(
             inflight: None,
             dispatched_at: 0,
             captured: Vec::new(),
+            pending_lines: std::collections::VecDeque::new(),
+            draining: false,
             alive: false,
             exit_status: None,
             reap_pending: false,
@@ -172,7 +174,6 @@ pub(crate) fn run_as_coordinator(
             DEFAULT_SCALE_UP_AFTER_MS
         },
         bail: ctx.test_options.bail,
-        dots: ctx.test_options.reporters.dots,
         test_records: if ctx.test_options.reporters.junit {
             (0..n).map(|_| Default::default()).collect()
         } else {

src/runtime/cli/test_command.rs

@@ -1357,6 +1357,10 @@ impl CommandLineReporter {
             Self::test_case_report(result, buntest, sequence, test_entry, elapsed_ns, failure)
         });
         if let Some(idx) = this.worker_ipc_file_idx {
+            // The line goes into the worker's own stderr so it is sequenced
+            // with the error output that precedes it. The coordinator prints
+            // captured stderr up to this line when the frame arrives.
+            let _ = Output::error_writer().write_all(formatted_line);
             ParallelRunner::worker_emit_test_done(idx, formatted_line, report.as_ref());
         } else {
             let _ = Output::error_writer().write_all(formatted_line);

test/cli/test/parallel.test.ts

@@ -784,6 +784,60 @@ test("--parallel never interleaves console output across files", async () => {
   expect(exitCode).toBe(0);
 });
 
+test("--parallel keeps each failure's error output and GitHub annotation with its (fail) line", async () => {
+  // Many failures with a long message per file. The error output rides the
+  // worker's stderr pipe and the result line rides the IPC channel, so the
+  // coordinator must print the stderr bytes that precede each result line
+  // before that line, and nothing past it.
+  const body = (tag: string) =>
+    `import {test} from "bun:test";
+     const big = Buffer.alloc(3000, "X").toString();
+     for (let i = 0; i < 60; i++) test("${tag}" + i, async () => {
+       if (i % 7 == 0) await Bun.sleep(1);
+       throw new Error("E-${tag}" + i + "\\n" + big + "\\nend");
+     });`;
+  using dir = tempDir("parallel-error-order", {
+    "a.test.js": body("a"),
+    "b.test.js": body("b"),
+    "c.test.js": body("c"),
+  });
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "test", "--parallel=3"],
+    env: { ...bunEnv, GITHUB_ACTIONS: "true", BUN_TEST_PARALLEL_SCALE_MS: "0" },
+    cwd: String(dir),
+    stderr: "pipe",
+    stdout: "pipe",
+  });
+  const [stderr, exitCode] = await Promise.all([proc.stderr.text(), proc.exited]);
+
+  // Every annotation is one intact line that names its test.
+  const annotations = stderr.split("\n").filter(line => line.startsWith("::error "));
+  expect(annotations.length).toBe(180);
+  for (const line of annotations) {
+    expect(line).toMatch(/^::error file=[abc]\.test\.js,line=\d+,col=\d+,title=error: E-[abc]\d+::end%0A/);
+  }
+  // The code frame for a test is followed by that test's own (fail) line.
+  let pendingError: string | null = null;
+  let checked = 0;
+  for (const line of stderr.split("\n")) {
+    const err = line.match(/^error: (E-[abc]\d+)$/);
+    if (err) {
+      expect(pendingError).toBeNull();
+      pendingError = err[1];
+      continue;
+    }
+    const fail = line.match(/^\(fail\) ([abc]\d+)/);
+    if (fail) {
+      expect(pendingError).toBe("E-" + fail[1]);
+      pendingError = null;
+      checked++;
+    }
+  }
+  expect(checked).toBe(180);
+  expect(stderr).toContain("180 fail");
+  expect(exitCode).toBe(1);
+});
+
 test("--parallel lazily scales workers based on file duration", async () => {
   // Each test file appends its PID so we can count distinct worker processes.
   const body = (sleepMs: number) =>