ipc: send a queue item larger than 2 GiB in capped write requests

src/runtime/ipc.rs

@@ -894,6 +894,9 @@ enum ContinueSendReason {
     OnWritable,
 }
 
+/// The most bytes one write request carries: the transport reports the count it took as an `i32`.
+const MAX_WRITE_LEN: usize = i32::MAX as usize;
+
 #[derive(bun_ptr::CellRefCounted)]
 pub(crate) struct SendQueue {
     ref_count: Cell<u32>,
@@ -1536,14 +1539,18 @@ impl SendQueue {
         enum Done {
             AwaitAck,
             Completed(SendHandle),
+            Capped,
             Partial,
             NoProgress,
             Error,
         }
         let done = self.queue.with_mut(|queue| {
             let first = &mut queue[0];
             let to_send_len = first.data.list.len() - first.data.cursor;
-            if n as usize == to_send_len {
+            let Ok(written) = usize::try_from(n) else {
+                return Done::Error;
+            };
+            if written == to_send_len {
                 if first.handle.is_some() {
                     // the message was fully written, but it had a handle.
                     // we must wait for ACK or NACK before sending any more messages.
@@ -1561,12 +1568,16 @@ impl SendQueue {
                     // shift the queue and try to send the next item immediately.
                     Done::Completed(queue.remove(0))
                 }
-            } else if n > 0 && n < i32::try_from(first.data.list.len()).expect("int cast") {
+            } else if written > 0 && written < to_send_len {
                 // the item was partially sent; update the cursor and wait for writable to send the rest
                 // (if we tried to send a handle, a partial write means the handle wasn't sent yet.)
-                first.data.cursor += usize::try_from(n).expect("int cast");
-                Done::Partial
-            } else if n == 0 {
+                first.data.cursor += written;
+                if written == MAX_WRITE_LEN {
+                    Done::Capped
+                } else {
+                    Done::Partial
+                }
+            } else if written == 0 {
                 // no bytes written; wait for writable
                 Done::NoProgress
             } else {
@@ -1581,6 +1592,10 @@ impl SendQueue {
                 item.complete(&global_this); // call the callback & deinit
                 self.continue_send(&global_this, ContinueSendReason::OnWritable);
             }
+            Done::Capped => {
+                // The transport took a whole capped request, so no writable event follows.
+                self.continue_send(&global_this, ContinueSendReason::OnWritable);
+            }
             Done::Partial | Done::NoProgress => {}
             Done::Error => {
                 // error. close socket.
@@ -1705,7 +1720,7 @@ impl SendQueue {
                 let first = &queue[0];
                 let data = &first.data.list[first.data.cursor..];
                 log!("SendQueue#write len {}", data.len());
-                let write_len = data.len().min(i32::MAX as usize);
+                let write_len = data.len().min(MAX_WRITE_LEN);
                 Box::from(&data[0..write_len])
             });
 
@@ -1760,6 +1775,7 @@ impl SendQueue {
                 let first = &queue[0];
                 let data = &first.data.list[first.data.cursor..];
                 log!("SendQueue#write len {}", data.len());
+                let data = &data[..data.len().min(MAX_WRITE_LEN)];
                 if let Some(fd_unwrapped) = fd {
                     socket.write_fd(data, fd_unwrapped.native())
                 } else {

test/js/bun/spawn/spawn.ipc.test.ts

@@ -1,6 +1,7 @@
 import { spawn } from "bun";
 import { describe, expect, it } from "bun:test";
 import { bunEnv, bunExe, gcTick, isWindows } from "harness";
+import { totalmem } from "node:os";
 import path from "path";
 
 describe.each(["advanced", "json"])("ipc mode %s", mode => {
@@ -384,3 +385,54 @@ it.skipIf(isWindows)("advanced serialization advertises wire format version 2",
   expect(JSON.parse(stdout.trim())).toEqual([1, 2, 0, 0, 0]);
   expect(exitCode).toBe(0);
 });
+
+// Messages queued behind an in-flight message coalesce into one queue item, so that
+// item passes 2 GiB when the peer reads slowly. The write completion cast its length
+// to i32 and panicked with "int cast: TryFromIntError(PosOverflow)". The child peaks
+// at 4 to 6 GB of RSS. Inside a container os.totalmem() reports the host's RAM;
+// process.constrainedMemory() reports the cgroup limit there.
+// Not on Windows: the parent has to drain a whole 2 GiB write request before the next
+// one starts, and the sender copies each request, so the child peaks above 10 GB.
+const memory = Math.min(totalmem(), process.constrainedMemory() || Infinity);
+describe.skipIf(isWindows || memory < 16 * 1024 ** 3)("send queue past 2 GiB", () => {
+  // The item has to pass 2^31 bytes, and the child copies each message three times to
+  // queue it: about 7 s on a debug ASAN build, more when the machine is loaded.
+  const timeout = 60_000;
+  it(
+    "keeps sending a queue item larger than 2 GiB",
+    async () => {
+      const chunkLength = 100 * 1024 * 1024;
+      // All 22 sends run in one synchronous tick. The first message is in flight, and
+      // the other 21 (2100 MiB) coalesce into one item behind it.
+      const childSource = `
+        const chunk = Buffer.alloc(${chunkLength}, "x").toString();
+        for (let i = 0; i < 22; i++) process.send(chunk);
+        process.on("message", () => process.exit(0));
+      `;
+      const { promise, resolve, reject } = Promise.withResolvers<number[]>();
+      const lengths: number[] = [];
+      await using child = spawn([bunExe(), "-e", childSource], {
+        env: bunEnv,
+        stdio: ["ignore", "inherit", "inherit"],
+        serialization: "advanced",
+        ipc(message, subprocess) {
+          if (lengths.length === 2) return;
+          lengths.push(message.length);
+          // The second message is the first one inside the coalesced item. Its arrival
+          // proves the child wrote that item across many partial writes. Stop the
+          // child there instead of draining the whole queue.
+          if (lengths.length === 2) {
+            resolve(lengths);
+            subprocess.send("stop");
+          }
+        },
+        onExit(_subprocess, exitCode, signalCode) {
+          reject(new Error(`child exited (${exitCode}, ${signalCode}) before the second message`));
+        },
+      });
+      expect(await promise).toEqual([chunkLength, chunkLength]);
+      expect(await child.exited).toBe(0);
+    },
+    timeout,
+  );
+});