Fix quadratic backlog drain in bun test --parallel IPC

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

@@ -108,13 +108,17 @@ pub type Socket = uws::NewSocketHandler<false>;
 #[cfg(not(windows))]
 pub struct PosixBackend {
     pub(crate) socket: Cell<Socket>,
+    /// Bytes at the front of `out` already written to the kernel;
+    /// front-draining per partial write instead is quadratic in backlog size.
+    out_head: Cell<usize>,
 }
 
 #[cfg(not(windows))]
 impl Default for PosixBackend {
     fn default() -> Self {
         Self {
             socket: Cell::new(Socket::DETACHED),
+            out_head: Cell::new(0),
         }
     }
 }
@@ -448,13 +452,27 @@ impl<Owner: ChannelOwner> Channel<Owner> {
         {
             while !self.done.get() {
                 let mut pending = self.out.replace(Vec::new());
-                if pending.is_empty() {
+                let mut head = self.backend.out_head.get();
+                debug_assert!(head <= pending.len());
+                if pending.len() <= head {
+                    self.backend.out_head.set(0);
                     self.out.set(pending);
                     return;
                 }
-                let wrote = self.backend.socket.get().write(pending.as_slice());
-                let w = usize::try_from(wrote).unwrap_or(0).min(pending.len());
-                pending.drain_front(w);
+                let wrote = self.backend.socket.get().write(&pending[head..]);
+                let w = usize::try_from(wrote)
+                    .unwrap_or(0)
+                    .min(pending.len() - head);
+                head += w;
+                if head == pending.len() {
+                    pending.clear();
+                    head = 0;
+                } else if head >= pending.len() - head {
+                    // Sent prefix caught up to the tail: compact (amortized linear).
+                    pending.drain_front(head);
+                    head = 0;
+                }
+                self.backend.out_head.set(head);
                 self.out.with_mut(|cur| {
                     pending.extend_from_slice(cur);
                     *cur = pending;

test/cli/test/parallel.test.ts

@@ -876,6 +876,55 @@ test("--parallel: a test producing a >64MB result line is truncated, not treated
   expect(exitCode).toBe(1);
 }, 90_000);
 
+test("--parallel: back-to-back huge result lines drain in linear time without corrupting the channel", async () => {
+  // Two tests whose status lines each come in just under the 64MB IPC frame
+  // cap, from the same worker. The second frame is queued while the first
+  // backlog is still draining, so this exercises the write cursor, the
+  // amortized compaction, and appends to a partially-sent backlog. The 60s
+  // kill is the regression signal: draining the backlog by memmoving the
+  // whole remainder after every partial write is quadratic in frame size,
+  // and on macOS (~8KB socketpair buffers) one 64MB frame alone took ~90s
+  // of memmove.
+  const TITLE_LENGTH = 60_000_000;
+  using dir = tempDir("parallel-huge-backlog", {
+    "huge.test.js": `import {test,expect} from "bun:test";
+      test(Buffer.alloc(${TITLE_LENGTH}, "A").toString(), () => expect(1).toBe(2));
+      test(Buffer.alloc(${TITLE_LENGTH}, "B").toString(), () => expect(1).toBe(2));`,
+    "ok.test.js": `import {test,expect} from "bun:test"; test("ok",()=>expect(1).toBe(1));`,
+  });
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "test", "--parallel=2"],
+    env: { ...bunEnv, BUN_TEST_PARALLEL_SCALE_MS: "0" },
+    cwd: String(dir),
+    stderr: "pipe",
+    stdout: "pipe",
+    timeout: 60_000,
+    killSignal: "SIGKILL",
+  });
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  expect(stdout).toContain("PARALLEL");
+  // Both huge tests failed normally: the channel survived both frames (no
+  // worker "crashed"), both 60MB status lines arrived intact (each stays
+  // just under the 64MB frame cap, so no truncation), and ok.test.js's pass
+  // survived on the other worker.
+  expect(stderr).not.toContain("crashed");
+  // Each title must arrive as one contiguous run of exactly TITLE_LENGTH
+  // characters: a shorter run (dropped bytes) leaves indexOf at -1, and a
+  // longer run (duplicated bytes) puts the title's letter right after the
+  // first match. Checked via index math so a failure doesn't dump 60MB
+  // strings into the log.
+  for (const letter of ["A", "B"]) {
+    const title = Buffer.alloc(TITLE_LENGTH, letter).toString();
+    const start = stderr.indexOf(title);
+    expect(start).toBeGreaterThanOrEqual(0);
+    expect(stderr[start + TITLE_LENGTH]).not.toBe(letter);
+  }
+  expect(stderr).toContain("1 pass");
+  expect(stderr).toContain("2 fail");
+  expect(proc.signalCode).toBeNull();
+  expect(exitCode).toBe(1);
+}, 90_000);
+
 test("--parallel: a test writing garbage to fd 3 does not hang the coordinator", async () => {
   using dir = tempDir("parallel-hostile-fd3", {
     "ok.test.js": `import {test,expect} from "bun:test"; test("ok",()=>expect(1).toBe(1));`,