From fbb4a22e01893b815073fc2917c0882058c655d2 Mon Sep 17 00:00:00 2001 From: Kushagra Kumar Date: Tue, 20 Oct 2026 01:12:13 +0530 Subject: [PATCH] fix(collab-net): allow live sessions above 64 MiB and report accurate transport limits (#857) - Configure server write loop in hub.rs with max_server_message so outgoing ServerMsg payloads above 64 MiB (e.g. document snapshots in Welcome messages) are delivered without socket termination. - Set core max_message_bytes in local live session HostConfig to max_server_message. - Map tungstenite Capacity errors in guest read loop to Outcome::Failed with accurate size limits rather than Outcome::Refused which falsely reported 'The host refused'. - Add loopback tests verifying Welcome delivery above client message limits and accurate diagnostics on client capacity overrun. --- crates/collab-net/src/guest.rs | 12 ++++- crates/collab-net/src/hub.rs | 20 ++++---- crates/collab-net/src/link.rs | 2 +- crates/collab-net/src/loopback_tests.rs | 66 +++++++++++++++++++++++++ 4 files changed, 89 insertions(+), 11 deletions(-) diff --git a/crates/collab-net/src/guest.rs b/crates/collab-net/src/guest.rs index eb899b8d..1cfd784c 100644 --- a/crates/collab-net/src/guest.rs +++ b/crates/collab-net/src/guest.rs @@ -367,7 +367,17 @@ fn read_loop(mut ws: WebSocket, ctx: &ReaderCtx) { Ok(Message::Close(f)) => break f.map_or_else(|| closed(0, ""), |f| closed(f.code.into(), &f.reason)), Ok(Message::Text(_)) => break Outcome::Refused("the host sent a text message".into()), Ok(_) => {} - Err(tungstenite::Error::Capacity(_)) => break Outcome::Refused("the host sent a message over the size limit".into()), + Err(tungstenite::Error::Capacity(_)) => { + let max = ctx.cfg.limits.max_server_message; + let limit = if max >= 1 << 20 { + format!("{} MiB", max >> 20) + } else if max >= 1 << 10 { + format!("{} KiB", max >> 10) + } else { + format!("{max} bytes") + }; + break Outcome::Failed(format!("received message exceeded size limit (max {limit})")); + } Err(e) => break Outcome::Retry(e.to_string()), } }; diff --git a/crates/collab-net/src/hub.rs b/crates/collab-net/src/hub.rs index 5cf3e519..8ddde6e2 100644 --- a/crates/collab-net/src/hub.rs +++ b/crates/collab-net/src/hub.rs @@ -860,7 +860,7 @@ fn serve(mut stream: TcpStream, conn: u64, hub: &Sender, limits: NetLimit let Ok(peer) = stream.peer_addr().map(|a| a.ip().to_canonical()) else { return }; let proxied = limits.trusted_proxy.is_some_and(|p| p.to_canonical() == peer); let Ok((rd, wr, sock)) = Duplex::split(stream, None) else { return }; - let cfg = config(limits.max_client_message); + let write_cfg = config(limits.max_server_message); let mut forwarded = None; // The handshake callback's type is tungstenite's (its error is an HTTP response). #[allow(clippy::result_large_err)] @@ -904,14 +904,16 @@ fn serve(mut stream: TcpStream, conn: u64, hub: &Sender, limits: NetLimit max_block: limits.max_queued_block, max_all: limits.max_queued_total, }; - let writer = - match sock.try_clone().map(|s| spawn("live-write", move || write_loop(WebSocket::from_raw_socket(wr, Role::Server, Some(cfg)), &rx, &s))) { - Ok(Ok(t)) => t, - _ => { - shutdown(&sock); - return; - } - }; + let writer = match sock + .try_clone() + .map(|s| spawn("live-write", move || write_loop(WebSocket::from_raw_socket(wr, Role::Server, Some(write_cfg)), &rx, &s))) + { + Ok(Ok(t)) => t, + _ => { + shutdown(&sock); + return; + } + }; let end = |why: Option<&str>| { if let Some(why) = why { refuse(&out, why); diff --git a/crates/collab-net/src/link.rs b/crates/collab-net/src/link.rs index a84395aa..66d07477 100644 --- a/crates/collab-net/src/link.rs +++ b/crates/collab-net/src/link.rs @@ -139,7 +139,7 @@ impl LiveLink for NetLink { doc, name: o.name, limits: self.limits, - core: figcraft_collab::Limits::default(), + core: figcraft_collab::Limits { max_message_bytes: self.limits.max_server_message, ..figcraft_collab::Limits::default() }, }; let host = Host::start(cfg, inbox.clone()).map_err(|e| format!("can't listen on {ip}: {e}"))?; let invite = Invite::new(host.addr().to_string(), file, token); diff --git a/crates/collab-net/src/loopback_tests.rs b/crates/collab-net/src/loopback_tests.rs index 32c86bd0..9d86939c 100644 --- a/crates/collab-net/src/loopback_tests.rs +++ b/crates/collab-net/src/loopback_tests.rs @@ -642,3 +642,69 @@ fn the_host_removes_someone_and_they_cant_come_back() { host.stop(); assert_eq!(host.connections(), 0); } + +#[test] +fn welcome_messages_larger_than_client_message_limit_are_delivered() { + let mut doc = Document::new("A".repeat(3000)); + doc.repair(); + let probe = Probe::new(); + let limits = NetLimits { max_client_message: 1024, max_server_message: 64 << 10, ..fast_limits() }; + let cfg = HostConfig { + addr: SocketAddr::from(([127, 0, 0, 1], 0)), + any_port_if_taken: false, + file: FILE.into(), + token: TOKEN.into(), + doc, + name: "Hana".into(), + limits, + core: Limits { max_message_bytes: 64 << 10, ..Limits::default() }, + }; + let mut host = Host::start(cfg, probe.inbox.clone()).unwrap(); + let invite = Invite::new(host.addr().to_string(), FILE, TOKEN); + + let guest_probe = Probe::new(); + let mut guest_cfg = quick(GuestConfig::new(invite, "Ada".into())); + guest_cfg.limits = limits; + let _g = Guest::start(guest_cfg, guest_probe.inbox.clone()).unwrap(); + let mut gp = guest_probe; + gp.wait("welcome", welcomed); + assert!(welcomed(&gp.events)); + host.stop(); +} + +#[test] +fn oversized_server_message_reports_accurate_diagnostic() { + let mut doc = Document::new("B".repeat(3000)); + doc.repair(); + let probe = Probe::new(); + let limits = NetLimits { max_server_message: 64 << 10, ..fast_limits() }; + let cfg = HostConfig { + addr: SocketAddr::from(([127, 0, 0, 1], 0)), + any_port_if_taken: false, + file: FILE.into(), + token: TOKEN.into(), + doc, + name: "Hana".into(), + limits, + core: Limits { max_message_bytes: 64 << 10, ..Limits::default() }, + }; + let mut host = Host::start(cfg, probe.inbox.clone()).unwrap(); + let invite = Invite::new(host.addr().to_string(), FILE, TOKEN); + + let guest_probe = Probe::new(); + let mut guest_cfg = quick(GuestConfig::new(invite, "Ada".into())); + guest_cfg.limits = NetLimits { max_server_message: 512, ..fast_limits() }; + let _g = Guest::start(guest_cfg, guest_probe.inbox.clone()).unwrap(); + let mut gp = guest_probe; + gp.wait("accurate failure diagnostic", |e| { + e.iter().any(|ev| { + matches!( + ev, + Event::Status(Status::Failed { reason }) + if !reason.contains("The host refused") && reason.contains("received message exceeded size limit (max 512 bytes)") + ) + }) + }); + assert!(!welcomed(&gp.events)); + host.stop(); +}