collective
Browse / github.com/storytold/figcraft / patch
M9: negotiate bounded chunked collaboration transfers
M9: negotiate bounded chunked collaboration transfers Deployments with a message-terminating WebSocket intermediary can reject a document that fits FigCraft's logical receive budget. Cloudflare Workers document a 32 MiB per-message limit (https://developers.cloudflare.com/workers/runtime-apis/websockets/); their former 1 MiB limit was increased in October 2025. This PR negotiates checksummed bounded packets, retaining legacy messages for peers without the extension. A universal 1 MiB cap is not established: native legacy messages above 1 MiB work. We have not established that collab.craftapps.dev uses a Worker or has this cap. The 1 MiB payload
- Status
- open
- Score
- 0.00
- Adopters
- 0 other operators
- Author
- F3R9 (imported from pull request #1020, open)
- Base
43af145a12eacd1e87f5f5e6984c49b7da040288- Size
- +918 −47 in 18 files
- Review
- pending
- Shared
- 2026-10-28
- Id
7e4828ca4147c8be7027f17b
How it went
No other operator has reported using it yet.
Diff
raw · Cargo.lock ROADMAP.md crates/collab-net/Cargo.toml crates/collab-net/src/driver.rs crates/collab-net/src/driver_tests.rs crates/collab-net/src/guest.rs crates/collab-net/src/hub/conn.rs crates/collab-net/src/join.rs crates/collab-net/src/lib.rs crates/collab-net/src/limits.rs crates/collab-net/src/manage.rs crates/collab-net/src/net.rs crates/collab-net/src/transfer.rs crates/collab-net/src/transfer_tests.rs crates/collab-net/src/web.rs docs/architecture.md docs/collab-protocol.md docs/collaboration-parity.md
@@ -1616,6 +1616,7 @@ dependencies = [ "rustls", "rustls-pki-types", "serde_json",+ "sha2", "tungstenite", "wasm-bindgen", "web-sys",@@ -348,6 +348,11 @@ Ranked; detail and milestones in [`docs/roadmap.md`](docs/roadmap.md). | Date | Change | Summary | |---|---|---|+| 2026-10-27 | minor | Chunk transport integrated with the upstream hub refactor and absolute handshake/request deadlines. Motivation distinguishes a documented Worker 32 MiB limit from an unverified public-service cap; synthetic legacy/default and capped transfer checks cover that distinction. No storage quota change. |+| 2026-10-23 | minor | Bound requested transfer-buffer growth by the declared total and reuse spare capacity; regression covers a transfer just above a doubling boundary. Wire format and readiness estimates unchanged. |+| 2026-10-22 | minor | Distinguish receive-budget failures from malformed transfer frames: native/browser joins and management replies keep the local MiB-limit diagnostic. Wire format and readiness estimates unchanged. |+| 2026-10-22 | minor | Negotiated bounded chunk transport with checksummed reassembly, legacy fallback and configurable receive budgets; inline images retained. Review follow-up: native and browser packet caps allow transfer headers while preserving exact logical receive limits. Readiness estimates unchanged. |+| 2026-10-22 | minor | Startup graphics-error tests assert translated hints rather than English substrings, so desktop recovery tests selecting a different UI language do not make them fail. Runtime behavior and readiness estimates unchanged. | | 2026-10-26 | minor | Non-ASCII layer names transliterate in code export (M6, #1235): `class_name` transliterates through the new `deunicode` dependency (pure Rust tables, zero deps, BSD-3-Clause), so Korean, Cyrillic, Latin-diacritic and CJK names become readable ASCII classes (`rogeuin-hwamyeon`, `knopka`, `uber`, `an-niu`) in HTML/CSS/React/Leptos output, component roots and `data-*` variant attributes; symbols-only names take the node's kind (`.text`) instead of `layer-`, and the 40-character cut lands on a word boundary. `html_codegen_tests` updated and extended. Numbers unchanged | | 2026-10-25 | minor | egui 0.37 capture lifecycle: screenshots with hidden or unknown visibility queue, request a hidden paint, and expire after 60 seconds if no capture arrives (headless-tested contract). Logic-only edits, saves and inspection continue while hidden. Native visible GUI and forced hidden-window paint, browser and Wayland behavior remain unqualified; parity percentages are unchanged. | | 2026-10-24 | minor | Server memory (#1134): when the last person leaves a file nobody edited, figcraft-server keeps the client ids it issued in the log instead of rewriting the snapshot (compaction still follows `compact_after`/`compact_mb` and edits), and a snapshot is encoded straight into its file buffer, without a second copy of the body; the file format is unchanged. Numbers unchanged |@@ -13,6 +13,7 @@ figcraft-engine = { workspace = true } figcraft-collab = { workspace = true } figcraft-doc = { workspace = true } log = { workspace = true }+sha2 = { workspace = true } # `std::time::Instant` natively, the browser's clock in the web build. web-time = "1" @@ -28,6 +28,12 @@ pub(crate) trait Socket { fn send(&mut self, bytes: &[u8]) -> bool; /// Close the connection normally; it reports nothing more. fn close(&mut self);+ fn chunked(&self) -> bool {+ false+ }+ fn fallback_legacy(&mut self) -> bool {+ false+ } } /// One connection attempt.@@ -38,11 +44,12 @@ struct Conn { seen: Seen, last_sent: Instant, last_heard: Instant,+ transfer: crate::transfer::Receiver, } impl Conn { fn new(now: Instant) -> Self {- Self { started: now, opened: false, welcomed: false, seen: Seen::default(), last_sent: now, last_heard: now }+ Self { started: now, opened: false, welcomed: false, seen: Seen::default(), last_sent: now, last_heard: now, transfer: Default::default() } } } @@ -114,11 +121,17 @@ impl<S: Socket> Driver<S> { /// A binary message from the host. pub(crate) fn message(&mut self, bytes: &[u8], now: Instant) { let Some(c) = self.conn.as_mut() else { return };- if bytes.len() > self.cfg.limits.max_server_message {+ if bytes.len() > self.cfg.limits.max_server_message.saturating_add(crate::transfer::HEADER) { return self.end(too_large(&self.cfg.limits), now); } c.last_heard = now;- let taken = match message::decode::<ServerMsg>(bytes, self.cfg.limits.max_server_message) {+ let bytes = match c.transfer.push(bytes, self.sock.chunked(), self.cfg.limits.max_server_message) {+ Ok(Some(b)) => b,+ Ok(None) => return,+ Err(crate::transfer::Error::TooLarge) => return self.end(too_large(&self.cfg.limits), now),+ Err(crate::transfer::Error::Invalid(reason)) => return self.end(Outcome::Refused(reason), now),+ };+ let taken = match message::decode::<ServerMsg>(&bytes, self.cfg.limits.max_server_message) { // The join code's answer: join the file it opens, on a new connection. Ok(ServerMsg::Resolved { file, token }) if self.cfg.code.is_some() => { return match resolved(&mut self.cfg, file, token) {@@ -158,6 +171,11 @@ impl<S: Socket> Driver<S> { /// The connection closed (`code` and `reason` from the close frame; 1006 when it just broke). pub(crate) fn closed(&mut self, code: u16, reason: &str, now: Instant) { let Some(opened) = self.conn.as_ref().map(|c| c.opened) else { return };+ if !opened && code == 1006 && self.sock.fallback_legacy() {+ self.sock.close();+ self.connect(now);+ return;+ } let outcome = match closed(code, reason) { Outcome::Retry(_) if !opened => Outcome::Retry(format!("no answer from {}", self.cfg.invite.server.display())), Outcome::Retry(_) if code == 1006 => Outcome::Retry("the connection dropped".into()),@@ -248,7 +266,7 @@ impl<S: Socket> Driver<S> { if let Some(c) = self.conn.as_mut() { c.last_sent = now; }- message::encode(msg).is_ok_and(|b| self.sock.send(&b))+ message::encode(msg).is_ok_and(|b| crate::transfer::packets(&b, self.sock.chunked()).all(|p| self.sock.send(&p))) } /// The connection ended: close it, then stop or plan the reconnect.@@ -96,6 +96,121 @@ fn ms(n: u64) -> Duration { Duration::from_millis(n) } +#[derive(Default)]+struct ChunkSocket {+ sent: Vec<Vec<u8>>,+ opens: usize,+ legacy: bool,+}++impl Socket for ChunkSocket {+ fn open(&mut self, _: &str) -> Result<(), String> {+ self.opens += 1;+ Ok(())+ }+ fn send(&mut self, bytes: &[u8]) -> bool {+ self.sent.push(bytes.to_vec());+ true+ }+ fn close(&mut self) {}+ fn chunked(&self) -> bool {+ !self.legacy+ }+ fn fallback_legacy(&mut self) -> bool {+ !std::mem::replace(&mut self.legacy, true)+ }+}++#[test]+fn browser_driver_only_exposes_a_complete_verified_large_welcome() {+ let now = Instant::now();+ let inbox = Inbox::new(Arc::new(|| {}));+ let mut d = Driver::start(ChunkSocket::default(), GuestConfig::new(invite(), "Ada".into()), inbox.clone(), now);+ d.opened(now);+ inbox.drain();+ let mut server = Server::new(Document::new("Synthetic / ".repeat(100_000)), TOKEN, Limits::default());+ let ClientMsg::Hello(h) = Client::hello(FILE, TOKEN, "Ada", "") else { panic!("hello") };+ let (_, out) = server.join(h).unwrap();+ let welcome = out+ .into_iter()+ .find_map(|o| match o {+ Outgoing::To(_, m @ ServerMsg::Welcome { .. }) => Some(m),+ _ => None,+ })+ .unwrap();+ let bytes = message::encode(&welcome).unwrap();+ let chunks: Vec<_> = crate::transfer::packets(&bytes, true).collect();+ assert!(chunks.len() > 1);+ for (i, b) in chunks.iter().enumerate() {+ d.message(b, now);+ let events = inbox.drain();+ if i + 1 < chunks.len() {+ assert!(events.is_empty(), "the engine must not see a partial snapshot");+ } else {+ assert_eq!(events.iter().filter(|e| matches!(e, Event::Sync(ServerMsg::Welcome { .. }))).count(), 1);+ }+ }+}++#[test]+fn browser_accepts_framing_overhead_at_the_logical_receive_limit() {+ let now = Instant::now();+ let inbox = Inbox::new(Arc::new(|| {}));+ let bytes = encode(&ServerMsg::Refused { reason: "R".repeat(crate::transfer::CHUNK) });+ let chunks: Vec<_> = crate::transfer::packets(&bytes, true).collect();+ assert_eq!(chunks.len(), 2);+ assert!(chunks[0].len() > bytes.len());+ let mut cfg = GuestConfig::new(invite(), "Ada".into());+ cfg.limits.max_server_message = bytes.len();+ let mut d = Driver::start(ChunkSocket::default(), cfg, inbox.clone(), now);+ d.opened(now);+ inbox.drain();+ d.message(&chunks[0], now);+ assert!(inbox.drain().is_empty());+ d.message(&chunks[1], now);+ assert!(inbox.drain().iter().any(|e| matches!(e,+ Event::Status(Status::Failed { reason }) if reason.starts_with("The host refused: ")+ )));+ assert!(d.done());+}++#[test]+fn chunked_and_legacy_receive_limits_keep_the_local_budget_diagnostic() {+ for chunked in [false, true] {+ let now = Instant::now();+ let inbox = Inbox::new(Arc::new(|| {}));+ let mut cfg = GuestConfig::new(invite(), "Ada".into());+ cfg.limits.max_server_message = crate::transfer::CHUNK + 128;+ let mut d = Driver::start(ChunkSocket::default(), cfg, inbox.clone(), now);+ d.opened(now);+ inbox.drain();+ let source = vec![1; crate::transfer::CHUNK * 3];+ let first = crate::transfer::packets(&source, chunked).next().unwrap();+ d.message(&first, now);+ let expected = TOO_LARGE_TO_JOIN.replace("{mb}", "1");+ assert!(inbox.drain().iter().any(|e| matches!(e, Event::Status(Status::Failed { reason }) if reason == &expected)));+ assert!(d.done());+ }+}++#[test]+fn browser_handshake_falls_back_once_but_never_replays_an_open_connection() {+ let now = Instant::now();+ let inbox = Inbox::new(Arc::new(|| {}));+ let mut d = Driver::start(ChunkSocket::default(), GuestConfig::new(invite(), "Ada".into()), inbox, now);+ d.closed(1006, "", now);+ assert_eq!(d.sock.opens, 2);+ assert!(d.sock.legacy);+ d.closed(1006, "", now);+ assert_eq!(d.sock.opens, 2, "no endless fallback loop");+ let inbox = Inbox::new(Arc::new(|| {}));+ let mut d = Driver::start(ChunkSocket::default(), GuestConfig::new(invite(), "Ada".into()), inbox, now);+ d.opened(now);+ d.closed(1006, "", now);+ assert_eq!(d.sock.opens, 1);+ assert!(!d.sock.legacy, "an already open connection must not be downgraded");+}+ #[test] fn hello_then_the_welcome_releases_presence() { let t0 = Instant::now();@@ -19,16 +19,16 @@ use figcraft_collab::Replica; use figcraft_collab::message::{self, Admin, AdminReply, ClientMsg, HostInfo, Presence, ServerMsg}; use figcraft_doc::Document; use figcraft_engine::live::{Event, Published, Status};+use tungstenite::Message; use tungstenite::protocol::frame::coding::CloseCode; use tungstenite::protocol::{CloseFrame, Role as WsRole, WebSocket};-use tungstenite::{Bytes, Message}; use crate::inbox::Inbox; use crate::invite::{Endpoint, Invite}; pub use crate::join::GuestConfig; use crate::join::{Next, Outcome, Resuming, Seen, Taken, after, closed, hello, resolved, take, too_large, unresolved}; use crate::manage;-use crate::net::{Duplex, NetLimits, config, open, shutdown};+use crate::net::{Duplex, NetLimits, config, shutdown}; use crate::throttle::{PRESENCE_INTERVAL, Throttle, earliest}; /// What [`Guest::publish`] uploads, and where.@@ -212,9 +212,7 @@ fn upload(job: &Publish, mut cfg: GuestConfig) -> Result<(GuestConfig, Published } fn connect(cfg: &GuestConfig) -> Result<Conn, String> {- let (rd, wr, sock) = open(&cfg.invite.server, cfg.connect_timeout, &cfg.limits)?;- let (reader, _) = tungstenite::client::client_with_config(cfg.invite.endpoint(), rd, Some(config(cfg.limits.max_server_message)))- .map_err(|e| format!("{} doesn't answer as a FigCraft session: {e}", cfg.invite.server.display()))?;+ let (reader, wr, sock) = crate::net::connect_websocket(&cfg.invite.server, cfg.connect_timeout, &cfg.limits)?; let writer = WebSocket::from_raw_socket(wr, WsRole::Client, Some(config(cfg.limits.max_client_message))); reader.get_ref().set_read_timeout(Some(cfg.limits.idle_timeout)).map_err(|e| e.to_string())?; Ok(Conn { reader, writer, sock })@@ -337,7 +335,7 @@ fn session( /// Send one message; false when the connection is gone. fn send(ws: &mut WebSocket<Duplex>, msg: &ClientMsg, last_sent: &mut Instant) -> bool { *last_sent = Instant::now();- message::encode(msg).is_ok_and(|b| ws.send(Message::Binary(Bytes::from(b))).is_ok())+ message::encode(msg).is_ok_and(|b| crate::net::send_binary(ws, &b).is_ok()) } struct ReaderCtx {@@ -350,20 +348,29 @@ struct ReaderCtx { fn read_loop(mut ws: WebSocket<Duplex>, ctx: &ReaderCtx) { let mut seen = Seen::default();+ let mut transfer = crate::transfer::Receiver::default(); let outcome = loop { match ws.read() {- Ok(Message::Binary(b)) => match message::decode::<ServerMsg>(&b, ctx.cfg.limits.max_server_message) {- Ok(m) => {- let welcomed = || {- let _ = ctx.tx.send(Cmd::Welcomed(ctx.epoch));- };- match take(m, &ctx.cfg, &ctx.inbox, &mut seen, &ctx.resume, &welcomed) {- Taken::Continue | Taken::Welcomed => {}- Taken::End(o) => break o,+ Ok(Message::Binary(b)) => {+ let b = match transfer.push(&b, ws.get_ref().chunked, ctx.cfg.limits.max_server_message) {+ Ok(Some(b)) => b,+ Ok(None) => continue,+ Err(crate::transfer::Error::TooLarge) => break too_large(&ctx.cfg.limits),+ Err(crate::transfer::Error::Invalid(reason)) => break Outcome::Refused(reason),+ };+ match message::decode::<ServerMsg>(&b, ctx.cfg.limits.max_server_message) {+ Ok(m) => {+ let welcomed = || {+ let _ = ctx.tx.send(Cmd::Welcomed(ctx.epoch));+ };+ match take(m, &ctx.cfg, &ctx.inbox, &mut seen, &ctx.resume, &welcomed) {+ Taken::Continue | Taken::Welcomed => {}+ Taken::End(o) => break o,+ } }+ Err(e) => break Outcome::Refused(format!("the host sent a message this version can't read ({e})")), }- Err(e) => break Outcome::Refused(format!("the host sent a message this version can't read ({e})")),- },+ } 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(_) => {}@@ -145,9 +145,17 @@ const BUSY: &str = "the server is busy: try again in a moment"; /// Read the next binary message (skipping pings, until `deadline`) and its size, or why the connection should end. fn next_message(ws: &mut WebSocket<Duplex>, max: usize, deadline: Option<Instant>) -> Result<Option<(ClientMsg, usize)>, String> {+ let mut transfer = crate::transfer::Receiver::default(); loop { return match ws.read() { Ok(Message::Binary(b)) => {+ let Some(b) = transfer.push(&b, ws.get_ref().chunked, max).map_err(|error| match error {+ crate::transfer::Error::TooLarge => too_large(max),+ crate::transfer::Error::Invalid(reason) => reason,+ })?+ else {+ continue;+ }; // Whole now: it counts as in flight from here, no longer as being read. ws.get_mut().read_done(); message::decode::<ClientMsg>(&b, max).map(|m| Some((m, b.len()))).map_err(|_| "that message couldn't be read".to_string())@@ -190,14 +198,28 @@ 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((mut rd, wr, sock)) = Duplex::split(stream, None) else { return };+ let Ok((mut rd, mut wr, sock)) = Duplex::split(stream, None) else { return }; // The handshake and the hello must be in by the deadline, however slowly they trickle in. rd.read_until(Some(hello_deadline)); let cfg = config(limits.max_client_message); let mut forwarded = None;+ let mut chunked = false; // The handshake callback's type is tungstenite's (its error is an HTTP response). #[allow(clippy::result_large_err)]- let read_client = |req: &Request, resp: Response| -> Result<Response, ErrorResponse> {+ let read_client = |req: &Request, mut resp: Response| -> Result<Response, ErrorResponse> {+ let protocol = req+ .headers()+ .get("Sec-WebSocket-Protocol")+ .and_then(|v| v.to_str().ok())+ .is_some_and(|v| v.split(',').any(|s| s.trim() == crate::transfer::PROTOCOL));+ let header = req.headers().get("X-FigCraft-Transfer").is_some_and(|v| v == crate::transfer::PROTOCOL);+ chunked = protocol || header;+ if protocol {+ resp.headers_mut().insert("Sec-WebSocket-Protocol", tungstenite::http::HeaderValue::from_static(crate::transfer::PROTOCOL));+ }+ if header {+ resp.headers_mut().insert("X-FigCraft-Transfer", tungstenite::http::HeaderValue::from_static(crate::transfer::PROTOCOL));+ } if proxied { forwarded = forwarded_for(req); }@@ -211,6 +233,8 @@ fn serve(mut stream: TcpStream, conn: u64, hub: &Sender<HubIn>, limits: NetLimit return; } };+ ws.get_mut().chunked = chunked;+ wr.chunked = chunked; if proxied && forwarded.is_none() { log::debug!("live session: connection {conn} came through the proxy without X-Forwarded-For"); }@@ -360,7 +384,7 @@ fn write_loop(mut ws: WebSocket<Duplex>, rx: &Receiver<Out>, sock: &TcpStream) { while let Ok(o) = rx.recv() { match o { Out::Bytes(b, _charge) => {- if ws.send(Message::Binary(b)).is_err() {+ if crate::net::send_binary(&mut ws, &b).is_err() { close = None; break; }@@ -98,7 +98,11 @@ pub(crate) fn unresolved(cfg: &GuestConfig, outcome: Outcome) -> Outcome { /// The host sent a message over this app's size limit (the file's welcome, usually): final, and /// not the host's doing. pub(crate) fn too_large(limits: &NetLimits) -> Outcome {- Outcome::Failed(TOO_LARGE_TO_JOIN.replace("{mb}", &(limits.max_server_message >> 20).to_string()))+ Outcome::Failed(receive_limit_reason(limits))+}++pub(crate) fn receive_limit_reason(limits: &NetLimits) -> String {+ TOO_LARGE_TO_JOIN.replace("{mb}", &(limits.max_server_message >> 20).to_string()) } /// The server's answer to the join code: the invite is whole now.@@ -21,6 +21,9 @@ mod join; pub mod limits; mod sessions; pub mod throttle;+mod transfer;+#[cfg(all(test, not(target_arch = "wasm32")))]+mod transfer_tests; #[cfg(not(target_arch = "wasm32"))] pub mod guest;@@ -3,6 +3,28 @@ use std::net::IpAddr; use std::time::Duration; +/// Keep the existing 1 GiB default. Larger documents require an explicit local memory budget,+/// in addition to the server operator's file limit; never accept an arbitrary server's budget.+fn receiving_limit(value: Option<&str>) -> usize {+ let default = 1usize << 30;+ let Some(mb) = value.and_then(|v| v.parse::<u64>().ok()) else { return default };+ if !(1..=4096).contains(&mb) {+ return default;+ }+ mb.checked_mul(1 << 20).and_then(|n| usize::try_from(n).ok()).filter(|n| *n <= isize::MAX as usize).unwrap_or(default)+}++fn configured_receiving_limit() -> usize {+ #[cfg(not(target_arch = "wasm32"))]+ {+ receiving_limit(std::env::var("FIGCRAFT_MAX_RECEIVE_MB").ok().as_deref())+ }+ #[cfg(target_arch = "wasm32")]+ {+ receiving_limit(None)+ }+}+ /// Transport limits (the collaboration core has its own: [`figcraft_collab::Limits`]). #[derive(Clone, Copy, Debug, PartialEq)] pub struct NetLimits {@@ -12,9 +34,10 @@ pub struct NetLimits { /// before: its hello. pub max_client_message: usize, pub max_hello: usize,- /// Largest message a client accepts from the host (a welcome carries the whole document): 1 GiB,- /// room for any file up to 1023 MiB and well past the servers' defaults, while a hostile server- /// (any link can name one) can't make the app or a browser tab buffer more than that.+ /// Largest logical message a client accepts, including a reassembled transfer: 1 GiB by+ /// default. Native clients may explicitly choose 1..=4096 MiB with+ /// `FIGCRAFT_MAX_RECEIVE_MB`, within the platform's allocation range. The browser keeps+ /// its 1 GiB budget. A server cannot enlarge this local memory budget. pub max_server_message: usize, /// Messages a client may send per second on average, and in a burst; beyond that it is /// disconnected.@@ -69,7 +92,7 @@ impl Default for NetLimits { max_connections: 64, max_client_message: figcraft_collab::Limits::default().max_message_bytes, max_hello: 64 << 10,- max_server_message: 1 << 30,+ max_server_message: configured_receiving_limit(), rate: 100.0, burst: 400.0, hello_timeout: Duration::from_secs(10),@@ -93,6 +116,23 @@ impl Default for NetLimits { } } +#[cfg(test)]+mod receiving_tests {+ use super::receiving_limit;++ #[test]+ fn larger_receive_budgets_are_explicit_bounded_and_platform_checked() {+ assert_eq!(receiving_limit(None), 1 << 30);+ for bad in ["0", "4097", "-1", "invalid", "18446744073709551615"] {+ assert_eq!(receiving_limit(Some(bad)), 1 << 30);+ }+ assert_eq!(receiving_limit(Some("512")), 512 << 20);+ if usize::BITS == 64 {+ assert_eq!(receiving_limit(Some("2048")), 2usize << 30);+ }+ }+}+ #[cfg(test)] mod tests { use super::*;@@ -11,10 +11,10 @@ use figcraft_collab::message::{ PluginReply, PluginReq, Resolve, ServerMsg, }; use figcraft_engine::live::{Shared, SharedLink};-use tungstenite::{Bytes, Message, WebSocket};+use tungstenite::{Message, WebSocket}; use crate::invite::{Endpoint, Invite};-use crate::net::{Duplex, NetLimits, config, open};+use crate::net::{Duplex, NetLimits}; type Ws = WebSocket<Duplex>; @@ -199,9 +199,7 @@ pub fn plugins(server: &Endpoint, key: &str, req: PluginReq, limits: &NetLimits, /// A WebSocket to `server` that waits for answers up to the request-reply timeout (a large upload /// takes the server a while to keep). fn connect(server: &Endpoint, limits: &NetLimits, timeout: Duration) -> Result<Ws, String> {- let (stream, _, _) = open(server, timeout, limits)?;- let (ws, _) = tungstenite::client::client_with_config(server.url(), stream, Some(config(limits.max_server_message)))- .map_err(|e| format!("{} doesn't answer as a FigCraft server: {e}", server.display()))?;+ let (ws, _, _) = crate::net::connect_websocket(server, timeout, limits)?; // A publish uploads a whole file; a large project on a slow link can take several minutes to // receive and answer. The wait uses `request_reply_timeout`, not the session `idle_timeout`. ws.get_ref().set_read_timeout(Some(limits.request_reply_timeout)).map_err(|e| e.to_string())?;@@ -222,7 +220,7 @@ fn close(mut ws: Ws) { fn send(ws: &mut Ws, msg: &ClientMsg) -> Result<(), String> { let bytes = message::encode(msg).map_err(|e| e.to_string())?;- ws.send(Message::Binary(Bytes::from(bytes))).map_err(|e| format!("the connection to the server dropped ({e})"))+ crate::net::send_binary(ws, &bytes).map_err(|e| format!("the connection to the server dropped ({e})")) } /// How an error starts when the server said no (then its reason), rather than not answering.@@ -230,20 +228,31 @@ pub(crate) const REFUSED: &str = "the server refused: "; /// The next answer `pick` takes, or why there is none. fn reply<T>(ws: &mut Ws, limits: &NetLimits, pick: impl Fn(ServerMsg) -> Option<T>) -> Result<T, String> {+ let mut transfer = crate::transfer::Receiver::default(); loop { return match ws.read() {- Ok(Message::Binary(b)) => match message::decode::<ServerMsg>(&b, limits.max_server_message) {- Ok(ServerMsg::Refused { reason }) => Err(format!("{REFUSED}{reason}")),- Ok(ServerMsg::Pong) => continue,- Ok(m) => pick(m).ok_or_else(|| "the server sent something other than an answer".into()),- Err(e) => Err(format!("the server sent a message this version can't read ({e})")),- },+ Ok(Message::Binary(b)) => {+ let Some(b) = transfer.push(&b, ws.get_ref().chunked, limits.max_server_message).map_err(|error| match error {+ crate::transfer::Error::TooLarge => crate::join::receive_limit_reason(limits),+ crate::transfer::Error::Invalid(reason) => reason,+ })?+ else {+ continue;+ };+ match message::decode::<ServerMsg>(&b, limits.max_server_message) {+ Ok(ServerMsg::Refused { reason }) => Err(format!("{REFUSED}{reason}")),+ Ok(ServerMsg::Pong) => continue,+ Ok(m) => pick(m).ok_or_else(|| "the server sent something other than an answer".into()),+ Err(e) => Err(format!("the server sent a message this version can't read ({e})")),+ }+ } Ok(Message::Close(f)) => Err(match f.map(|f| f.reason.to_string()).filter(|r| !r.is_empty()) { Some(r) => format!("the server closed the connection: {r}"), None => "the server closed the connection".into(), }), Ok(Message::Text(_)) => Err("the server sent a text message".into()), Ok(_) => continue,+ Err(tungstenite::Error::Capacity(_)) => Err(crate::join::receive_limit_reason(limits)), Err(e) => Err(format!("no answer from the server ({e})")), }; }@@ -252,6 +261,7 @@ fn reply<T>(ws: &mut Ws, limits: &NetLimits, pick: impl Fn(ServerMsg) -> Option< #[cfg(test)] mod tests { use super::*;+ use tungstenite::Bytes; /// Whether `close` returns within 3 s against a server that never answers the close and does /// `serve` instead, with reads waiting up to `idle`.@@ -20,9 +20,45 @@ use tungstenite::protocol::WebSocketConfig; use crate::invite::Endpoint; use crate::tls::{self, Tls}; -/// tungstenite's settings for messages up to `max` bytes.+/// Allow framing overhead on the wire; the receiver still enforces the logical `max`. pub fn config(max: usize) -> WebSocketConfig {- WebSocketConfig::default().max_message_size(Some(max)).max_frame_size(Some(max))+ let packet_max = max.saturating_add(crate::transfer::HEADER);+ WebSocketConfig::default().max_message_size(Some(packet_max)).max_frame_size(Some(packet_max))+}++/// Request the optional chunked transport without changing the collaboration protocol.+pub(crate) fn request(url: &str) -> Result<tungstenite::handshake::client::Request, String> {+ use tungstenite::client::IntoClientRequest;+ let mut request = url.into_client_request().map_err(|e| e.to_string())?;+ request.headers_mut().insert("X-FigCraft-Transfer", tungstenite::http::HeaderValue::from_static(crate::transfer::PROTOCOL));+ Ok(request)+}++pub(crate) fn chunked(response: &tungstenite::handshake::client::Response) -> bool {+ response.headers().get("X-FigCraft-Transfer").is_some_and(|p| p == crate::transfer::PROTOCOL)+}++/// An old server ignores the optional header: the same connection uses legacy messages.+/// No extra connection, authentication attempt or document operation is needed for fallback.+pub(crate) fn connect_websocket(+ server: &Endpoint,+ timeout: Duration,+ limits: &NetLimits,+) -> Result<(tungstenite::WebSocket<Duplex>, Duplex, TcpStream), String> {+ let (rd, mut wr, sock) = open(server, timeout, limits)?;+ let (mut ws, response) = tungstenite::client::client_with_config(request(&server.url())?, rd, Some(config(limits.max_server_message)))+ .map_err(|e| format!("{} doesn't answer as a FigCraft server: {e}", server.display()))?;+ wr.chunked = chunked(&response);+ ws.get_mut().chunked = wr.chunked;+ Ok((ws, wr, sock))+}++/// Keep the encoded operation atomic, while bounding each WebSocket message to one piece.+pub(crate) fn send_binary(ws: &mut tungstenite::WebSocket<Duplex>, bytes: &[u8]) -> Result<(), tungstenite::Error> {+ for packet in crate::transfer::packets(bytes, ws.get_ref().chunked) {+ ws.send(tungstenite::Message::Binary(packet.into()))?;+ }+ Ok(()) } /// One side (read or write) of a connection: reads from its own handle, writes whole buffers@@ -33,6 +69,7 @@ pub struct Duplex { tls: Option<Tls>, /// What the bytes read count against (a server's reader), if anything. meter: Option<Meter>,+ pub(crate) chunked: bool, /// When reading must be over, however slowly the bytes come ([`Duplex::read_until`]). deadline: Option<Instant>, }@@ -41,8 +78,8 @@ impl Duplex { /// Two handles on `stream` (for the reader and the writer) and one to shut it down with. pub fn split(stream: TcpStream, tls: Option<Tls>) -> io::Result<(Duplex, Duplex, TcpStream)> { let write = Arc::new(Mutex::new(stream.try_clone()?));- let a = Duplex { read: stream.try_clone()?, write: write.clone(), tls: tls.clone(), meter: None, deadline: None };- let b = Duplex { read: stream.try_clone()?, write, tls, meter: None, deadline: None };+ let a = Duplex { read: stream.try_clone()?, write: write.clone(), tls: tls.clone(), meter: None, deadline: None, chunked: false };+ let b = Duplex { read: stream.try_clone()?, write, tls, meter: None, deadline: None, chunked: false }; Ok((a, b, stream)) } new file mode 100644@@ -0,0 +1,249 @@+//! Negotiated, bounded pieces of one collaboration message. Document operations still become+//! visible atomically, after the entire message has arrived and passed its SHA-256 check.++use sha2::{Digest, Sha256};++pub(crate) const PROTOCOL: &str = "figcraft.chunks.v1";+pub(crate) const CHUNK: usize = 1 << 20;+pub(crate) const HEADER: usize = 49;+const VERSION: u8 = 2;++#[derive(Debug, PartialEq, Eq)]+pub(crate) enum Error {+ TooLarge,+ Invalid(String),+}++impl From<&str> for Error {+ fn from(reason: &str) -> Self {+ Self::Invalid(reason.to_string())+ }+}++impl std::fmt::Display for Error {+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {+ match self {+ Self::TooLarge => f.write_str("the collaboration message exceeds the receiving limit"),+ Self::Invalid(reason) => f.write_str(reason),+ }+ }+}++impl std::error::Error for Error {}++/// Legacy peers keep wire version 1. Only a negotiated connection sends version 2 pieces.+pub(crate) fn packets(bytes: &[u8], chunked: bool) -> Packets<'_> {+ let chunked = chunked && bytes.len() > CHUNK;+ let hash = if chunked { Sha256::digest(bytes).into() } else { [0; 32] };+ Packets { bytes, offset: 0, chunked, hash }+}++pub(crate) struct Packets<'a> {+ bytes: &'a [u8],+ offset: usize,+ chunked: bool,+ hash: [u8; 32],+}++impl Iterator for Packets<'_> {+ type Item = Vec<u8>;++ fn next(&mut self) -> Option<Self::Item> {+ if self.offset >= self.bytes.len() {+ return None;+ }+ if !self.chunked {+ self.offset = self.bytes.len();+ return Some(self.bytes.to_vec());+ }+ let end = self.offset.saturating_add(CHUNK).min(self.bytes.len());+ let mut out = Vec::with_capacity(HEADER + end - self.offset);+ out.push(VERSION);+ out.extend_from_slice(&(self.bytes.len() as u64).to_le_bytes());+ out.extend_from_slice(&(self.offset as u64).to_le_bytes());+ out.extend_from_slice(&self.hash);+ out.extend_from_slice(self.bytes.get(self.offset..end)?);+ self.offset = end;+ Some(out)+ }+}++#[derive(Default)]+pub(crate) struct Receiver {+ bytes: Vec<u8>,+ total: Option<usize>,+ hash: [u8; 32],+ digest: Sha256,+}++impl Receiver {+ /// No allocation follows an untrusted size declaration. Storage grows only with bytes+ /// actually received, inside `max`; partial messages belong to this connection only.+ pub(crate) fn push(&mut self, bytes: &[u8], negotiated: bool, max: usize) -> Result<Option<Vec<u8>>, Error> {+ if bytes.first() != Some(&VERSION) {+ if self.total.is_some() {+ return Err("a collaboration transfer was interrupted by another message".into());+ }+ if bytes.len() > max {+ return Err(Error::TooLarge);+ }+ return Ok(Some(bytes.to_vec()));+ }+ if !negotiated {+ return Err("chunked collaboration was not negotiated".into());+ }+ if bytes.len() <= HEADER || bytes.len() > HEADER + CHUNK {+ return Err("invalid collaboration chunk length".into());+ }+ let number = |range: std::ops::Range<usize>| -> Result<u64, Error> {+ let array: [u8; 8] = bytes.get(range).and_then(|v| v.try_into().ok()).ok_or("invalid chunk header")?;+ Ok(u64::from_le_bytes(array))+ };+ let total = usize::try_from(number(1..9)?).map_err(|_| Error::TooLarge)?;+ if total > max {+ return Err(Error::TooLarge);+ }+ let offset = usize::try_from(number(9..17)?).map_err(|_| Error::from("chunk offset does not fit this platform"))?;+ let hash: [u8; 32] = bytes.get(17..HEADER).and_then(|v| v.try_into().ok()).ok_or("invalid chunk checksum")?;+ let payload = bytes.get(HEADER..).ok_or("invalid chunk payload")?;+ if total == 0 || offset != self.bytes.len() || payload.len() > total.saturating_sub(offset) {+ return Err("collaboration chunk is out of order".into());+ }+ if self.total.is_some_and(|n| n != total) || (self.total.is_some() && self.hash != hash) {+ return Err("collaboration transfer metadata changed".into());+ }+ self.total = Some(total);+ self.hash = hash;+ if self.bytes.capacity() - self.bytes.len() < payload.len() {+ // Retain geometric growth without requesting capacity beyond the declared total.+ let grow = self.bytes.len().max(payload.len()).min(total - self.bytes.len());+ self.bytes.try_reserve_exact(grow).map_err(|_| "not enough memory for the collaboration transfer")?;+ }+ self.bytes.extend_from_slice(payload);+ self.digest.update(payload);+ if self.bytes.len() != total {+ return Ok(None);+ }+ let digest: [u8; 32] = std::mem::take(&mut self.digest).finalize().into();+ if digest != self.hash {+ return Err("collaboration transfer checksum mismatch".into());+ }+ self.total = None;+ Ok(Some(std::mem::take(&mut self.bytes)))+ }+}++#[cfg(test)]+mod tests {+ use super::*;++ #[test]+ fn reassembly_growth_stays_within_the_total_and_reuses_spare_capacity() {+ let source = vec![37; CHUNK * 8 + 1];+ let mut rx = Receiver::default();+ let mut allocations = 0;+ for packet in packets(&source, true) {+ let (len, capacity) = (rx.bytes.len(), rx.bytes.capacity());+ let payload_len = packet.len() - HEADER;+ let complete = rx.push(&packet, true, source.len()).unwrap();+ let current_capacity = complete.as_ref().map_or(rx.bytes.capacity(), Vec::capacity);+ assert!(current_capacity <= source.len());+ if capacity - len >= payload_len {+ assert_eq!(current_capacity, capacity, "reuse space already allocated");+ }+ allocations += usize::from(current_capacity != capacity);+ if let Some(bytes) = complete {+ assert_eq!(bytes, source);+ }+ }+ assert!(allocations <= 5, "bounded geometric growth, not a reallocation per chunk");+ }++ #[test]+ fn large_message_is_atomic_and_every_piece_is_bounded() {+ let source = vec![37; CHUNK * 2 + 31];+ let chunks: Vec<_> = packets(&source, true).collect();+ assert_eq!(chunks.len(), 3);+ let mut rx = Receiver::default();+ for (i, p) in chunks.iter().enumerate() {+ assert!(p.len() <= CHUNK + HEADER);+ let received = rx.push(p, true, source.len()).unwrap();+ if i == chunks.len() - 1 {+ assert_eq!(received, Some(source.clone()));+ } else {+ assert_eq!(received, None, "no partial document can reach the engine");+ }+ }+ assert!(rx.bytes.is_empty());+ }++ #[test]+ fn legacy_peer_and_small_messages_keep_the_original_wire_format() {+ for (bytes, enabled) in [(vec![1, 2, 3], true), (vec![1; CHUNK + 1], false)] {+ let p: Vec<_> = packets(&bytes, enabled).collect();+ assert_eq!(p, vec![bytes.clone()]);+ assert_eq!(Receiver::default().push(&p[0], false, bytes.len()).unwrap(), Some(bytes));+ }+ }++ #[test]+ fn receive_budget_errors_are_distinct_from_invalid_frames() {+ let source = vec![1; CHUNK + 1];+ let p: Vec<_> = packets(&source, true).collect();+ let mut rx = Receiver::default();+ assert_eq!(rx.push(&p[0], true, CHUNK), Err(Error::TooLarge));+ assert!(rx.bytes.is_empty());+ assert_eq!(Receiver::default().push(&source, false, CHUNK), Err(Error::TooLarge));+ assert!(matches!(Receiver::default().push(&p[1], true, source.len()), Err(Error::Invalid(_))));+ let mut zero = p[0].clone();+ zero[1..9].copy_from_slice(&0u64.to_le_bytes());+ assert!(matches!(Receiver::default().push(&zero, true, source.len()), Err(Error::Invalid(_))));+ let mut short = p[0].clone();+ short[1..9].copy_from_slice(&1u64.to_le_bytes());+ assert!(matches!(Receiver::default().push(&short, true, source.len()), Err(Error::Invalid(_))));+ let mut huge = p[0].clone();+ huge[1..9].copy_from_slice(&u64::MAX.to_le_bytes());+ assert_eq!(Receiver::default().push(&huge, true, CHUNK), Err(Error::TooLarge));+ }++ #[test]+ fn hostile_headers_are_refused_before_allocating() {+ let source = vec![4; CHUNK + 1];+ let p: Vec<_> = packets(&source, true).collect();+ let mut rx = Receiver::default();+ assert!(rx.push(&p[0], true, CHUNK).is_err());+ assert!(rx.bytes.is_empty());+ assert!(Receiver::default().push(&p[0], false, source.len()).is_err());+ assert!(Receiver::default().push(&p[1], true, source.len()).is_err());+ }++ #[test]+ fn missing_duplicate_corrupt_and_interleaved_pieces_do_not_commit() {+ let source = vec![4; CHUNK + 1];+ let p: Vec<_> = packets(&source, true).collect();+ let mut rx = Receiver::default();+ assert!(rx.push(&p[0], true, source.len()).unwrap().is_none());+ assert!(rx.push(&p[0], true, source.len()).is_err());+ let mut rx = Receiver::default();+ rx.push(&p[0], true, source.len()).unwrap();+ assert!(rx.push(&[1, 4], true, source.len()).is_err());+ let mut rx = Receiver::default();+ rx.push(&p[0], true, source.len()).unwrap();+ let mut corrupted = p[1].clone();+ *corrupted.last_mut().unwrap() ^= 1;+ assert!(rx.push(&corrupted, true, source.len()).is_err());+ }++ #[test]+ fn a_new_connection_does_not_inherit_an_incomplete_transfer() {+ let source = vec![4; CHUNK + 1];+ let p: Vec<_> = packets(&source, true).collect();+ let mut rx = Receiver::default();+ rx.push(&p[0], true, source.len()).unwrap();+ drop(rx);+ let mut rx = Receiver::default();+ assert!(rx.push(&p[1], true, source.len()).is_err());+ assert!(rx.push(&p[0], true, source.len()).unwrap().is_none());+ assert_eq!(rx.push(&p[1], true, source.len()).unwrap(), Some(source));+ }+}new file mode 100644@@ -0,0 +1,266 @@+//! Real sockets: the same document is too large for a legacy message budget, but a negotiated+//! connection receives bounded pieces and exposes exactly one complete document to the engine.++use std::net::SocketAddr;+use std::time::Duration;++use figcraft_collab::message::{self, ServerMsg};+use figcraft_collab::{Client, Limits};+use figcraft_doc::Document;+use figcraft_engine::live::Event;+use tungstenite::Message;++use crate::guest::{Guest, GuestConfig};+use crate::host::{Host, HostConfig};+use crate::invite::Invite;+use crate::loopback_tests::Probe;+use crate::net::{NetLimits, send_binary};+use crate::transfer::{CHUNK, Receiver};++const TOKEN: &str = "SyntheticTransferSecret";++#[test]+fn framing_fits_a_tight_websocket_budget_without_relaxing_the_logical_limit() {+ use std::net::{TcpListener, TcpStream};+ use tungstenite::protocol::{Role, WebSocket};++ let payload = vec![7; CHUNK + 1];+ let packets: Vec<_> = crate::transfer::packets(&payload, true).collect();+ assert_eq!(packets.len(), 2);+ assert!(packets[0].len() > payload.len());+ let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap();+ let addr = listener.local_addr().unwrap();+ let sender = std::thread::spawn(move || {+ let (stream, _) = listener.accept().unwrap();+ stream.set_write_timeout(Some(Duration::from_secs(5))).unwrap();+ let mut ws = WebSocket::from_raw_socket(stream, Role::Server, None);+ for packet in packets {+ ws.send(Message::Binary(packet.into())).unwrap();+ }+ });+ let stream = TcpStream::connect(addr).unwrap();+ stream.set_read_timeout(Some(Duration::from_secs(5))).unwrap();+ let mut ws = WebSocket::from_raw_socket(stream, Role::Client, Some(crate::net::config(payload.len())));+ let mut receiver = Receiver::default();+ for i in 0..2 {+ let Message::Binary(packet) = ws.read().unwrap() else { panic!("expected a chunk") };+ if i == 0 {+ assert_eq!(Receiver::default().push(&packet, true, CHUNK), Err(crate::transfer::Error::TooLarge));+ }+ let complete = receiver.push(&packet, true, payload.len()).unwrap();+ if i == 0 {+ assert!(complete.is_none());+ } else {+ assert_eq!(complete.unwrap(), payload);+ }+ }+ sender.join().unwrap();+}++fn host() -> std::io::Result<(Host, Invite, String)> {+ host_with_name("Synthetic document / ".repeat(80_000))+}++fn host_with_name(name: String) -> std::io::Result<(Host, Invite, String)> {+ let probe = Probe::new();+ let host = Host::start(+ HostConfig {+ addr: SocketAddr::from(([127, 0, 0, 1], 0)),+ any_port_if_taken: false,+ file: "large-test".into(),+ token: TOKEN.into(),+ doc: Document::new(&name),+ name: "Host".into(),+ limits: NetLimits::default(),+ core: Limits::default(),+ },+ probe.inbox,+ )?;+ let invite = Invite::new(host.addr().to_string(), "large-test", TOKEN);+ Ok((host, invite, name))+}++#[test]+#[ignore = "explicit large-memory transport regression; run separately"]+fn sixty_five_mib_document_crosses_real_socket() {+ // Synthetic content only. This exercises transport above 64 MiB, not the separate+ // persistent server's file quota, which an operator must explicitly configure.+ let (_host, invite, name) = host_with_name("S".repeat(65 * CHUNK)).unwrap();+ // Model the documented Cloudflare Worker 32 MiB message cap, not an asserted+ // configuration of collab.craftapps.dev. This is a real local socket, not a live Worker.+ let wire_limit = 32 * CHUNK;+ let wire_config = || tungstenite::protocol::WebSocketConfig::default().max_message_size(Some(wire_limit)).max_frame_size(Some(wire_limit));+ let stream = std::net::TcpStream::connect(&invite.server.host).unwrap();+ stream.set_read_timeout(Some(Duration::from_secs(30))).unwrap();+ let (mut legacy, _) = tungstenite::client::client_with_config(invite.endpoint(), stream, Some(wire_config())).unwrap();+ legacy.send(Message::Binary(message::encode(&Client::hello("large-test", TOKEN, "Legacy cap test", "")).unwrap().into())).unwrap();+ assert!(matches!(legacy.read(), Err(tungstenite::Error::Capacity(_))));+ drop(legacy);+ let limits = NetLimits::default();+ let (mut ws, _, _) = crate::net::connect_websocket(&invite.server, Duration::from_secs(30), &limits).unwrap();+ ws.set_config(|c| *c = wire_config());+ send_binary(&mut ws, &message::encode(&Client::hello("large-test", TOKEN, "Large test", "")).unwrap()).unwrap();+ let mut rx = Receiver::default();+ let mut pieces = 0;+ loop {+ let Message::Binary(bytes) = ws.read().unwrap() else { continue };+ assert!(bytes.len() <= CHUNK + 49 && bytes.len() < wire_limit);+ pieces += 1;+ let Some(bytes) = rx.push(&bytes, true, limits.max_server_message).unwrap() else { continue };+ assert!(bytes.len() > 64 * CHUNK);+ let welcome = message::decode::<ServerMsg>(&bytes, limits.max_server_message).unwrap();+ assert_eq!(Client::join(welcome).unwrap().document().name, name);+ break;+ }+ assert!(pieces > 64);+ eprintln!("65 MiB local socket: legacy rejected by 32 MiB message cap; chunked welcome verified in {pieces} bounded packets");+ let _ = ws.close(None);+}++#[test]+fn legacy_default_receives_messages_larger_than_one_mib() {+ let (_host, invite, name) = host().unwrap();+ let stream = std::net::TcpStream::connect(&invite.server.host).unwrap();+ stream.set_read_timeout(Some(Duration::from_secs(5))).unwrap();+ let limits = NetLimits::default();+ let (mut ws, _) =+ tungstenite::client::client_with_config(invite.endpoint(), stream, Some(crate::net::config(limits.max_server_message))).unwrap();+ ws.send(Message::Binary(message::encode(&Client::hello("large-test", TOKEN, "Default legacy", "")).unwrap().into())).unwrap();+ let Message::Binary(bytes) = ws.read().unwrap() else { panic!("expected a legacy welcome") };+ assert!(bytes.len() > CHUNK);+ assert_eq!(Client::join(message::decode::<ServerMsg>(&bytes, limits.max_server_message).unwrap()).unwrap().document().name, name);+ let _ = ws.close(None);+}++#[test]+fn large_welcome_crosses_real_socket_in_bounded_pieces() {+ let (_host, invite, name) = host().unwrap();+ let limits = NetLimits::default();+ let (mut ws, _, _) = crate::net::connect_websocket(&invite.server, Duration::from_secs(5), &limits).unwrap();+ assert!(ws.get_ref().chunked);+ ws.set_config(|c| *c = crate::net::config(CHUNK + 49));+ let hello = message::encode(&Client::hello("large-test", TOKEN, "Test", "")).unwrap();+ send_binary(&mut ws, &hello).unwrap();+ let mut rx = Receiver::default();+ let mut pieces = 0;+ loop {+ let Message::Binary(bytes) = ws.read().unwrap() else { continue };+ assert!(bytes.len() <= CHUNK + 49);+ pieces += 1;+ if let Some(bytes) = rx.push(&bytes, true, limits.max_server_message).unwrap() {+ let welcome = message::decode::<ServerMsg>(&bytes, limits.max_server_message).unwrap();+ assert_eq!(Client::join(welcome).unwrap().document().name, name);+ break;+ }+ }+ assert!(pieces > 1);+ let _ = ws.close(None);+}++#[test]+fn reproduces_the_legacy_whole_message_limit_on_the_same_document() {+ let (_host, invite, _) = host().unwrap();+ let stream = std::net::TcpStream::connect(&invite.server.host).unwrap();+ stream.set_read_timeout(Some(Duration::from_secs(5))).unwrap();+ let (mut ws, _) = tungstenite::client::client_with_config(invite.endpoint(), stream, Some(crate::net::config(CHUNK + 49))).unwrap();+ ws.send(Message::Binary(message::encode(&Client::hello("large-test", TOKEN, "Legacy", "")).unwrap().into())).unwrap();+ assert!(matches!(ws.read(), Err(tungstenite::Error::Capacity(_))), "the legacy welcome cannot fit the very same per-message budget");+}++#[test]+fn large_edit_uploads_and_is_confirmed_as_one_operation() {+ let (_host, invite, _) = host().unwrap();+ let limits = NetLimits::default();+ let (mut ws, _, _) = crate::net::connect_websocket(&invite.server, Duration::from_secs(5), &limits).unwrap();+ send_binary(&mut ws, &message::encode(&Client::hello("large-test", TOKEN, "Edit", "")).unwrap()).unwrap();+ let mut rx = Receiver::default();+ let mut client = loop {+ let Message::Binary(b) = ws.read().unwrap() else { continue };+ let Some(b) = rx.push(&b, true, limits.max_server_message).unwrap() else { continue };+ let m = message::decode::<ServerMsg>(&b, limits.max_server_message).unwrap();+ if matches!(m, ServerMsg::Welcome { .. }) {+ break Client::join(m).unwrap();+ }+ };+ let mut doc = client.document().clone();+ doc.name = "Changed document / ".repeat(90_000);+ let edit = client.edit(&doc).unwrap().unwrap();+ let bytes = message::encode(&edit).unwrap();+ assert!(bytes.len() > CHUNK);+ send_binary(&mut ws, &bytes).unwrap();+ loop {+ let Message::Binary(b) = ws.read().unwrap() else { continue };+ let Some(b) = rx.push(&b, true, limits.max_server_message).unwrap() else { continue };+ if let ServerMsg::Changes(accepted) = message::decode::<ServerMsg>(&b, limits.max_server_message).unwrap() {+ assert_eq!(accepted.seq, 1, "the entire large edit is one accepted batch");+ break;+ }+ }+ let _ = ws.close(None);+}++#[test]+fn native_guest_receives_the_complete_large_document() {+ let (_host, invite, name) = host().unwrap();+ let mut probe = Probe::new();+ let _guest = Guest::start(GuestConfig::new(invite, "Test".into()), probe.inbox.clone()).unwrap();+ probe.wait("complete chunked welcome", |events| events.iter().any(|e| matches!(e, Event::Sync(ServerMsg::Welcome { .. }))));+ let welcome = probe+ .events+ .into_iter()+ .find_map(|e| match e {+ Event::Sync(m @ ServerMsg::Welcome { .. }) => Some(m),+ _ => None,+ })+ .unwrap();+ assert_eq!(Client::join(welcome).unwrap().document().name, name);+}++#[test]+fn a_chunked_welcome_above_the_budget_is_a_local_size_failure_not_a_host_refusal() {+ let (_host, invite, _) = host().unwrap();+ let mut cfg = GuestConfig::new(invite, "Budget test".into());+ cfg.limits.max_server_message = CHUNK + 128;+ let expected = figcraft_engine::live::TOO_LARGE_TO_JOIN.replace("{mb}", "1");+ let mut probe = Probe::new();+ let _guest = Guest::start(cfg, probe.inbox.clone()).unwrap();+ probe.wait("local receive-budget failure", |events| {+ events.iter().any(|e| matches!(e, Event::Status(figcraft_engine::live::Status::Failed { reason }) if reason == &expected))+ });+ assert!(!probe.events.iter().any(|e| matches!(e, Event::Sync(ServerMsg::Welcome { .. }))));+}++#[test]+fn legacy_client_is_not_sent_chunk_envelopes() {+ let (_host, invite, _) = host().unwrap();+ let stream = std::net::TcpStream::connect(&invite.server.host).unwrap();+ stream.set_read_timeout(Some(Duration::from_secs(5))).unwrap();+ let (mut ws, _) = tungstenite::client::client(invite.endpoint(), stream).unwrap();+ let hello = message::encode(&Client::hello("large-test", TOKEN, "Legacy", "")).unwrap();+ ws.send(Message::Binary(hello.into())).unwrap();+ let Message::Binary(bytes) = ws.read().unwrap() else { panic!("expected welcome") };+ assert_eq!(bytes.first(), Some(&message::WIRE_VERSION));+ assert!(bytes.len() > CHUNK);+ assert!(matches!(message::decode::<ServerMsg>(&bytes, 8 * CHUNK).unwrap(), ServerMsg::Welcome { .. }));+ let _ = ws.close(None);+}++#[test]+fn new_client_uses_legacy_on_the_same_connection_when_the_extension_is_not_selected() {+ let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();+ let address = listener.local_addr().unwrap();+ let server = std::thread::spawn(move || {+ // The old server ignores the optional header; no second handshake is necessary.+ for _ in 0..1 {+ let (stream, _) = listener.accept().unwrap();+ stream.set_read_timeout(Some(Duration::from_secs(5))).unwrap();+ let mut ws = tungstenite::accept(stream).unwrap();+ let _ = ws.read();+ }+ });+ let endpoint = Invite::new(address.to_string(), "test", TOKEN).server;+ let (ws, _, _) = crate::net::connect_websocket(&endpoint, Duration::from_secs(5), &NetLimits::default()).unwrap();+ assert!(!ws.get_ref().chunked);+ drop(ws);+ server.join().unwrap();+}@@ -90,12 +90,14 @@ struct Handlers { struct BrowserSocket { id: u64, ws: Option<(WebSocket, Handlers)>,+ legacy: bool, } impl Socket for BrowserSocket { fn open(&mut self, url: &str) -> Result<(), String> { self.close();- let ws = WebSocket::new(url).map_err(|e| format!("the browser won't open {url}: {}", js_error(&e)))?;+ let socket = if self.legacy { WebSocket::new(url) } else { WebSocket::new_with_str(url, crate::transfer::PROTOCOL) };+ let ws = socket.map_err(|e| format!("the browser won't open {url}: {}", js_error(&e)))?; ws.set_binary_type(BinaryType::Arraybuffer); let id = self.id; let open = Closure::<dyn FnMut()>::new(move || with_driver(id, |d, now| d.opened(now)));@@ -121,6 +123,16 @@ impl Socket for BrowserSocket { self.ws.as_ref().is_some_and(|(ws, _)| ws.ready_state() == WebSocket::OPEN && ws.send_with_u8_array(bytes).is_ok()) } + fn chunked(&self) -> bool {+ self.ws.as_ref().is_some_and(|(ws, _)| ws.protocol() == crate::transfer::PROTOCOL)+ }++ fn fallback_legacy(&mut self) -> bool {+ let retry = !self.legacy;+ self.legacy = true;+ retry+ }+ fn close(&mut self) { let Some((ws, _handlers)) = self.ws.take() else { return }; ws.set_onopen(None);@@ -175,7 +187,7 @@ impl LiveLink for NetLink { let id = NEXT_ID.with(|n| n.replace(n.get() + 1)); let inbox = self.sessions.inbox(); let cfg = GuestConfig { code, ..GuestConfig::new(invite, name.to_string()) };- let driver = Driver::start(BrowserSocket { id, ws: None }, cfg, inbox.clone(), Instant::now());+ let driver = Driver::start(BrowserSocket { id, ws: None, legacy: false }, cfg, inbox.clone(), Instant::now()); let timer = Closure::<dyn FnMut()>::new(move || with_driver(id, |d, now| d.tick(now))); let mut g = Guest { driver, timer, timeout: None }; g.arm();@@ -82,6 +82,7 @@ input. | Date | Change | Summary | |---|---|---|+| 2026-10-22 | minor | Negotiated bounded chunk transport with checksummed reassembly, legacy fallback and configurable receive budgets; inline images retained. Readiness estimates unchanged. | | 2026-10-18 | minor | Added `vectordrawable` (L4), the Android Vector Drawable (XML) writer | | 2026-10-11 | minor | `collab-net` hosts through one hub for the app's sessions and the standalone server; added the `figcraft-server` app | | 2026-10-10 | minor | Merged main after #101–#103: WASM plugins (plugins area 0% → 35% breadth, 20% ready; ecosystem 5% → 20%) and the multiplayer core without networking (multiplayer 5% → 15% breadth, 5% ready; collaboration 8% → 12%); menu breadth 97.5% → 98.5% (200/203); ready for real work ~46% → ~47%; full parity 420–790 h → 390–745 h; beta 210–380 h → 200–365 h. Fixed the stated weight sums (areas 1–16 sum to 94, total 113; percentages were already computed with those) |@@ -1,5 +1,64 @@ # Real-time collaboration protocol (draft) +## Chunked transport extension (2026-10-20)++### Motivation and evidence (2026-10-27)++There is no demonstrated universal 1 MiB WebSocket limit in FigCraft or common proxies.+The native client defaults to a 1 GiB logical receive budget, and a synthetic legacy welcome+larger than 1 MiB succeeds without negotiation. A transparent [Caddy](https://caddyserver.com/docs/caddyfile/directives/reverse_proxy#streaming)/[nginx](https://nginx.org/en/docs/http/websocket.html) upgrade tunnel is+not evidence of a message-size cap; HTTP request-body limits must not be confused with it.++There is a documented message-terminating intermediary use case: [Cloudflare Workers](https://developers.cloudflare.com/workers/runtime-apis/websockets/)+close WebSockets with code 1009 above **32 MiB** per received message. Their [2025-10-31 changelog](https://developers.cloudflare.com/changelog/post/2025-10-31-increased-websocket-message-size-limit/)+states the earlier 1 MiB limit was raised to 32 MiB. This applies to Workers, not every+Cloudflare-proxied WebSocket, and does not establish how `collab.craftapps.dev` is deployed.++The explicit 65 MiB socket regression models that documented 32 MiB message cap locally:+the legacy welcome fails while the negotiated welcome succeeds in bounded pieces. It does+not contact a live Worker or the public collaboration service. The 1 MiB **chunk payload**+is a conservative packet size, not an asserted intermediary limit. Storage/logical budgets+remain enforced; constant-memory loading and a need on the public service are not claimed.++### Encoding and negotiation++Peers negotiate `figcraft.chunks.v1` through an echoed `X-FigCraft-Transfer` handshake header+(native clients), or the WebSocket subprotocol (browsers cannot set custom headers). The+collaboration core, operation ordering, persistence and protocol version 2 are unchanged. An old+client receives the original binary messages; an old server ignores the native extension header+and the same connection uses legacy messages. No document command is replayed. The browser+retries a failed, unopened extension handshake once+in legacy mode; it never retries a connection that was already open this way.++On a negotiated connection, an encoded message larger than 1 MiB is sent as successive binary+messages. Each starts with byte `2`, the total encoded length (`u64`, little endian), its offset+(`u64`, little endian), and the complete encoded message's SHA-256 (32 bytes), followed by at+most 1 MiB of payload. Messages of 1 MiB or less retain wire version 1. Exactly one transfer may+be in progress in each direction. The receiver requires contiguous offsets, identical metadata,+nonempty bounded pieces, a total within its local receiving budget, and the matching checksum.+It allocates only for payload that has actually arrived, never for a declared length. The+engine sees nothing until the complete encoded operation has been verified. Connection loss+discards the partial message; the existing sequence-number reconnect protocol resumes from the+last complete operation. A partly received welcome restarts from the beginning: byte-offset+resumption and UI progress are not implemented by this extension.++Chunking does **not** bypass a server's `max_file_mb`, cumulative connection budgets or a+client's memory budget. Native clients keep the 1 GiB receiving default; an operator can+explicitly start a native client with `FIGCRAFT_MAX_RECEIVE_MB=2048` for a larger file. Values+outside 1..=4096 MiB or the platform's allocation range fall back to the default. The browser+keeps its 1 GiB limit. Raise the server's configured `max_file_mb` separately, and ensure adequate+memory: this first implementation still assembles and decodes the complete logical message.+It is not lazy image loading or constant-memory document import.++Synthetic regression tests in `transfer_tests.rs` reproduce a legacy welcome exceeding a+WebSocket message budget, then receive the same document under that budget in verified pieces,+upload a large edit as one accepted batch, and check both legacy-peer compatibility directions.+Codec tests reject corrupt, reordered, duplicate, unnegotiated and oversized transfers, and+check that a new connection cannot inherit a partial transfer.+The separate memory-intensive transport regression uses a synthetic 65 MiB document:+`cargo test -p figcraft-collab-net sixty_five_mib_document_crosses_real_socket -- --ignored`.+It checks the transport above 64 MiB, not the persistent server's independent storage quota.+ Status: milestone M9. **Step 1 (the core, no networking) is implemented** in `crates/collab`: identifiers, the object/property model, change sets, ordering and validation, client reconciliation, derived data and the message encoding. **Step 2 is implemented** in@@ -999,3 +1058,19 @@ chooses to share on the network; the default is the invite link with a random to join codes (online through the server, or carrying the host's address on the local network) and by your online files (the app remembers what it owns and opens it again as its owner). 6. **Accounts**: people signing in, per-person permissions, teams and projects.++Receive-budget failures are distinct from invalid framing. A declared total above the local+budget (or an oversized legacy message) keeps the existing local MiB-limit diagnostic in native+and browser joins and management replies, rather than reporting a host refusal. Zero totals,+out-of-order offsets and payloads beyond their declared total remain malformed-frame refusals.+The receiving-budget opt-in described above remains explicit; this does not raise any limit.++Native WebSocket caps and the browser's raw packet check allow the 49-byte transfer header+above the logical receive budget. Reassembly and message decoding still enforce the exact+configured budget, including legacy messages. Socket and browser regressions cover a payload+just above one chunk whose first framed packet exceeds that budget.++Reassembly grows its buffer only when the incoming payload exceeds spare capacity. Exact+reservation requests retain geometric growth while staying within the declared total; they do+not deliberately double a nearly complete transfer beyond its receive budget. This bounds+requested vector capacity, not allocator overhead or decoding memory.@@ -12,7 +12,7 @@ cursors, selections, text carets, cursor chat and spotlights shown live ([`colla | Feature | Figma | FigCraft | Hours | |---|---|---|---|-| Real-time co-editing | yes | **core built, no networking** (#103): client-scoped ids, per-property change sets with fractional positions, server ordering and validation, optimistic clients, binary messages; randomized multi-client convergence tests over real commands. Live sessions over the network landed (`crates/collab-net`): the WebSocket transport (keep-alive, size and rate limits, reconnect with backoff), hosting from the app (this machine or the local network) with invite links, live cursors with name tags, selections, avatars and follow, `collab.*` commands. Document sync between apps works over them (the host's server orders every batch; guests open the host's file as a new tab). Sessions are per document: each open file has its own or none, several run at once (host one, join another, publish a third) and keep syncing behind the tab in front, and closing a file ends its session only. Revert to saved is refused until the document leaves its live session; its access checks and sync association stay intact. Undo is per person (only your own steps, others' later writes left alone), refused edits roll back with a toast, and a reconnect resumes as the same client with what it missed (or a snapshot) and sends what was made offline. The browser build joins (same messages through the browser's WebSocket API); `figcraft://` invite links open the desktop app and join. Styles, variables, collections, modes and comments made in a session take ids from their creator's space (the server checks it), so concurrent creations all survive; modes, per-mode values and replies inside one entry still resolve last-writer-wins. A standalone `figcraft-server` keeps many files, persists them (a snapshot plus a log, compacted), loses nothing confirmed across a restart (apps resume as the same participants) and takes files published from the app (File ▸ Publish to server…). The desktop app and the CLI open `wss://` links too (TLS through a reverse proxy), and releases ship the server for Linux with a deployment guide (systemd, Caddy, nginx, Docker) and `install.sh`. A server can be open to the public: anyone publishes without the key, within limits per address, and files nobody opens expire. Message limits follow the server's `max_file_mb`, so files above 64 MiB publish and open (the client takes at most 1 GiB from a server, and the server warns when large limits meet open publishing, #705); a damaged generation field in a file's log no longer drops the edits flushed to it (#687). Left: server archives beyond Linux | 25–55 |+| Real-time co-editing | yes | **core built, no networking** (#103): client-scoped ids, per-property change sets with fractional positions, server ordering and validation, optimistic clients, binary messages; randomized multi-client convergence tests over real commands. Live sessions over the network landed (`crates/collab-net`): the WebSocket transport (keep-alive, size and rate limits, reconnect with backoff), hosting from the app (this machine or the local network) with invite links, live cursors with name tags, selections, avatars and follow, `collab.*` commands. Document sync between apps works over them (the host's server orders every batch; guests open the host's file as a new tab). Sessions are per document: each open file has its own or none, several run at once (host one, join another, publish a third) and keep syncing behind the tab in front, and closing a file ends its session only. Revert to saved is refused until the document leaves its live session; its access checks and sync association stay intact. Undo is per person (only your own steps, others' later writes left alone), refused edits roll back with a toast, and a reconnect resumes as the same client with what it missed (or a snapshot) and sends what was made offline. The browser build joins (same messages through the browser's WebSocket API); `figcraft://` invite links open the desktop app and join. Styles, variables, collections, modes and comments made in a session take ids from their creator's space (the server checks it), so concurrent creations all survive; modes, per-mode values and replies inside one entry still resolve last-writer-wins. A standalone `figcraft-server` keeps many files, persists them (a snapshot plus a log, compacted), loses nothing confirmed across a restart (apps resume as the same participants) and takes files published from the app (File ▸ Publish to server…). The desktop app and the CLI open `wss://` links too (TLS through a reverse proxy), and releases ship the server for Linux with a deployment guide (systemd, Caddy, nginx, Docker) and `install.sh`. A server can be open to the public: anyone publishes without the key, within limits per address, and files nobody opens expire. Message limits follow the server's `max_file_mb`, so files above 64 MiB publish and open (the client defaults to at most 1 GiB from a server; native clients may explicitly opt into up to 4 GiB, and the server warns when large limits meet open publishing, #705); a damaged generation field in a file's log no longer drops the edits flushed to it (#687). Negotiated chunked transport bounds individual WebSocket messages to 1 MiB plus a header and verifies a complete operation before exposing it, with legacy fallback; the whole operation still resides in memory, and server quotas remain enforced. Synthetic 65 MiB transfer checked; full-size persistent-server validation, progress, byte-offset resume and lazy image loading remain. Left: server archives beyond Linux | 25–55 | | Live cursors, selections, viewports; follow / observation mode; spotlight | yes | **done, in live sessions**: named cursors, selection outlines, avatars, follow (click an avatar), others' text carets and selected characters with their names, a note when two people type in one layer; spotlight (your avatar's menu, `collab.spotlight`): everyone follows your view and page, people who join later too, anyone who moves their view stops following and a banner offers to follow again, a newer spotlight takes over. Left: polish (no accounts, so anyone in the session can spotlight) | 2–4 | | Cursor chat, audio, high-fives | yes | **partial, in live sessions**: cursor chat (`/` opens a bubble at your pointer; everyone sees it as you type, it fades a few seconds after the last change, Enter sends and Escape takes it back, 100 characters; `collab.cursorChat` for agents); sent as ephemeral presence the host checks and rate-limits, never stored. No audio, high-fives or emote reactions | 2–5 | | Share links, permissions (view / edit), teams and projects | yes | **partial**: sharing with a join code from the Share popover (online: the file goes to the server in Preferences and the server names its edit and view links with short codes; local network: the app hosts and the code carries its address; one field joins with a code or a link, in the popover and on the home screen; the browser build also joins from a link in its URL's fragment, `#join=…`, which never reaches the web host's logs, capped at 6 KiB and checked like a `figcraft://` link (#712; unit-tested, not run in a browser)); `figcraft-server` share links per file with edit or view access (a viewer sees changes and presence, a "View only" badge and only the moving tools; the server refuses its edits), revocable at once, tokens stored as hashes; managed from the app's Share popover (first links), the server's command line or a managing connection. On a server open to the public, whoever published a file manages it with its owner token (links, name, removal: `collab.manageFile`), which the app remembers: Home ▸ Online lists your online files (open again as owner, start and stop sharing, copy the code, rename, remove, forget), with one request per server for what became of them; anyone in a file can report it, and the operator blocks addresses. The person hosting a session, or a file's owner, removes someone from it (right-click their avatar, `collab.removePerson`): they are disconnected and their address can't join that file again while it runs. The list is per computer: no accounts, teams or projects | 8–17 |@@ -26,6 +26,9 @@ cursors, selections, text carets, cursor chat and spotlights shown live ([`colla | Date | Change | Summary | |---|---|---|+| 2026-10-22 | minor | Distinguish receive-budget failures from malformed transfer frames: native/browser joins and management replies keep the local MiB-limit diagnostic. Wire format and readiness estimates unchanged. |+| 2026-10-22 | minor | Negotiated bounded chunk transport with checksummed reassembly, legacy fallback and configurable receive budgets; inline images retained. Readiness estimates unchanged. |+| 2026-10-20 | minor | M9: optional chunked WebSocket transport with checksum, bounded reassembly, legacy compatibility and an explicit native receive budget. Synthetic socket tests cover 65 MiB; full-size persistent-server validation, transfer progress, byte-offset resume and lazy assets remain unimplemented. Readiness and estimates unchanged. | | 2026-10-24 | minor | Branches in live sessions (gap 17): creating, editing, reviewing, updating from main and merging a branch work while its file is in a session. The branch stays with its author and makes its ids in their own client space 2^24 above the main file's, so the server takes the layers it adds; merging is one synced edit and one step of the author's undo history; others' edits on main merge against the branch's own base and conflicts resolve as offline; a view link still refuses merging; a branch made outside the session that adds layers, styles, variables or comments is refused with a reason. A branch's history gets entries for its creation and updates from main. No wire format change. Row wording also brought up to date with #798 (export settings and guides merge item by item). Numbers unchanged | | 2026-10-17 | minor | From community batch #735: server message limits follow `max_file_mb`, so files above 64 MiB publish and open, the client takes at most 1 GiB and the server warns about large limits with open publishing (#705); a damaged generation field no longer drops a log's flushed edits (#687); the web build joins a session from a `#join=` link in its URL (#712, unit-tested and built for wasm, not run in a browser). Percentages and hours unchanged | | 2026-10-14 | minor | Removing someone from a live session (M9, from the owner's use: whoever joined with the invite couldn't be put out short of ending the session for everyone): right-click a person's avatar ▸ "Remove <name> from session", or `collab.removePerson {client}`, when you host the session or own the file on a server (owner token). The hub closes their connection as the session's end for them (they keep their copy and don't reconnect) and bars their address from that file while the session or server runs, so the code or link they hold no longer gets them in; everyone else stays. Protocol addition within version 2 (`Admin removePeer`; older servers say they can't). Addresses are the only identity without accounts: people behind one address are barred together. Percentages and hours unchanged |