From c5f6aae56dc4873dee18b12ddd069bb22a69aebe Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Tue, 1 Sep 2026 23:33:27 +0000 Subject: [PATCH 1/3] ipc: compare the send queue write count as usize, not i32 The send queue coalesces every message queued behind a partial write into the head item. When the peer has not read yet, that item can pass 2 GiB. on_write_complete converted the item length to i32 to compare it with the write count and panicked with "int cast: TryFromIntError(PosOverflow)". Convert the write count to usize instead and compare it with the usize lengths. On Windows, a write request is capped at i32::MAX bytes and libuv completes it in full, so a partial completion there starts the next request at once instead of waiting for a writable event that never comes. --- src/runtime/ipc.rs | 67 +++++++++++++++++------------ test/js/bun/spawn/spawn.ipc.test.ts | 45 +++++++++++++++++++ 2 files changed, 84 insertions(+), 28 deletions(-) diff --git a/src/runtime/ipc.rs b/src/runtime/ipc.rs index 83f717326800..c5315da5db96 100644 --- a/src/runtime/ipc.rs +++ b/src/runtime/ipc.rs @@ -1539,34 +1539,39 @@ impl SendQueue { 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 { - 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. - let item = queue.remove(0); - self.waiting_for_ack.with_mut(|w| { - if w.is_some() { - log!("[error] already waiting for ack. this should never happen."); - } - // shift the item off the queue and move it to waiting_for_ack - *w = Some(item); - }); - Done::AwaitAck - } else { - // the message was fully sent, but there may be more items in the queue. - // shift the queue and try to send the next item immediately. - Done::Completed(queue.remove(0)) + // `n` is the write count from the socket (at most i32::MAX per write). + // The coalesced item can be larger than that, so compare as usize. + match usize::try_from(n) { + Ok(written) 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. + let item = queue.remove(0); + self.waiting_for_ack.with_mut(|w| { + if w.is_some() { + log!("[error] already waiting for ack. this should never happen."); + } + // shift the item off the queue and move it to waiting_for_ack + *w = Some(item); + }); + Done::AwaitAck + } else { + // the message was fully sent, but there may be more items in the queue. + // 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") { - // 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 { - // no bytes written; wait for writable - Done::NoProgress - } else { - Done::Error + Ok(0) => { + // no bytes written; wait for writable + Done::NoProgress + } + Ok(written) if 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 += written; + Done::Partial + } + _ => Done::Error, } }); match done { @@ -1577,7 +1582,13 @@ impl SendQueue { item.complete(&global_this); // call the callback & deinit self.continue_send(&global_this, ContinueSendReason::OnWritable); } - Done::Partial | Done::NoProgress => {} + Done::Partial => { + // libuv completes a request in full or fails. A partial write is + // the i32::MAX cap in `write`, and no writable event follows it. + #[cfg(windows)] + self.continue_send(&global_this, ContinueSendReason::OnWritable); + } + Done::NoProgress => {} Done::Error => { // error. close socket. self.close_socket(CloseReason::Failure, CloseFrom::User); diff --git a/test/js/bun/spawn/spawn.ipc.test.ts b/test/js/bun/spawn/spawn.ipc.test.ts index a5e0e16a3a6b..98da35a86ed0 100644 --- a/test/js/bun/spawn/spawn.ipc.test.ts +++ b/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 => { @@ -295,3 +296,47 @@ 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); }); + +// The send queue coalesces every message queued behind a partial write into one +// buffer. When the peer has not read yet, that buffer can pass 2 GiB. The write +// completion compared the socket's i32 write count against that length as an i32 +// and panicked with "int cast: TryFromIntError(PosOverflow)". The child peaks at +// 4 to 6 GB of RSS (the 2.2 GB buffer, its last capacity doubling, freed payloads). +describe.skipIf(isWindows || totalmem() < 16 * 1024 ** 3)("send queue past 2 GiB", () => { + // Debug and ASAN builds copy the 2.2 GB about four times before the first message. + const timeout = 60_000; + it( + "keeps sending after more than 2 GiB is queued before the first writable event", + async () => { + const chunkLength = 100 * 1024 * 1024; + // 22 chunks = 2200 MiB, all queued in one synchronous tick, so the head item + // passes i32::MAX bytes before the event loop sees the socket writable. + 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(); + let received = 0; + await using child = spawn([bunExe(), "-e", childSource], { + env: bunEnv, + stdio: ["ignore", "inherit", "inherit"], + serialization: "advanced", + ipc(message, subprocess) { + // The first complete message proves the child kept writing past the cap. + // Stop the child there instead of draining the whole queue. + if (received++ === 0) { + resolve(message.length); + subprocess.send("stop"); + } + }, + onExit(_subprocess, exitCode, signalCode) { + reject(new Error(`child exited (${exitCode}, ${signalCode}) before the first message`)); + }, + }); + expect(await promise).toBe(chunkLength); + expect(await child.exited).toBe(0); + }, + timeout, + ); +}); From b231b14ca54ce237a8efd22cdd71ebe66735e469 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 2 Sep 2026 03:50:46 +0000 Subject: [PATCH 2/3] test: gate the 2 GiB ipc test on the cgroup memory limit too --- test/js/bun/spawn/spawn.ipc.test.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/test/js/bun/spawn/spawn.ipc.test.ts b/test/js/bun/spawn/spawn.ipc.test.ts index 98da35a86ed0..f5b42ed557d5 100644 --- a/test/js/bun/spawn/spawn.ipc.test.ts +++ b/test/js/bun/spawn/spawn.ipc.test.ts @@ -302,7 +302,10 @@ it.skipIf(isWindows)("advanced serialization advertises wire format version 2", // completion compared the socket's i32 write count against that length as an i32 // and panicked with "int cast: TryFromIntError(PosOverflow)". The child peaks at // 4 to 6 GB of RSS (the 2.2 GB buffer, its last capacity doubling, freed payloads). -describe.skipIf(isWindows || totalmem() < 16 * 1024 ** 3)("send queue past 2 GiB", () => { +// Inside a container os.totalmem() reports the host's RAM; +// process.constrainedMemory() reports the cgroup limit there. +const memory = Math.min(totalmem(), process.constrainedMemory() || Infinity); +describe.skipIf(isWindows || memory < 16 * 1024 ** 3)("send queue past 2 GiB", () => { // Debug and ASAN builds copy the 2.2 GB about four times before the first message. const timeout = 60_000; it( From 7b3d97707a7827d4e51392bdb3503e8483c8afce Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 1 Oct 2026 15:58:14 +0000 Subject: [PATCH 3/3] ipc: continue a capped write on every transport A write request carries at most i32::MAX bytes, because the transport reports the count it took as an i32. When the transport takes a whole capped request and bytes remain, no writable event follows: usockets arms the poll only for a short write, and libuv has no such event. on_write_complete now starts the next request at once in that case, on every transport. Before, only Windows did. write() caps the request on both transports with the same constant. The test now waits for the second message. A partially written item no longer takes appended messages, so the first message is outside the coalesced item. --- src/runtime/ipc.rs | 77 +++++++++++++++-------------- test/js/bun/spawn/spawn.ipc.test.ts | 40 ++++++++------- 2 files changed, 63 insertions(+), 54 deletions(-) diff --git a/src/runtime/ipc.rs b/src/runtime/ipc.rs index 43733fb2bb96..fbc8f506c5f7 100644 --- a/src/runtime/ipc.rs +++ b/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, @@ -1536,6 +1539,7 @@ impl SendQueue { enum Done { AwaitAck, Completed(SendHandle), + Capped, Partial, NoProgress, Error, @@ -1543,39 +1547,41 @@ impl SendQueue { let done = self.queue.with_mut(|queue| { let first = &mut queue[0]; let to_send_len = first.data.list.len() - first.data.cursor; - // `n` is the write count from the socket (at most i32::MAX per write). - // The coalesced item can be larger than that, so compare as usize. - match usize::try_from(n) { - Ok(written) 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. - let item = queue.remove(0); - self.waiting_for_ack.with_mut(|w| { - if w.is_some() { - log!("[error] already waiting for ack. this should never happen."); - } - // shift the item off the queue and move it to waiting_for_ack - *w = Some(item); - }); - Done::AwaitAck - } else { - // the message was fully sent, but there may be more items in the queue. - // shift the queue and try to send the next item immediately. - Done::Completed(queue.remove(0)) - } - } - Ok(0) => { - // no bytes written; wait for writable - Done::NoProgress + 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. + let item = queue.remove(0); + self.waiting_for_ack.with_mut(|w| { + if w.is_some() { + log!("[error] already waiting for ack. this should never happen."); + } + // shift the item off the queue and move it to waiting_for_ack + *w = Some(item); + }); + Done::AwaitAck + } else { + // the message was fully sent, but there may be more items in the queue. + // shift the queue and try to send the next item immediately. + Done::Completed(queue.remove(0)) } - Ok(written) if 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 += written; + } 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 += written; + if written == MAX_WRITE_LEN { + Done::Capped + } else { Done::Partial } - _ => Done::Error, + } else if written == 0 { + // no bytes written; wait for writable + Done::NoProgress + } else { + Done::Error } }); match done { @@ -1586,13 +1592,11 @@ impl SendQueue { item.complete(&global_this); // call the callback & deinit self.continue_send(&global_this, ContinueSendReason::OnWritable); } - Done::Partial => { - // libuv completes a request in full or fails. A partial write is - // the i32::MAX cap in `write`, and no writable event follows it. - #[cfg(windows)] + Done::Capped => { + // The transport took a whole capped request, so no writable event follows. self.continue_send(&global_this, ContinueSendReason::OnWritable); } - Done::NoProgress => {} + Done::Partial | Done::NoProgress => {} Done::Error => { // error. close socket. self.close_socket(CloseReason::Failure, CloseFrom::User); @@ -1716,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]) }); @@ -1771,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 { diff --git a/test/js/bun/spawn/spawn.ipc.test.ts b/test/js/bun/spawn/spawn.ipc.test.ts index bd898a425257..92d1da59331d 100644 --- a/test/js/bun/spawn/spawn.ipc.test.ts +++ b/test/js/bun/spawn/spawn.ipc.test.ts @@ -386,47 +386,51 @@ it.skipIf(isWindows)("advanced serialization advertises wire format version 2", expect(exitCode).toBe(0); }); -// The send queue coalesces every message queued behind a partial write into one -// buffer. When the peer has not read yet, that buffer can pass 2 GiB. The write -// completion compared the socket's i32 write count against that length as an i32 -// and panicked with "int cast: TryFromIntError(PosOverflow)". The child peaks at -// 4 to 6 GB of RSS (the 2.2 GB buffer, its last capacity doubling, freed payloads). -// Inside a container os.totalmem() reports the host's RAM; +// 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", () => { - // Debug and ASAN builds copy the 2.2 GB about four times before the first message. + // 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 after more than 2 GiB is queued before the first writable event", + "keeps sending a queue item larger than 2 GiB", async () => { const chunkLength = 100 * 1024 * 1024; - // 22 chunks = 2200 MiB, all queued in one synchronous tick, so the head item - // passes i32::MAX bytes before the event loop sees the socket writable. + // 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(); - let received = 0; + const { promise, resolve, reject } = Promise.withResolvers(); + const lengths: number[] = []; await using child = spawn([bunExe(), "-e", childSource], { env: bunEnv, stdio: ["ignore", "inherit", "inherit"], serialization: "advanced", ipc(message, subprocess) { - // The first complete message proves the child kept writing past the cap. - // Stop the child there instead of draining the whole queue. - if (received++ === 0) { - resolve(message.length); + 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 first message`)); + reject(new Error(`child exited (${exitCode}, ${signalCode}) before the second message`)); }, }); - expect(await promise).toBe(chunkLength); + expect(await promise).toEqual([chunkLength, chunkLength]); expect(await child.exited).toBe(0); }, timeout,