bun test --parallel: send each earlier failed attempt of a retried test in its own frame

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

@@ -178,6 +178,8 @@ impl ChildrenOfWorkers {
 pub(crate) struct FileTestRecords {
     /// `TestDone` payloads past the formatted line; see `runner::decode_test_case`.
     pub(crate) tests: Vec<Box<[u8]>>,
+    /// `TestAttempt` payloads past the file index, each with the index in `tests` of the `TestDone` that follows it.
+    pub(crate) attempts: Vec<(usize, Box<[u8]>)>,
     pub(crate) elapsed_ns: u64,
     /// Unix time in milliseconds, or 0 if the file never started.
     pub(crate) started_unix_ms: i64,
@@ -726,6 +728,15 @@ impl<'a> Coordinator<'a> {
                 self.last_printed_dot = is_dot;
                 Output::flush();
             }
+            frame::Kind::TestAttempt => {
+                let idx = rd.u32();
+                if w.inflight != Some(idx) {
+                    return;
+                }
+                if let Some(file) = self.test_records.get_mut(idx as usize) {
+                    file.attempts.push((file.tests.len(), Box::from(rd.p)));
+                }
+            }
             frame::Kind::FileDone => {
                 let mut nums = [0u32; 10];
                 for n in nums.iter_mut() {

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

@@ -34,6 +34,8 @@ pub(crate) enum Kind {
     CoverageFile,
     /// str line — one write to a read-only built-in that failed in a file that ran compiled, sent ahead of the file's FileDone
     RefusedChange,
+    /// u32 file_idx, then one failure as in TestDone: an earlier failed attempt of the retried test that the file's next TestDone reports
+    TestAttempt,
 }
 
 impl TryFrom<u8> for Kind {
@@ -50,6 +52,7 @@ impl TryFrom<u8> for Kind {
             6 => Kind::Shutdown,
             7 => Kind::CoverageFile,
             8 => Kind::RefusedChange,
+            9 => Kind::TestAttempt,
             _ => return Err(()),
         })
     }

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

@@ -20,15 +20,21 @@ pub(crate) fn replay_test_records(coord: &mut Coordinator) {
     let Some(junit) = coord.reporter.reporters.junit.as_deref_mut() else {
         return;
     };
-    for (idx, file) in files.iter().enumerate() {
+    for (idx, file) in files.into_iter().enumerate() {
         let rel = junit_file_name(coord.files[idx].as_bytes());
         junit.file_start_unix_ms = if file.started_unix_ms != 0 {
             file.started_unix_ms
         } else {
             bun_core::time::milli_timestamp()
         };
-        for payload in &file.tests {
-            if let Some(test) = runner::decode_test_case(&mut Reader { p: payload }, rel) {
+        let mut attempts = file.attempts.into_iter().peekable();
+        for (index, payload) in file.tests.into_iter().enumerate() {
+            let earlier_failures =
+                core::iter::from_fn(|| attempts.next_if(|(test, _)| *test == index))
+                    .map(|(_, attempt)| runner::decode_failure(&mut Reader { p: &attempt }))
+                    .collect();
+            let test = runner::decode_test_case(&mut Reader { p: &payload }, rel, earlier_failures);
+            if let Some(test) = test {
                 junit.record_test_case(&test).expect("oom");
             }
         }

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

@@ -1016,6 +1016,14 @@ pub(crate) fn worker_emit_test_done(
     // SAFETY: cmds_ptr was set from &mut WorkerCommands in run_as_worker; pointee
     // outlives all callers (process exits before it is dropped).
     let cmds = unsafe { &mut *cmds_ptr };
+    if let Some(test) = test {
+        for failure in &test.earlier_failures {
+            cmds.emit(frame::Kind::TestAttempt, |wf| {
+                wf.u32(file_idx);
+                encode_failure(wf, failure);
+            });
+        }
+    }
     cmds.emit(frame::Kind::TestDone, |wf| {
         wf.u32(file_idx);
         wf.str(formatted_line);
@@ -1036,14 +1044,27 @@ fn encode_test_case(wf: &mut Frame, t: &test_command::TestCaseReport<'_>) {
         wf.str(name);
         wf.u32(line);
     }
-    // Earlier attempts' failures first, then the last attempt's if it failed.
+    // The last attempt's failure. Each of `earlier_failures` goes ahead in a `TestAttempt` frame.
     wf.u32(u32::from(t.failure.is_some()));
-    wf.u32(u32::try_from(t.earlier_failures.len()).expect("int cast"));
-    for f in t.earlier_failures.iter().chain(&t.failure) {
-        wf.str(&f.name);
-        wf.str(&f.message);
-        wf.str(&f.body);
-        wf.u32(u32::from(f.uncaught));
+    if let Some(failure) = &t.failure {
+        encode_failure(wf, failure);
+    }
+}
+
+fn encode_failure(wf: &mut Frame, f: &test_command::TestFailure) {
+    wf.str(&f.name);
+    wf.str(&f.message);
+    wf.str(&f.body);
+    wf.u32(u32::from(f.uncaught));
+}
+
+/// Inverse of `encode_failure`.
+pub(crate) fn decode_failure(rd: &mut frame::Reader<'_>) -> test_command::TestFailure {
+    test_command::TestFailure {
+        name: rd.str().to_vec(),
+        message: rd.str().to_vec(),
+        body: rd.str().to_vec(),
+        uncaught: rd.u32() != 0,
     }
 }
 
@@ -1052,6 +1073,7 @@ fn encode_test_case(wf: &mut Frame, t: &test_command::TestCaseReport<'_>) {
 pub(crate) fn decode_test_case<'a>(
     rd: &mut frame::Reader<'a>,
     file: &'a [u8],
+    earlier_failures: Vec<test_command::TestFailure>,
 ) -> Option<test_command::TestCaseReport<'a>> {
     use crate::test_runner::execution::Result;
     let status =
@@ -1068,16 +1090,7 @@ pub(crate) fn decode_test_case<'a>(
     for _ in 0..n {
         scopes.push((rd.str(), rd.u32()));
     }
-    let has_failure = rd.u32() != 0;
-    let earlier = rd.u32() as usize;
-    let mut read_failure = || test_command::TestFailure {
-        name: rd.str().to_vec(),
-        message: rd.str().to_vec(),
-        body: rd.str().to_vec(),
-        uncaught: rd.u32() != 0,
-    };
-    let earlier_failures: Vec<_> = (0..earlier).map(|_| read_failure()).collect();
-    let failure = has_failure.then(read_failure);
+    let failure = (rd.u32() != 0).then(|| decode_failure(rd));
     Some(test_command::TestCaseReport {
         file,
         scopes,

src/runtime/cli/test_command.rs

@@ -2047,7 +2047,7 @@ impl CommandLineReporter {
     /// Everything a structured reporter needs about one finished test. Built
     /// once in `handle_test_completed`; the serial runner hands it straight
     /// to `JunitReporter::record_test_case`, a `--parallel` worker encodes it
-    /// on its `TestDone` frame and the coordinator replays it.
+    /// on its `TestAttempt` and `TestDone` frames and the coordinator replays it.
     fn test_case_report<'a>(
         status: bun_test::Execution::Result,
         buntest: &bun_test::BunTest,

test/js/junit-reporter/junit.test.js

@@ -1,6 +1,6 @@
 import { file, spawn } from "bun";
 import { describe, expect, it } from "bun:test";
-import { bunEnv, bunExe, tempDir } from "harness";
+import { bunEnv, bunExe, isASAN, isDebug, tempDir } from "harness";
 import { join } from "node:path";
 
 const xml2js = require("xml2js");
@@ -621,6 +621,171 @@ describe("junit reporter", () => {
     expect(proc.exitCode).toBe(1);
   });
 
+  it.concurrent("writes the earlier attempts of retried tests with --parallel as without", async () => {
+    await using tmpDir = tempDir("junit-retry-parallel", {
+      "package.json": "{}",
+      "a.test.js": `
+        import { test } from "bun:test";
+        const turn = () => new Promise(resolve => setImmediate(resolve));
+        const otherEnded = Promise.withResolvers();
+        const attempts = {};
+        const attempt = name => (attempts[name] = (attempts[name] ?? 0) + 1);
+
+        test("never passes", { retry: 3 }, () => {
+          throw new Error("never passes, attempt " + attempt("never"));
+        });
+        test("passes at last", { retry: 3 }, () => {
+          const nth = attempt("at last");
+          if (nth < 4) throw new Error("passes at last, attempt " + nth);
+        });
+        test("passes", () => {});
+        test("times out, then throws", { retry: 1, timeout: 50 }, async () => {
+          if (attempt("timeout") === 1) await new Promise(() => {});
+          throw new Error("thrown by the second attempt");
+        });
+        // The test declared first ends last. The attempts of each stay with it.
+        test.concurrent("fails once and ends last", { retry: 1 }, async () => {
+          if (attempt("last") === 1) throw new Error("first attempt of the last to end");
+          await otherEnded.promise;
+          for (let i = 0; i < 5; i++) await turn();
+        });
+        test.concurrent("fails once and ends first", { retry: 1 }, async () => {
+          if (attempt("first") === 1) throw new Error("first attempt of the first to end");
+          otherEnded.resolve();
+        });
+      `,
+      "b.test.js": `import { test } from "bun:test"; test("other file", () => {});`,
+    });
+    const report = async (outfile, ...args) => {
+      await using proc = spawn([bunExe(), "test", ...args, "--reporter=junit", `--reporter-outfile=${outfile}`], {
+        cwd: String(tmpDir),
+        env: bunEnv,
+        stdout: "pipe",
+        stderr: "ignore",
+      });
+      const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
+      const xml = await file(join(String(tmpDir), outfile)).text();
+      return {
+        parallel: stdout.includes("PARALLEL"),
+        elements: xml.match(/<testcase name="[^"]*"|<\w*[fF]ailure [^>]*>/g),
+        xml: filterJunitXmlOutput(xml),
+        exitCode,
+      };
+    };
+    const [serial, parallel] = await Promise.all([report("serial.xml"), report("parallel.xml", "--parallel=2")]);
+    expect(serial.elements).toEqual([
+      `<testcase name="never passes"`,
+      `<failure type="Error" message="never passes, attempt 4">`,
+      `<rerunFailure type="Error" message="never passes, attempt 1">`,
+      `<rerunFailure type="Error" message="never passes, attempt 2">`,
+      `<rerunFailure type="Error" message="never passes, attempt 3">`,
+      `<testcase name="passes at last"`,
+      `<flakyFailure type="Error" message="passes at last, attempt 1">`,
+      `<flakyFailure type="Error" message="passes at last, attempt 2">`,
+      `<flakyFailure type="Error" message="passes at last, attempt 3">`,
+      `<testcase name="passes"`,
+      `<testcase name="times out, then throws"`,
+      `<failure type="Error" message="thrown by the second attempt">`,
+      `<rerunFailure type="TimeoutError" message="test timed out" />`,
+      `<testcase name="fails once and ends first"`,
+      `<flakyFailure type="Error" message="first attempt of the first to end">`,
+      `<testcase name="fails once and ends last"`,
+      `<flakyFailure type="Error" message="first attempt of the last to end">`,
+      `<testcase name="other file"`,
+    ]);
+    expect(parallel).toEqual({ ...serial, parallel: true });
+    expect(serial.exitCode).toBe(1);
+  });
+
+  it.concurrent(
+    "with --parallel, a retried test whose worker exits is in the report with all its attempts or not at all",
+    async () => {
+      // The frame of the first attempt is larger than a socket buffer: the worker has not
+      // sent all that it has for "flaky" when the next test ends the process.
+      await using tmpDir = tempDir("junit-retry-parallel-exit", {
+        "package.json": "{}",
+        "a.test.js": `
+          import { test } from "bun:test";
+          const big = Buffer.alloc(1 << 20, "x").toString();
+          let attempt = 0;
+          test("flaky", { retry: 2 }, () => {
+            if (++attempt === 1) throw new Error(big);
+          });
+          test("exits", () => {
+            process.exit(1);
+          });
+        `,
+        "b.test.js": `import { test } from "bun:test"; test("other file", () => {});`,
+      });
+      await using proc = spawn([bunExe(), "test", "--parallel=2", "--reporter=junit", "--reporter-outfile=junit.xml"], {
+        cwd: String(tmpDir),
+        env: bunEnv,
+        stdout: "ignore",
+        stderr: "ignore",
+      });
+      const exitCode = await proc.exited;
+      const xml = await file(join(String(tmpDir), "junit.xml")).text();
+      const elements = xml.match(/<testcase name="[^"]*"|<\w*[fF]ailure /g);
+      const crashed = [`<testcase name="(worker crashed)"`, `<failure `, `<testcase name="other file"`];
+      expect([crashed, [`<testcase name="flaky"`, `<flakyFailure `, ...crashed]]).toContainEqual(elements);
+      expect(exitCode).toBe(1);
+    },
+  );
+
+  it(
+    "with --parallel, hundreds of failed attempts after a huge error are one failed test case",
+    async () => {
+      // A worker sends each earlier failed attempt in an IPC frame of its own, so no frame
+      // grows with the retry count. The first attempt alone has more text than the 64 MiB of
+      // a frame: its message fits, and its stack, which repeats the message, is cut.
+      const retries = 300;
+      const messageLength = 34 * 1024 * 1024;
+      await using tmpDir = tempDir("junit-retry-parallel-huge", {
+        "package.json": "{}",
+        "a.test.js": `
+          import { test } from "bun:test";
+          let attempt = 0;
+          test("fails every time", { retry: ${retries} }, () => {
+            attempt++;
+            throw new Error(attempt === 1 ? Buffer.alloc(${messageLength}, "a").toString() : "short " + attempt);
+          });
+        `,
+        "b.test.js": `import { test } from "bun:test"; test("other file", () => {});`,
+      });
+      await using proc = spawn([bunExe(), "test", "--parallel=2", "--reporter=junit", "--reporter-outfile=junit.xml"], {
+        cwd: String(tmpDir),
+        env: bunEnv,
+        stdout: "pipe",
+        stderr: "ignore",
+      });
+      const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
+      expect(stdout).toContain("PARALLEL");
+      // An element per line. The line of the first attempt is megabytes long, so the
+      // assertions take pieces of it: a failure must not print it.
+      const lines = (await file(join(String(tmpDir), "junit.xml")).text()).split("\n").map(line => line.trimStart());
+      const element = /^<(?:testcase name="[^"]*"|\w*[fF]ailure type="\w+" message="[^"]{0,16})/;
+      expect(lines.flatMap(line => line.match(element) ?? [])).toEqual([
+        `<testcase name="fails every time"`,
+        `<failure type="Error" message="short ${retries + 1}`,
+        `<rerunFailure type="Error" message="aaaaaaaaaaaaaaaa`,
+        ...Array.from({ length: retries - 1 }, (_, i) => `<rerunFailure type="Error" message="short ${i + 2}`),
+        `<testcase name="other file"`,
+      ]);
+      const first = lines.find(line => line.startsWith(`<rerunFailure type="Error" message="a`));
+      // The message is whole, and the stack fills the rest of the frame up to the marker.
+      expect(first.indexOf(`">Error: a`)).toBe(`<rerunFailure type="Error" message="`.length + messageLength);
+      const cut = "a&#10;... [output truncated: would exceed --parallel IPC frame limit]&#10;</rerunFailure>";
+      expect(first.slice(-cut.length)).toBe(cut);
+      expect(first.length).toBeWithin(63 * 1024 * 1024, 65 * 1024 * 1024);
+      expect(filterJunitXmlOutput(lines[1])).toBe(
+        `<testsuites name="bun test" tests="2" assertions="0" failures="1" skipped="0">`,
+      );
+      expect(exitCode).toBe(1);
+    },
+    // The report of this one test has 64 MiB, which is what a frame holds.
+    isDebug || isASAN ? 60_000 : 20_000,
+  );
+
   it("omits U+FFFE and U+FFFF, which XML 1.0 forbids", async () => {
     await using tmpDir = tempDir("junit-noncharacters", {
       "package.json": "{}",