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.
This commit is contained in:
Kushagra Kumar
2026-10-20 01:12:13 +05:30
parent a1369310d4
commit fbb4a22e01
4 changed files with 89 additions and 11 deletions
+11 -1
View File
@@ -367,7 +367,17 @@ fn read_loop(mut ws: WebSocket<Duplex>, 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()),
}
};
+11 -9
View File
@@ -860,7 +860,7 @@ fn serve(mut stream: TcpStream, conn: u64, hub: &Sender<HubIn>, 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<HubIn>, 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);
+1 -1
View File
@@ -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);
+66
View File
@@ -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();
}