From fccaa431014d9dc3c4565aae05849f74b79f8d8a Mon Sep 17 00:00:00 2001 From: fluid-chey Date: Wed, 4 Nov 2026 12:17:25 -0700 Subject: [PATCH 1/8] M9: a joined live session keeps reconnecting and says when it's disconnected A session you joined could end for reasons that pass, or after about 85 s of retries, and its tab then looked like an unsaved local file. Work done there reached no one, with only a 3 s toast to say so. A live connection now reconnects until you leave, pausing a minute after each 12 attempts. Only refusals that are for good stop it, so "too many messages", "busy" and the like retry. Messages queued while offline go out paced so a long queue can't trip the server's rate limit. A session that ended stays on its tab: the tab shows a red mark and a strip across the canvas offers Retry now or Try again, Save a copy and Dismiss. Joining again uses the same tab, and what changed offline opens beside it as an unsaved copy. A session that gets out of step joins again by itself. Co-Authored-By: Claude Opus 5.5 --- apps/figcraft-server/src/codes_tests.rs | 2 +- apps/figcraft-server/src/server_tests.rs | 2 +- crates/collab-net/src/driver.rs | 47 +++++++-- crates/collab-net/src/driver_tests.rs | 115 +++++++++++++++++++++-- crates/collab-net/src/guest.rs | 23 ++++- crates/collab-net/src/join.rs | 53 +++++++++-- crates/collab-net/src/link.rs | 6 ++ crates/collab-net/src/session_tests.rs | 5 +- crates/collab-net/src/sessions_tests.rs | 4 +- crates/collab-net/src/throttle.rs | 27 ++++++ crates/collab-net/src/web.rs | 4 + crates/engine/src/collab_live_tests.rs | 111 ++++++++++++++++++++++ crates/engine/src/live_tests.rs | 10 +- crates/engine/src/sharing/live.rs | 47 +++++++++ crates/engine/src/sharing/live/events.rs | 58 ++++++++++-- crates/engine/src/sharing/live/join.rs | 27 +++++- crates/engine/src/sharing/live/sync.rs | 9 +- crates/engine/src/sharing/live/types.rs | 2 + crates/ui-egui/locales/en.toml | 7 ++ crates/ui-egui/src/canvas.rs | 10 +- crates/ui-egui/src/icons.rs | 2 + crates/ui-egui/src/icons/draw.rs | 8 ++ crates/ui-egui/src/live.rs | 64 +++++++++++++ crates/ui-egui/src/live_joining_tests.rs | 49 ++++++++++ crates/ui-egui/src/live_tests.rs | 2 +- crates/ui-egui/src/tabs.rs | 26 +++-- docs/collab-protocol.md | 18 +++- docs/collaboration-parity.md | 1 + docs/mcp.md | 5 +- docs/server.md | 2 +- 30 files changed, 687 insertions(+), 59 deletions(-) diff --git a/apps/figcraft-server/src/codes_tests.rs b/apps/figcraft-server/src/codes_tests.rs index cbc7f70a7..42b15961e 100644 --- a/apps/figcraft-server/src/codes_tests.rs +++ b/apps/figcraft-server/src/codes_tests.rs @@ -118,7 +118,7 @@ fn a_file_shared_online_is_joined_by_its_codes() { }); for guest in [&b, &c] { assert!(guest.message.as_deref().unwrap_or_default().contains("Sharing was stopped"), "{:?}", guest.message); - assert_eq!(guest.live().status, Status::Off, "over, not failed: nothing retries"); + assert!(guest.live().rejoinable() && !guest.live().status.active(), "over: nothing retries, and the tab says so"); } assert_eq!(layers(&b), before + 2, "Bo keeps the copy"); let folder = dir.path().join("files").join(&p.file); diff --git a/apps/figcraft-server/src/server_tests.rs b/apps/figcraft-server/src/server_tests.rs index f60150da9..f3d084499 100644 --- a/apps/figcraft-server/src/server_tests.rs +++ b/apps/figcraft-server/src/server_tests.rs @@ -160,7 +160,7 @@ fn revoked_links_stop_working_at_once() { assert_eq!((files.len(), files[0].people), (1, 3)); // Revoked: Cy is out at once, the others stay. assert_eq!(ask(Admin::Revoke { file: file.clone(), link: review.link.id.clone() }), AdminReply::Done); - pump(&mut [&mut a, &mut b, &mut c], "Cy out", |s| s[2].live().status == Status::Off && s[0].live().peers.len() == 1); + pump(&mut [&mut a, &mut b, &mut c], "Cy out", |s| s[2].live().rejoinable() && s[0].live().peers.len() == 1); // Told as when a host stops: the session is over for Cy, whose copy stays open (no retries). assert!(c.message.as_deref().unwrap_or_default().contains("Sharing was stopped"), "{:?}", c.message); assert!(live(&b)); diff --git a/crates/collab-net/src/driver.rs b/crates/collab-net/src/driver.rs index 3ddd3b3c1..d96650371 100644 --- a/crates/collab-net/src/driver.rs +++ b/crates/collab-net/src/driver.rs @@ -18,6 +18,7 @@ use figcraft_collab::message::{self, ClientMsg, Presence, ServerMsg}; use figcraft_engine::live::{Event, Status}; +use std::collections::VecDeque; use std::time::Duration; // `std::time::Instant` on native builds; the browser's clock in the web build. use web_time::Instant; @@ -28,7 +29,7 @@ use crate::invite::Endpoint; use crate::join::{GuestConfig, Next, Outcome, Resuming, Seen, Taken, after, ask, closed, hello, resolved, take, too_large, unresolved}; use crate::limits::NetLimits; use crate::requests::Publish; -use crate::throttle::{PRESENCE_INTERVAL, Throttle, earliest}; +use crate::throttle::{PRESENCE_INTERVAL, Pace, Throttle, earliest}; /// A WebSocket the driver opens, writes and closes; what it receives comes back through the /// driver's event methods. @@ -79,6 +80,9 @@ pub(crate) struct Driver { /// Your latest presence while not welcomed (sent with the welcome). pending: Option, throttle: Throttle, + /// Sync messages waiting their turn (a long queue from while offline), and when each may go. + pace: Pace, + paced: VecDeque<(Instant, ClientMsg)>, reconnect_at: Option, /// What this connection asks a server (a managing one), and what follows its answers. asking: Option<(Box, Then)>, @@ -86,7 +90,7 @@ pub(crate) struct Driver { } impl Driver { - fn new(sock: S, cfg: GuestConfig, inbox: Inbox) -> Self { + fn new(sock: S, cfg: GuestConfig, inbox: Inbox, now: Instant) -> Self { Self { sock, cfg, @@ -97,6 +101,8 @@ impl Driver { resume: Resuming::default(), pending: None, throttle: Throttle::new(PRESENCE_INTERVAL), + pace: Pace::new(now), + paced: VecDeque::new(), reconnect_at: None, asking: None, done: false, @@ -105,7 +111,7 @@ impl Driver { /// Start joining (progress arrives in `inbox`). pub(crate) fn start(sock: S, cfg: GuestConfig, inbox: Inbox, now: Instant) -> Self { - let mut d = Self::new(sock, cfg, inbox); + let mut d = Self::new(sock, cfg, inbox, now); d.inbox.push(Event::Status(Status::Connecting)); d.connect(now); d @@ -114,7 +120,7 @@ impl Driver { /// Start with what `next` asks a server (a background request): the events it ends in go to /// `inbox`, then the join it ends in, if any, as `cfg` says. pub(crate) fn asking(sock: S, cfg: GuestConfig, next: exchange::Next, inbox: Inbox, now: Instant) -> Self { - let mut d = Self::new(sock, cfg, inbox); + let mut d = Self::new(sock, cfg, inbox, now); d.follow(next, now); d } @@ -303,8 +309,31 @@ impl Driver { if !self.conn.as_ref().is_some_and(|c| c.welcomed) { return; } - if !self.write(&msg) { - self.end(Outcome::Retry("the connection dropped".into()), now); + let wait = self.pace.wait(now); + if wait.is_zero() && self.paced.is_empty() { + if !self.write(&msg) { + self.end(Outcome::Retry("the connection dropped".into()), now); + } + } else { + self.paced.push_back((now + wait, msg)); + } + } + + /// Send the sync messages whose turn has come; false when one couldn't go. + fn send_paced(&mut self, now: Instant) -> bool { + while let Some((_, msg)) = self.paced.pop_front_if(|(at, _)| *at <= now) { + if !self.write(&msg) { + return false; + } + } + true + } + + /// Reconnect now, if it is waiting to: the pauses grow from the beginning again. + pub(crate) fn retry(&mut self, now: Instant) { + if !self.done && self.reconnect_at.is_some() { + self.attempt = 0; + self.connect(now); } } @@ -327,7 +356,7 @@ impl Driver { [Some(c.last_ping + l.keepalive), Some(c.last_heard + c.silence_limit(l)), if c.welcomed { self.throttle.due() } else { None }] } }); - earliest(conn.into_iter().flatten().chain([self.reconnect_at])) + earliest(conn.into_iter().flatten().chain([self.reconnect_at, self.paced.front().map(|(at, _)| *at)])) } /// Do what's due at `now`: reconnect, send held presence or a ping, give up on a connection @@ -357,7 +386,7 @@ impl Driver { if now >= last_heard + silence { return self.end(Outcome::Retry("the host stopped answering".into()), now); } - let ok = self.throttle.take_due(now).is_none_or(|p| self.write(&ClientMsg::Presence(p))); + let ok = self.send_paced(now) && self.throttle.take_due(now).is_none_or(|p| self.write(&ClientMsg::Presence(p))); let ok = ok && (now < last_ping + l.keepalive || self.ping(now)); if !ok { self.end(Outcome::Retry("the connection dropped".into()), now); @@ -388,6 +417,8 @@ impl Driver { self.sock.close(); } self.throttle = Throttle::new(PRESENCE_INTERVAL); + // Made for this connection: the sync client sends again what the server didn't get. + self.paced.clear(); match after(outcome, &self.cfg, self.was_live, &mut self.attempt, &self.inbox) { Next::Stop(e) => { if let Some(e) = e { diff --git a/crates/collab-net/src/driver_tests.rs b/crates/collab-net/src/driver_tests.rs index ea7900745..57921df3d 100644 --- a/crates/collab-net/src/driver_tests.rs +++ b/crates/collab-net/src/driver_tests.rs @@ -252,7 +252,7 @@ fn a_silent_host_counts_as_dropped_and_the_client_reconnects() { } #[test] -fn reconnects_back_off_then_give_up() { +fn a_live_connection_reconnects_in_rounds_and_never_gives_up() { let t0 = Instant::now(); let (mut d, inbox) = start(Script::default(), t0); d.opened(t0); @@ -260,7 +260,7 @@ fn reconnects_back_off_then_give_up() { let mut now = t0; let mut pauses = Vec::new(); d.closed(1006, "", now); - while !d.done() { + for _ in 0..26 { let next = d.deadline().unwrap(); pauses.push(next - now); now = next; @@ -269,11 +269,114 @@ fn reconnects_back_off_then_give_up() { d.closed(1006, "", now); } let secs: Vec = pauses.iter().map(Duration::as_secs_f64).collect(); - assert_eq!(secs, [0.5, 1.0, 2.0, 4.0, 8.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0]); + let round = [1.0, 2.0, 4.0, 8.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0]; + assert_eq!(secs[0], 0.5); + assert_eq!(secs[1..12], round); + assert_eq!(secs[12], 60.0, "a longer pause after a round"); + assert_eq!(secs[13..24], round, "then as many again"); + assert_eq!(secs[24], 60.0); + assert!(!d.done()); let st = statuses(&inbox.drain()); - assert_eq!(st.len(), 14, "connecting, 12 reconnects, then failed"); - assert!(matches!(st.last(), Some(Status::Failed { reason }) if reason.starts_with("Lost the connection to the host")), "{st:?}"); - assert_eq!(d.deadline(), None); + assert!(!st.iter().any(|s| matches!(s, Status::Failed { .. })), "{st:?}"); + assert!(matches!(st.last(), Some(Status::Reconnecting { attempt: 27 })), "{st:?}"); +} + +#[test] +fn retry_now_skips_the_pause_and_starts_the_round_over() { + let t0 = Instant::now(); + let (mut d, inbox) = start(Script::default(), t0); + d.opened(t0); + d.message(&welcome(), t0); + d.closed(1006, "", t0); + let mut now = t0; + for _ in 0..12 { + now = d.deadline().unwrap(); + d.tick(now); + d.closed(1006, "", now); + } + assert_eq!(d.deadline().unwrap() - now, Duration::from_secs(60)); + let opened = d.sock.opened.len(); + d.retry(now); + assert_eq!(d.sock.opened.len(), opened + 1, "connects at once"); + inbox.drain(); + d.closed(1006, "", now); + assert_eq!(statuses(&inbox.drain()), vec![Status::Reconnecting { attempt: 1 }]); + assert_eq!(d.deadline().unwrap() - now, ms(500), "the pauses grow from the beginning again"); +} + +#[test] +fn a_session_that_was_live_only_stops_for_a_refusal_that_is_for_good() { + let t0 = Instant::now(); + // The words servers turn a session away with for now. + for reason in [ + "too many messages", + "the server is busy: try again in a moment", + "the session is full", + "no hello in time", + "this file can't be opened on the server (its data is damaged); tell the server's owner", + "too many connections from your address lately: wait a minute and try again", + ] { + for frame in [false, true] { + let (mut d, inbox) = start(Script::default(), t0); + d.opened(t0); + d.message(&welcome(), t0); + inbox.drain(); + if frame { + d.closed(1008, reason, t0); + } else { + d.message(&encode(&ServerMsg::Refused { reason: reason.into() }), t0); + } + assert!(!d.done(), "{reason}"); + assert_eq!(statuses(&inbox.drain()), vec![Status::Reconnecting { attempt: 1 }], "{reason}"); + } + } + // And the words for good. + for reason in [ + "this link isn't valid (it may have been revoked)", + "this file isn't on the server (it may have been removed)", + "this file was removed from the server", + "you were removed from this session", + "this server no longer accepts connections from your address", + "this server doesn't accept connections from your address", + "wrong session token", + "protocol 9 isn't supported", + "message too large: this server takes at most 64 MiB at once (a change that big can't be kept here)", + ] { + let (mut d, inbox) = start(Script::default(), t0); + d.opened(t0); + d.message(&welcome(), t0); + d.closed(1008, reason, t0); + assert!(d.done(), "{reason}"); + assert!(matches!(statuses(&inbox.drain()).last(), Some(Status::Failed { reason: r }) if r.contains(reason)), "{reason}"); + } +} + +#[test] +fn a_long_queue_goes_out_paced_and_in_order() { + let t0 = Instant::now(); + let (mut d, _inbox) = start(Script::default(), t0); + d.opened(t0); + d.message(&welcome(), t0); + let before = d.sock.sent.len(); + for seq in 0..500 { + d.send(ClientMsg::Ack { seq }, t0); + } + let at_once = d.sock.sent.len() - before; + assert!((1..=200).contains(&at_once), "{at_once} at once: a server drops a connection past 400"); + let mut now = t0; + while d.sock.sent.iter().filter(|m| matches!(m, ClientMsg::Ack { .. })).count() < 500 { + now = d.deadline().unwrap(); + d.tick(now); + } + let seqs: Vec = d.sock.sent.iter().filter_map(|m| if let ClientMsg::Ack { seq } = m { Some(*seq) } else { None }).collect(); + assert_eq!(seqs, (0..500).collect::>()); + assert!(now - t0 >= Duration::from_secs(4), "the rest take their time: {:?}", now - t0); + // Unsent ones were made for this connection: the sync client sends them again. + for seq in 0..500 { + d.send(ClientMsg::Ack { seq }, now); + } + d.closed(1006, "", now); + assert_eq!(d.deadline().unwrap() - now, ms(500), "nothing left to send but the reconnect"); } #[test] diff --git a/crates/collab-net/src/guest.rs b/crates/collab-net/src/guest.rs index bdb5b98b2..84572b884 100644 --- a/crates/collab-net/src/guest.rs +++ b/crates/collab-net/src/guest.rs @@ -31,7 +31,7 @@ use crate::manage; use crate::net::{Duplex, NetLimits, config, open, shutdown}; use crate::requests; pub use crate::requests::Publish; -use crate::throttle::{PRESENCE_INTERVAL, Throttle, earliest}; +use crate::throttle::{PRESENCE_INTERVAL, Pace, Throttle, earliest}; /// A joined (or joining) session. Dropping it leaves. pub struct Guest { @@ -44,6 +44,8 @@ enum Cmd { /// For the server, in order (document changes). Send(ClientMsg), Leave, + /// Try now rather than after the pause under way (reconnecting). + Retry, /// The reader of connection `epoch` got its welcome. Welcomed(u64), /// Connection `epoch` ended. @@ -133,6 +135,11 @@ impl Guest { let _ = self.cmd.send(Cmd::Send(msg)); } + /// Reconnect now, if it is waiting to. + pub fn retry(&self) { + let _ = self.cmd.send(Cmd::Retry); + } + /// Leave (returns at once: a connection attempt in progress ends on its own). pub fn leave(&mut self) { let _ = self.cmd.send(Cmd::Leave); @@ -227,6 +234,11 @@ fn supervise(cfg: &GuestConfig, inbox: &Inbox, rx: &Receiver, me: &Sender return, Ok(Cmd::Presence(p)) => pending = Some(p), + // A fresh start: the pauses grow from the beginning again. + Ok(Cmd::Retry) => { + attempt = 0; + break; + } // From the old connection's client: the next welcome starts over. Ok(_) => {} Err(RecvTimeoutError::Timeout) => break, @@ -267,6 +279,7 @@ fn session( let hello = hello(cfg, resume); let mut welcomed = false; let mut throttle = Throttle::new(PRESENCE_INTERVAL); + let mut pace = Pace::new(Instant::now()); let outcome = if send(&mut writer, &hello) { loop { let ping_at = last_ping + cfg.limits.keepalive; @@ -274,7 +287,11 @@ fn session( let mut out = None; let mut now_send = Vec::new(); match rx.recv_timeout(wake.saturating_duration_since(Instant::now())) { - Ok(Cmd::Send(m)) if welcomed => now_send.push(m), + Ok(Cmd::Send(m)) if welcomed => { + // A long queue from while offline doesn't go out faster than the server takes it. + std::thread::sleep(pace.wait(Instant::now())); + now_send.push(m); + } // Made for the previous connection (the sync client sends again what the server // didn't get once it has caught up with this connection's welcome). Ok(Cmd::Send(_)) => {} @@ -292,7 +309,7 @@ fn session( break Outcome::Left; } // From an earlier connection. - Ok(Cmd::Welcomed(_) | Cmd::Lost(..)) => {} + Ok(Cmd::Welcomed(_) | Cmd::Lost(..) | Cmd::Retry) => {} Err(RecvTimeoutError::Timeout) => {} } if !now_send.iter().all(|m| send(&mut writer, m)) { diff --git a/crates/collab-net/src/join.rs b/crates/collab-net/src/join.rs index 033186af2..63f94bec2 100644 --- a/crates/collab-net/src/join.rs +++ b/crates/collab-net/src/join.rs @@ -39,8 +39,10 @@ pub struct GuestConfig { /// The pause before the first reconnect; it doubles each time up to `max_backoff`. pub backoff: Duration, pub max_backoff: Duration, - /// Reconnect attempts before giving up. + /// Reconnect attempts in a row before the longer pause (`round_pause`), after which it tries + /// as many again: a connection that was live is never given up on by itself. pub attempts: u32, + pub round_pause: Duration, /// Join as the client you were (an app that starts over after a crash): the first hello /// carries it, and a server that still knows the id gives it back. pub resume: Option, @@ -57,6 +59,7 @@ impl GuestConfig { backoff: Duration::from_millis(500), max_backoff: Duration::from_secs(10), attempts: 12, + round_pause: Duration::from_secs(60), resume: None, } } @@ -65,6 +68,14 @@ impl GuestConfig { pub(crate) fn backoff(&self, attempt: u32) -> Duration { self.backoff.saturating_mul(1u32 << attempt.saturating_sub(1).min(16)).min(self.max_backoff) } + + /// The pause before reconnect attempt `attempt` (from 1): growing within a round of + /// [`attempts`](Self::attempts), and [`round_pause`](Self::round_pause) between rounds. + pub(crate) fn pause(&self, attempt: u32) -> Duration { + let round = self.attempts.max(1); + let into_round = attempt.saturating_sub(1) % round; + if into_round == 0 && attempt > round { self.round_pause } else { self.backoff(into_round + 1) } + } } /// How a connection ended. @@ -143,11 +154,38 @@ pub(crate) enum Next { Reconnect(Duration), } -/// After `outcome`: a refusal or the end of the session is final, and so is a first join that -/// can't reach the host; a live connection that dropped tries again with growing pauses until the -/// attempts run out. +/// What a server refuses for good, whenever you ask: the link, the file or your place in it is +/// gone, or the two versions can't talk. Anything else a server refuses a session with (it is +/// busy, a limit was hit, the file couldn't be read just now) may pass, and servers of any version +/// say so in these words, so a reason not listed here means try again. +const FINAL_REFUSALS: [&str; 11] = [ + "isn't valid", + "isn't on the server", + "was removed from the server", + "were removed from this session", + "no longer accepts connections", + "doesn't accept connections", + "wrong session token", + "isn't supported", + "couldn't be read", + "message too large", + "this version can't read", +]; + +/// Whether `reason`, from a server refusing a session that was live, is for good. +pub(crate) fn final_refusal(reason: &str) -> bool { + FINAL_REFUSALS.iter().any(|f| reason.contains(f)) +} + +/// After `outcome`: the end of the session is final, and so are a refusal for good and a first +/// join that can't reach the host. A live connection that dropped or was turned away for now tries +/// again until you leave: with growing pauses, and a longer one between rounds. pub(crate) fn after(outcome: Outcome, cfg: &GuestConfig, was_live: bool, attempt: &mut u32, inbox: &Inbox) -> Next { let failed = |reason: String| Next::Stop(Some(Event::Status(Status::Failed { reason }))); + let outcome = match outcome { + Outcome::Refused(why) if was_live && !final_refusal(&why) => Outcome::Retry(why), + o => o, + }; match outcome { Outcome::Left => Next::Stop(None), Outcome::Refused(why) => failed(format!("The host refused: {why}")), @@ -155,13 +193,10 @@ pub(crate) fn after(outcome: Outcome, cfg: &GuestConfig, was_live: bool, attempt Outcome::Ended(reason) => Next::Stop(Some(Event::Ended { reason })), Outcome::Retry(why) if !was_live => failed(format!("Couldn't join the live session: {why}")), Outcome::Retry(why) => { - *attempt += 1; - if *attempt > cfg.attempts { - return failed(format!("Lost the connection to the host ({why})")); - } + *attempt = attempt.saturating_add(1); log::info!("live session: connection lost ({why}); reconnecting, attempt {attempt}"); inbox.push(Event::Status(Status::Reconnecting { attempt: *attempt })); - Next::Reconnect(cfg.backoff(*attempt)) + Next::Reconnect(cfg.pause(*attempt)) } } } diff --git a/crates/collab-net/src/link.rs b/crates/collab-net/src/link.rs index fdcba5b97..0024da781 100644 --- a/crates/collab-net/src/link.rs +++ b/crates/collab-net/src/link.rs @@ -266,6 +266,12 @@ impl LiveLink for NetLink { } } + fn retry(&mut self, session: SessionId) { + if let Some(Active::Guest(g)) = self.sessions.get(session) { + g.retry(); + } + } + fn presence(&mut self, session: SessionId, p: &Presence) { match self.sessions.get(session) { Some(Active::Host(h)) => h.presence(p.clone()), diff --git a/crates/collab-net/src/session_tests.rs b/crates/collab-net/src/session_tests.rs index 4d5680eed..51b9e89b6 100644 --- a/crates/collab-net/src/session_tests.rs +++ b/crates/collab-net/src/session_tests.rs @@ -124,8 +124,9 @@ fn following_a_viewport_until_you_move() { fn the_host_stopping_ends_the_session_for_guests() { let (mut host, mut guest, _, _) = host_and_join(); host.execute("collab.stop", &json!({})).unwrap(); - pump(&mut host, &mut guest, "the end", |_, g| g.live().status == Status::Off); - assert!(guest.live().peers.is_empty()); + pump(&mut host, &mut guest, "the end", |_, g| g.live().rejoinable()); + assert!(guest.live().peers.is_empty() && !guest.live().syncing()); + assert!(guest.live().disconnected(), "the tab says so"); assert_eq!(guest.message.as_deref(), Some(crate::host::ENDED)); } diff --git a/crates/collab-net/src/sessions_tests.rs b/crates/collab-net/src/sessions_tests.rs index 4a2579f8e..d21668150 100644 --- a/crates/collab-net/src/sessions_tests.rs +++ b/crates/collab-net/src/sessions_tests.rs @@ -158,8 +158,8 @@ fn two_files_are_hosted_and_joined_independently() { assert!(host.live_of(hb).presence.get(&ada(&host, hb)).is_none_or(|p| p.cursor.is_none()), "no cursor in the file that isn't in front"); // The host closes B: its session ends for everyone, A's goes on. host.execute("file.close", &json!({"index": 1, "force": true})).unwrap(); - pump(&mut [&mut host, &mut guest], "B's session ending", |s| s[0].live.iter().count() == 1 && s[1].live.iter().count() == 1); - assert_eq!(guest.live_of(gb).status, Status::Off); + pump(&mut [&mut host, &mut guest], "B's session ending", |s| s[0].live.iter().count() == 1 && s[1].live_of(gb).rejoinable()); + assert!(guest.live_of(gb).disconnected(), "the guest's tab says so"); assert_eq!(guest.docs.len(), 3, "the guest keeps its copy of B"); assert_eq!((host.live_of(ha).status.clone(), host.live_of(ha).peers.len()), (Status::Live, 1)); assert_eq!(guest.live_of(ga).status, Status::Live); diff --git a/crates/collab-net/src/throttle.rs b/crates/collab-net/src/throttle.rs index beff3e6f8..1f90ff504 100644 --- a/crates/collab-net/src/throttle.rs +++ b/crates/collab-net/src/throttle.rs @@ -10,6 +10,33 @@ use web_time::Instant; /// How often presence goes out at most. pub const PRESENCE_INTERVAL: Duration = Duration::from_millis(33); +/// How many messages other than presence go out at once, and per second after that: with presence +/// and pings that stays under what a server takes from a connection before it drops it. +const PACE_BURST: f64 = 200.0; +const PACE_RATE: f64 = 60.0; + +/// Paces messages that must all go out (everything queued while offline leaves at once when the +/// connection comes back): `wait` says how long until the next may. +#[derive(Debug)] +pub struct Pace { + tokens: f64, + last: Instant, +} + +impl Pace { + pub fn new(now: Instant) -> Self { + Self { tokens: PACE_BURST, last: now } + } + + /// Take one message's turn at `now`: how long to wait before sending it (zero: at once). + pub fn wait(&mut self, now: Instant) -> Duration { + let passed = now.saturating_duration_since(self.last).as_secs_f64(); + self.last = now; + self.tokens = (self.tokens + passed * PACE_RATE).min(PACE_BURST) - 1.0; + if self.tokens >= 0.0 { Duration::ZERO } else { Duration::from_secs_f64(-self.tokens / PACE_RATE) } + } +} + #[derive(Debug)] pub struct Throttle { interval: Duration, diff --git a/crates/collab-net/src/web.rs b/crates/collab-net/src/web.rs index 32fbaad9d..8c54f6c23 100644 --- a/crates/collab-net/src/web.rs +++ b/crates/collab-net/src/web.rs @@ -242,6 +242,10 @@ impl LiveLink for NetLink { } } + fn retry(&mut self, session: SessionId) { + self.with(session, |d, now| d.retry(now)); + } + fn presence(&mut self, session: SessionId, p: &Presence) { self.with(session, |d, now| d.presence(p.clone(), now)); } diff --git a/crates/engine/src/collab_live_tests.rs b/crates/engine/src/collab_live_tests.rs index d53eeff26..6499c1895 100644 --- a/crates/engine/src/collab_live_tests.rs +++ b/crates/engine/src/collab_live_tests.rs @@ -318,6 +318,117 @@ fn a_dropped_connection_keeps_editing_and_catches_up() { assert_eq!((doc.node(r).unwrap().name.as_str(), doc.node(f).unwrap().name.as_str()), ("Sent", "Offline")); } +impl Room { + /// Person `i` has been told to join again: the server takes them as someone new. + fn welcome_anew(&mut self, i: usize) { + let ClientMsg::Hello(h) = Client::hello("f", "t", "Person", "") else { panic!("not a hello") }; + let (id, out) = self.server.join(h).unwrap(); + let w = welcome_of(&out, id); + let p = &mut self.people[i]; + p.id = id; + p.online = true; + p.script.lock().unwrap().events.extend([Event::Welcome { info: info(Role::Guest, id), peers: vec![] }, Event::Sync(w)]); + p.s.live_poll(); + } + + fn event(&mut self, i: usize, e: Event) { + let p = &mut self.people[i]; + p.script.lock().unwrap().events.push(e); + p.s.live_poll(); + } +} + +#[test] +fn reconnecting_can_be_told_to_try_now_and_shows_as_disconnected() { + let mut room = Room::new(base(), 1, 7); + assert!(!room.s(0).live().disconnected()); + assert!(room.s(0).execute("collab.reconnect", &json!({})).is_err(), "nothing was lost"); + room.disconnect(0); + assert!(room.s(0).live().disconnected()); + assert_eq!(room.s(0).execute("collab.reconnect", &json!({})).unwrap()["status"], "reconnecting"); + let session = room.s(0).live().id; + assert_eq!(room.people[0].script.lock().unwrap().retried, [session]); +} + +#[test] +fn a_session_that_cant_go_on_stays_on_its_tab_and_joins_again_there() { + let mut room = Room::new(base(), 2, 7); + let (f, r) = frame_and_rect(room.server.document()); + let uid = room.s(0).ds().unwrap().uid; + let tabs = room.s(0).docs.len(); + room.disconnect(0); + room.s(0).execute("node.set", &json!({"id": r, "props": {"name": "Offline"}})).unwrap(); + room.event(0, Event::Status(Status::Failed { reason: "The host refused: this link isn't valid".into() })); + // The tab says so, and is yours to go on working in. + let live = room.s(0).live(); + assert!(live.disconnected() && live.rejoinable() && live.doc_uid == Some(uid)); + assert!(room.s(0).live_read_only(uid).is_none()); + room.s(0).execute("node.set", &json!({"id": f, "props": {"name": "Offline too"}})).unwrap(); + // A try that fails leaves it just as it was. + assert_eq!(room.s(0).execute("collab.reconnect", &json!({})).unwrap()["status"], "connecting"); + assert!(room.s(0).live().disconnected() && !room.s(0).live().rejoinable()); + assert!(room.s(0).live_read_only(uid).is_none(), "it can still be edited while it tries"); + room.event(0, Event::Status(Status::Failed { reason: "Couldn't join the live session: no answer".into() })); + assert!(room.s(0).live().rejoinable()); + // Someone else's edit lands meanwhile. + room.s(1).execute("transform.move", &json!({"ids": [r], "dx": 5, "dy": 5})).unwrap(); + room.settle(); + // The next try joins in the same tab, and what was done offline opens beside it. + room.s(0).execute("collab.reconnect", &json!({})).unwrap(); + room.welcome_anew(0); + let s = room.s(0); + assert_eq!(s.live().status, Status::Live); + assert!(!s.live().disconnected()); + assert_eq!(s.docs.len(), tabs + 1, "the copy's is the only new tab"); + let (tab, copy) = (s.docs.iter().find(|d| d.uid == uid).unwrap(), s.docs.last().unwrap()); + assert_eq!(s.live().doc_uid, Some(uid)); + assert_ne!(tab.doc.node(r).unwrap().name, "Offline", "the session's tab shows the server's version"); + assert_eq!(copy.doc.name, format!("{} (offline copy)", tab.doc.name)); + assert_eq!(copy.doc.node(r).unwrap().name, "Offline"); + assert_eq!(copy.doc.node(f).unwrap().name, "Offline too"); + assert!(copy.is_dirty(), "closing the copy asks first"); + assert!(s.message.as_deref().is_some_and(|m| m.contains("offline copy"))); +} + +#[test] +fn joining_again_with_nothing_changed_opens_no_copy() { + let mut room = Room::new(base(), 1, 7); + let tabs = room.s(0).docs.len(); + room.disconnect(0); + room.event(0, Event::Ended { reason: "Sharing was stopped".into() }); + assert!(room.s(0).live().rejoinable(), "the end of a session you were in shows on its tab too"); + assert_eq!(room.s(0).message.as_deref(), Some("Sharing was stopped")); + room.s(0).execute("collab.reconnect", &json!({})).unwrap(); + room.welcome_anew(0); + assert_eq!(room.s(0).docs.len(), tabs); + assert_eq!(room.s(0).live().status, Status::Live); +} + +#[test] +fn a_session_out_of_step_joins_again_by_itself() { + let mut room = Room::new(base(), 1, 7); + let joins = room.people[0].script.lock().unwrap().joined.len(); + let tabs = room.s(0).docs.len(); + // A batch from far in the future: it can't be applied. + room.event(0, Event::Sync(ServerMsg::Changes(figcraft_collab::Accepted { seq: 999, author: 42, batch_id: 1, changes: vec![] }))); + assert_eq!(room.s(0).live().status, Status::Connecting); + assert!(room.s(0).live().disconnected()); + assert_eq!(room.people[0].script.lock().unwrap().joined.len(), joins + 1); + room.welcome_anew(0); + assert_eq!(room.s(0).live().status, Status::Live); + assert_eq!(room.s(0).docs.len(), tabs); +} + +#[test] +fn leaving_a_lost_session_makes_the_tab_a_plain_copy() { + let mut room = Room::new(base(), 1, 7); + room.disconnect(0); + room.event(0, Event::Status(Status::Failed { reason: "gone".into() })); + room.s(0).execute("collab.leave", &json!({})).unwrap(); + assert_eq!(room.s(0).live().status, Status::Off); + assert!(!room.s(0).live().disconnected()); +} + #[test] fn random_commands_with_undo_and_drops_converge_through_the_engine() { for seed in 0..6 { diff --git a/crates/engine/src/live_tests.rs b/crates/engine/src/live_tests.rs index 1bae1b106..0b2b7c993 100644 --- a/crates/engine/src/live_tests.rs +++ b/crates/engine/src/live_tests.rs @@ -17,7 +17,9 @@ pub(crate) struct Script { pub(crate) events: Vec, pub(crate) sent: Vec, pub(crate) hosted: Vec, - joined: Vec, + pub(crate) joined: Vec, + /// The sessions told to reconnect now, in order. + pub(crate) retried: Vec, pub(crate) left: usize, pub(crate) msgs: Vec, pub(crate) published: Vec, @@ -46,6 +48,10 @@ impl LiveLink for Mock { i.url = format!("ws://{}/?file=f&token=t", i.address); Ok(i) } + fn retry(&mut self, session: u64) { + self.0.lock().unwrap().retried.push(session); + } + fn join(&mut self, session: u64, link: &str, _: &str) -> Result<(), String> { if !link.starts_with("figcraft://") { return Err("not an invite".into()); @@ -229,7 +235,7 @@ fn events_build_who_is_here() { assert_eq!(s.live_summary()["attempt"], 1); push(&script, vec![Event::Status(Status::Failed { reason: "Lost the connection".into() })]); s.live_poll(); - assert!(s.live().info.is_none()); + assert!(s.live().peers.is_empty() && !s.live().syncing()); assert_eq!(s.live_summary()["reason"], "Lost the connection"); assert_eq!(s.message.as_deref(), Some("Lost the connection")); assert!(!s.live_poll(), "nothing new"); diff --git a/crates/engine/src/sharing/live.rs b/crates/engine/src/sharing/live.rs index 45d56776c..b54331520 100644 --- a/crates/engine/src/sharing/live.rs +++ b/crates/engine/src/sharing/live.rs @@ -140,6 +140,18 @@ pub struct Live { /// The branches of the session's file made or last updated from main in this session: every /// layer of their base was in it (see [`crate::branches`]). branch_bases: BTreeSet, + /// You were in this session and aren't now: what joining it again takes (`collab.reconnect`). + stranded: Option, +} + +/// What's kept of a session you joined once it can't go on by itself. The tab stays yours to work +/// in, and says so. +#[derive(Clone)] +struct Stranded { + /// The invite link you joined with (it holds the token: never shown). + link: String, + /// The server's state when it stopped: what's different from it in the tab is your own work. + confirmed: Box, } impl Live { @@ -158,6 +170,41 @@ impl Live { *self = Live { id: self.id, tab: self.tab, ..Live::default() }; } + /// Keep this session on its tab for joining again (see [`Stranded`]): a session you joined, or + /// one that already is; any other goes back to no session. + fn strand(&mut self) { + let joined = self.info.as_ref().filter(|i| i.role == Role::Guest && !i.link.is_empty()); + let held = self.stranded.take().or_else(|| { + let (client, info) = (self.client.as_ref()?, joined?); + Some(Stranded { link: info.link.clone(), confirmed: Box::new(client.confirmed().clone()) }) + }); + if held.is_none() { + return self.reset(); + } + *self = Live { + id: self.id, + tab: self.tab, + doc_uid: self.doc_uid, + info: self.info.take(), + server: std::mem::take(&mut self.server), + published: self.published.take(), + share: self.share.take(), + stranded: held, + ..Live::default() + }; + } + + /// You were in this session and aren't now: the connection dropped and it is trying again, or + /// it can't go on by itself. What you do in its tab isn't reaching anyone. + pub fn disconnected(&self) -> bool { + matches!(self.status, Status::Reconnecting { .. }) || self.stranded.is_some() + } + + /// The session can be joined again in its tab (`collab.reconnect`). + pub fn rejoinable(&self) -> bool { + self.stranded.is_some() && !self.status.active() + } + /// The document sync client (once the welcome arrived): peers, pending batches, the user's /// undo history. pub fn client(&self) -> Option<&Client> { diff --git a/crates/engine/src/sharing/live/events.rs b/crates/engine/src/sharing/live/events.rs index 757decbc5..5bc94c0f4 100644 --- a/crates/engine/src/sharing/live/events.rs +++ b/crates/engine/src/sharing/live/events.rs @@ -214,8 +214,15 @@ impl Session { // order. Event::Sync(msg) => live.deferred.push(msg), Event::Ended { reason } => { - self.message = Some(if behind.is_empty() { reason } else { format!("{}: {reason}", behind.trim_start()) }); - self.live.end(id); + let joined = live.syncing() && live.role() == Some(Role::Guest); + let text = if behind.is_empty() { reason.clone() } else { format!("{}: {reason}", behind.trim_start()) }; + if joined { + // The tab says so and stays yours: it doesn't pass for a file of your own. + self.live_fail(id, reason); + } else { + self.live.end(id); + } + self.message = Some(text); } Event::Published(p) => { // Shared with a code: the popover shows it, nothing more to say. @@ -389,6 +396,7 @@ impl Session { } (Some(Role::Guest), Some(uid)) => { // Joining again in this tab, or the file you published arriving from the server. + self.live_keep_offline_work(id, uid); let doc = Arc::new(client.document().clone()); let was_dirty = self.docs.iter().find(|d| d.uid == uid).is_some_and(DocState::is_dirty); self.apply_remote(uid, doc.clone()); @@ -435,15 +443,53 @@ impl Session { live.synced = synced; live.client = Some(client); live.deferred.clear(); + live.stranded = None; } - /// Session `id` is over with a message: it couldn't be joined, the connection was lost for - /// good or sync broke. Its tab shows why. + /// Session `id` is back on document `uid` after it couldn't go on, and the server's version is + /// about to replace the tab's: what you did since opens beside it, unsaved, so nothing is lost. + fn live_keep_offline_work(&mut self, id: SessionId, uid: u64) { + let Some(held) = self.live.sessions.get(&id).and_then(|l| l.stranded.as_ref()) else { return }; + let Some(local) = self.docs.iter().find(|d| d.uid == uid).map(|d| d.doc.clone()) else { return }; + let name = local.name.clone(); + // A document that can't be compared is kept too. + if figcraft_collab::diff(&held.confirmed, &local).is_ok_and(|d| d.changes.is_empty()) { + self.message = Some(format!("Reconnected to {name}")); + return; + } + let mut copy = DocState::new((*local).clone(), None); + let copy_name = format!("{name} (offline copy)"); + Arc::make_mut(&mut copy.doc).name.clone_from(©_name); + self.docs.push(copy); + self.message = Some(format!("Reconnected to {name}. What you changed while disconnected is open beside it as “{copy_name}”")); + } + + /// Join session `id` again in its tab (see [`Live::strand`]), as someone new: a fresh join needs + /// nothing from the connection that was lost. + pub(super) fn live_rejoin(&mut self, cmd: &str, id: SessionId) -> Result<()> { + let name = display_name(&self.prefs); + let Some(link) = self.live.sessions.get(&id).and_then(|l| l.stranded.as_ref()).map(|s| s.link.clone()) else { + return Err(crate::bad(cmd, "this file has no live session to join again")); + }; + self.live.leave(id); + if let Some(why) = crate::file_access::network_refused() { + return Err(EngineError::Disabled(cmd.to_string(), why)); + } + let Some(net) = self.live.link() else { return Err(super::unavailable(cmd)) }; + net.join_resuming(id, &link, &name, None).map_err(|e| crate::bad(cmd, e))?; + if let Some(live) = self.live.sessions.get_mut(&id) { + live.status = Status::Connecting; + } + Ok(()) + } + + /// Session `id` is over with a message: it couldn't be joined, it was refused or ended for + /// good, or sync broke. Its tab shows why, and a session you were in can be joined again there. pub(super) fn live_fail(&mut self, id: SessionId, reason: String) { log::warn!("live session: {reason}"); self.live.leave(id); if let Some(live) = self.live.sessions.get_mut(&id) { - live.reset(); + live.strand(); live.status = Status::Failed { reason: reason.clone() }; } self.message = Some(reason); @@ -456,7 +502,7 @@ impl Session { let live = self.live.doc(uid)?; match &live.client { Some(c) => limited_by(c.access()), - None if live.status.active() && live.role() != Some(Role::Host) => Some(PUBLISHING), + None if live.status.active() && live.role() != Some(Role::Host) && !live.disconnected() => Some(PUBLISHING), None => None, } } diff --git a/crates/engine/src/sharing/live/join.rs b/crates/engine/src/sharing/live/join.rs index 836267b2a..75ade0c80 100644 --- a/crates/engine/src/sharing/live/join.rs +++ b/crates/engine/src/sharing/live/join.rs @@ -48,6 +48,24 @@ fn leave(s: &mut Session, p: &Value) -> Result { Ok(json!({"ended": ended.is_some_and(|l| l.status.active())})) } +/// Back into the open file's session now: a reconnect that is waiting tries at once, and a session +/// that couldn't go on is joined again in its tab. +fn reconnect(s: &mut Session, _: &Value) -> Result { + const CMD: &str = "collab.reconnect"; + let live = s.live(); + let (id, reconnecting, rejoinable) = (live.id, matches!(live.status, Status::Reconnecting { .. }), live.rejoinable()); + if reconnecting { + if let Some(net) = s.live.link() { + net.retry(id); + } + } else if rejoinable { + s.live_rejoin(CMD, id)?; + } else { + return Err(bad(CMD, "this file has no lost live session to reconnect")); + } + Ok(json!({"status": s.live().status.id()})) +} + fn stop(s: &mut Session, p: &Value) -> Result { // A file shared online stays its owner's on the server: stopping takes the others' links (and // their codes) away, which disconnects whoever joined with them, and the owner stays in the @@ -192,10 +210,17 @@ pub fn specs() -> Vec { false, join ), + spec!( + "collab.reconnect", + "Reconnect live session", + "{} → {status}: for the open file's session when it is disconnected: a reconnect that is waiting tries now, and a session that couldn't go on (`failed`, the file open in its tab) is joined again there, what you changed since opening beside it. Refused when nothing was lost", + false, + reconnect + ), spec!( "collab.leave", "Leave live session", - "{session? (a session's id from `collab.peers`'s `sessions`, e.g. a join still on its way; default: the open file's session)} → {ended}: the synced file stays open as a local copy", + "{session? (a session's id from `collab.peers`'s `sessions`, e.g. a join still on its way; default: the open file's session)} → {ended}: the synced file stays open as a local copy (also how a tab stops showing a session that couldn't go on)", false, leave ), diff --git a/crates/engine/src/sharing/live/sync.rs b/crates/engine/src/sharing/live/sync.rs index 79f693731..0930151dc 100644 --- a/crates/engine/src/sharing/live/sync.rs +++ b/crates/engine/src/sharing/live/sync.rs @@ -118,7 +118,14 @@ impl Session { for m in std::mem::take(&mut live.deferred) { match client.receive(m, &EngineRecompute) { Ok(more) => out.extend(more), - Err(e) => return self.live_fail(id, format!("The live session got out of step ({e}); join again")), + Err(e) => { + self.live_fail(id, format!("The live session got out of step ({e})")); + // A fresh join has nothing to get out of step with. + if let Err(e) = self.live_rejoin("collab.reconnect", id) { + log::warn!("live session: couldn't join again: {e}"); + } + return; + } } } let doc = Arc::new(client.document().clone()); diff --git a/crates/engine/src/sharing/live/types.rs b/crates/engine/src/sharing/live/types.rs index 93455c0d8..f2108eb36 100644 --- a/crates/engine/src/sharing/live/types.rs +++ b/crates/engine/src/sharing/live/types.rs @@ -496,6 +496,8 @@ pub trait LiveLink: Send { /// End session `session` (stop hosting or leave); the others go on. Nothing for a session /// that is over already. fn leave(&mut self, session: SessionId); + /// Session `session` is waiting to reconnect: try now. + fn retry(&mut self, _session: SessionId) {} /// Your latest presence in `session` (the link throttles and coalesces). fn presence(&mut self, session: SessionId, p: &Presence); /// A message for `session`'s server (document changes), sent in order. diff --git a/crates/ui-egui/locales/en.toml b/crates/ui-egui/locales/en.toml index 11f0bb4ac..e24d81452 100644 --- a/crates/ui-egui/locales/en.toml +++ b/crates/ui-egui/locales/en.toml @@ -406,6 +406,12 @@ name = "English" "Decoration style" = "Decoration style" "Decoration" = "Decoration" "Deep select" = "Deep select" +"Disconnected. Reconnecting…" = "Disconnected. Reconnecting…" +"Disconnected. Your edits stay here and sync when the connection returns. Reconnecting…" = "Disconnected. Your edits stay here and sync when the connection returns. Reconnecting…" +"Dismiss" = "Dismiss" +"Not connected: {reason}. Your work is still in this tab." = "Not connected: {reason}. Your work is still in this tab." +"Retry now" = "Retry now" +"Save a copy…" = "Save a copy…" "Wrap style" = "Wrap style" "Balance" = "Balance" "Pretty" = "Pretty" @@ -415,6 +421,7 @@ name = "English" "Set hug contents" = "Set hug contents" "Set fill container" = "Set fill container" "Double-click an edge of an auto layout frame" = "Double-click an edge of an auto layout frame" +"{name}, disconnected from its live session" = "{name}, disconnected from its live session" "⌥ double-click an edge of an auto layout child" = "⌥ double-click an edge of an auto layout child" "Alt+double-click an edge of an auto layout child" = "Alt+double-click an edge of an auto layout child" "Default" = "Default" diff --git a/crates/ui-egui/src/canvas.rs b/crates/ui-egui/src/canvas.rs index 5ccfe5d46..6cea501c1 100644 --- a/crates/ui-egui/src/canvas.rs +++ b/crates/ui-egui/src/canvas.rs @@ -312,11 +312,13 @@ pub fn show(app: &mut FigApp, ui: &mut Ui, worker: &mut Worker) { crate::live::draw(app, &painter, &ctx); crate::live::agent::draw(app, &painter, &ctx); crate::live::chat_box(app, &ctx); - crate::live::spotlight_banner(app, &ctx, rect); + // The other banners float below the strip that says the session is disconnected. + let banners = rect.with_min_y(rect.min.y + crate::live::connection_banner(app, &ctx, rect)); + crate::live::spotlight_banner(app, &ctx, banners); crate::comments::popovers(app, &ctx); - crate::versions::preview_banner(app, &ctx, rect); - crate::branch_ui::banner(app, &ctx, rect); - crate::dev_mode::focus_bar(app, &ctx, rect); + crate::versions::preview_banner(app, &ctx, banners); + crate::branch_ui::banner(app, &ctx, banners); + crate::dev_mode::focus_bar(app, &ctx, banners); // The cursor: the tool's, unless something on the canvas with its own has the pointer. app.canvas.cursor = None; if let (true, Some(at)) = (own_cursor && resp.hovered(), ctx.input(|i| i.pointer.hover_pos())) { diff --git a/crates/ui-egui/src/icons.rs b/crates/ui-egui/src/icons.rs index f8480127f..cd6f34a08 100644 --- a/crates/ui-egui/src/icons.rs +++ b/crates/ui-egui/src/icons.rs @@ -205,6 +205,8 @@ icons! { Online, /// In a live session others get into: a point sending waves to both sides. LiveShared, + /// Disconnected from a live session: the point and its waves, struck through. + LiveDisconnected, /// A file of yours on a server that only you get into: a point inside a closed ring. LivePrivate, /// A design file: a pen nib leaning to the left, its grip filled. diff --git a/crates/ui-egui/src/icons/draw.rs b/crates/ui-egui/src/icons/draw.rs index 1d3925507..904b54107 100644 --- a/crates/ui-egui/src/icons/draw.rs +++ b/crates/ui-egui/src/icons/draw.rs @@ -809,6 +809,14 @@ fn draw(p: &P, icon: Icon) { p.arc(8.0, 8.0, r, PI - FRAC_PI_4, PI + FRAC_PI_4); } } + LiveDisconnected => { + p.dot(8.0, 8.0, 1.6); + for r in [3.6, 6.0] { + p.arc(8.0, 8.0, r, -FRAC_PI_4, FRAC_PI_4); + p.arc(8.0, 8.0, r, PI - FRAC_PI_4, PI + FRAC_PI_4); + } + p.line(3.0, 13.0, 13.0, 3.0); + } LivePrivate => { p.dot(8.0, 8.0, 1.6); p.circle(8.0, 8.0, 5.0); diff --git a/crates/ui-egui/src/live.rs b/crates/ui-egui/src/live.rs index efa64617c..c80548323 100644 --- a/crates/ui-egui/src/live.rs +++ b/crates/ui-egui/src/live.rs @@ -436,6 +436,70 @@ pub fn chat_box(app: &mut FigApp, ctx: &egui::Context) { app.ui.live.chat = Some(chat); } +/// While the active document's live session is disconnected: a thin strip across the top of the +/// canvas that says so, with what can be done about it. Returns how much of the canvas it takes. +pub fn connection_banner(app: &mut FigApp, ctx: &egui::Context, canvas: Rect) -> f32 { + const H: f32 = 30.0; + let live = app.session.live(); + let Some(uid) = app.session.docs.get(app.session.active).map(|d| d.uid).filter(|u| crate::tabs::live_lost(app, *u)) else { return 0.0 }; + let (text, retry, gone) = match &live.status { + Status::Failed { reason } => { + (trf("Not connected: {reason}. Your work is still in this tab.", &[("reason", &failure_text(reason))]), Some(tr("Try again")), true) + } + Status::Reconnecting { .. } => { + (tr("Disconnected. Your edits stay here and sync when the connection returns. Reconnecting…").to_string(), Some(tr("Retry now")), false) + } + _ => (tr("Disconnected. Reconnecting…").to_string(), None, false), + }; + // Saved as a file of its own: it is one now. + if gone && app.session.docs.iter().any(|d| d.uid == uid && d.path.is_some()) { + app.run("collab.leave", json!({})); + return 0.0; + } + let t = Tokens::get(ctx); + let ink = text_on(t.danger); + let (mut try_now, mut save, mut dismiss) = (false, false, false); + let strip = Rect::from_min_size(canvas.min, vec2(canvas.width(), H)); + egui::Area::new(egui::Id::unique("live_connection_banner")).order(egui::Order::Foreground).fixed_pos(strip.min).show(ctx, |ui| { + ui.set_min_size(strip.size()); + ui.set_max_size(strip.size()); + ui.painter().rect_filled(strip, egui::CornerRadius::ZERO, t.danger); + let button = |ui: &mut Ui, label: &str| { + ui.add( + egui::Button::new(egui::RichText::new(label).font(regular(11.0)).color(ink)) + .fill(Color32::TRANSPARENT) + .stroke(Stroke::new(1.0, ink)) + .corner_radius(egui::CornerRadius::same(4)), + ) + .clicked() + }; + let inner = strip.shrink2(vec2(12.0, 0.0)); + ui.scope_builder(egui::UiBuilder::new().max_rect(inner).layout(egui::Layout::right_to_left(egui::Align::Center)), |ui| { + ui.spacing_mut().item_spacing.x = 8.0; + if gone { + dismiss = button(ui, tr("Dismiss")); + save = button(ui, tr("Save a copy…")); + } + if let Some(label) = retry { + try_now = button(ui, label); + } + ui.with_layout(egui::Layout::left_to_right(egui::Align::Center), |ui| { + ui.add(egui::Label::new(egui::RichText::new(text).font(regular(11.0)).color(ink)).truncate()); + }); + }); + }); + if try_now { + app.run("collab.reconnect", json!({})); + } + if save { + app.save_as_dialog(); + } + if dismiss { + app.run("collab.leave", json!({})); + } + H +} + /// While you spotlight, or someone does and you stopped following them: a banner on the canvas to /// stop, or to follow again. pub fn spotlight_banner(app: &mut FigApp, ctx: &egui::Context, canvas: Rect) { diff --git a/crates/ui-egui/src/live_joining_tests.rs b/crates/ui-egui/src/live_joining_tests.rs index a5d46b8e8..7929a9334 100644 --- a/crates/ui-egui/src/live_joining_tests.rs +++ b/crates/ui-egui/src/live_joining_tests.rs @@ -97,3 +97,52 @@ fn leave_takes_a_session_id() { assert_eq!(h.app.session.execute("collab.leave", &json!({"session": id})).unwrap()["ended"], true); assert!(h.app.session.live.get(id).is_none()); } + +/// Joined, with the file in its tab. +fn joined(h: &mut Harness) -> u64 { + join(h); + h.welcome(); + h.sync(); + h.app.session.docs[h.app.session.active].uid +} + +const STRIP: &str = "Disconnected. Your edits stay here and sync when the connection returns. Reconnecting…"; + +#[test] +fn a_dropped_connection_shows_on_the_tab_and_across_the_canvas_with_retry() { + let (mut h, _) = Harness::new(); + let uid = joined(&mut h); + assert!(!crate::tabs::live_lost(&h.app, uid)); + assert!(!has_text(&h.frames(2), STRIP), "not while connected"); + h.send(vec![LiveEvent::Status(Status::Reconnecting { attempt: 3 })]); + let shapes = h.frames(2); + assert!(crate::tabs::live_lost(&h.app, uid)); + assert!(has_text(&shapes, STRIP)); + h.click(text_center(&shapes, "Retry now")); + assert!(h.app.ui.last_error.is_none(), "{:?}", h.app.ui.last_error); + assert_eq!(h.app.session.live().status, Status::Reconnecting { attempt: 3 }); + // Back: the strip goes. + h.welcome(); + assert!(!has_text(&h.frames(2), STRIP)); + assert!(!crate::tabs::live_lost(&h.app, uid)); +} + +#[test] +fn a_session_that_ended_stays_on_its_tab_until_dismissed() { + let (mut h, _) = Harness::new(); + let uid = joined(&mut h); + h.send(vec![LiveEvent::Status(Status::Failed { reason: "The host refused: this link isn't valid".into() })]); + let shapes = h.frames(2); + assert!(crate::tabs::live_lost(&h.app, uid)); + assert!(has_text(&shapes, "Not connected: The host refused: this link isn't valid. Your work is still in this tab.")); + assert!(has_text(&shapes, "Save a copy…")); + h.click(text_center(&shapes, "Try again")); + assert_eq!(h.app.session.live().status, Status::Connecting); + assert!(crate::tabs::live_lost(&h.app, uid), "still disconnected while it tries"); + h.send(vec![LiveEvent::Status(Status::Failed { reason: "no".into() })]); + let shapes = h.frames(2); + h.click(text_center(&shapes, "Dismiss")); + assert!(!has_text(&h.frames(2), "Try again")); + assert!(!crate::tabs::live_lost(&h.app, uid)); + assert_eq!(h.app.session.live().status, Status::Off); +} diff --git a/crates/ui-egui/src/live_tests.rs b/crates/ui-egui/src/live_tests.rs index cd8454afa..bd2dd4ce2 100644 --- a/crates/ui-egui/src/live_tests.rs +++ b/crates/ui-egui/src/live_tests.rs @@ -126,7 +126,7 @@ impl Harness { color: "#14AE5C".into(), file_name: "Shared".into(), pages: vec![], - link: String::new(), + link: "figcraft://join?host=127.0.0.1:47900&file=f&token=t".into(), url: String::new(), address: "127.0.0.1:47900".into(), network: false, diff --git a/crates/ui-egui/src/tabs.rs b/crates/ui-egui/src/tabs.rs index e59552d12..e043eb2d9 100644 --- a/crates/ui-egui/src/tabs.rs +++ b/crates/ui-egui/src/tabs.rs @@ -149,6 +149,7 @@ pub fn tab_bar(app: &mut FigApp, ui: &mut Ui) { let name = &d.doc.name; // In a live session: a mark after the name says whether others get in. let live = live_mark(app, d.uid); + let lost = live_lost(app, d.uid); let dirty = d.is_dirty(); let r = tab_rects[slot]; let body_r = Rect::from_min_max(r.min, pos2(r.max.x - 32.0, r.max.y)); @@ -171,8 +172,8 @@ pub fn tab_bar(app: &mut FigApp, ui: &mut Ui) { } }); } - let label = tab_label(name, live); - let resp = if live.is_some() { resp.on_hover_text(&label) } else { resp }; + let label = if lost { trf("{name}, disconnected from its live session", &[("name", name)]) } else { tab_label(name, live) }; + let resp = if live.is_some() || lost { resp.on_hover_text(&label) } else { resp }; crate::a11y::tab(&resp, &label, active); let progress = open_progress(&opening, d.uid, now); let tab = TabPaint { @@ -182,6 +183,7 @@ pub fn tab_bar(app: &mut FigApp, ui: &mut Ui) { width: w, state: [active, resp.hovered(), cresp.hovered(), dirty], live, + lost, progress, }; paint_tab(ui, &t, &tab); @@ -432,7 +434,7 @@ fn detach_tab(app: &mut FigApp, ui: &Ui, i: usize, at: Option) { /// How wide document tab `i` wants to be: its name, the live mark and the unsaved dot. fn preferred_width(app: &FigApp, i: usize) -> f32 { let d = &app.session.docs[i]; - let mark_w = if live_mark(app, d.uid).is_some() { 18.0 } else { 0.0 }; + let mark_w = if live_mark(app, d.uid).is_some() || live_lost(app, d.uid) { 18.0 } else { 0.0 }; let dot_w = if d.is_dirty() && mark_w > 0.0 { 12.0 } else { 0.0 }; (regular(12.0).size * 0.6 * d.doc.name.chars().count() as f32 + 64.0 + mark_w + dot_w).clamp(120.0, 220.0) } @@ -457,14 +459,16 @@ struct TabPaint<'a> { state: [bool; 4], /// In a live session: whether others get in. live: Option, + /// Disconnected from its live session. + lost: bool, /// How far the opening animation is (0..=1). progress: f32, } /// Draws a document tab: background, icon, name, live mark, unsaved dot and close button. fn paint_tab(ui: &Ui, t: &Tokens, tab: &TabPaint) { - let TabPaint { name, kind, rects: [r, close_r], width: w, state: [active, hovered, close_hovered, dirty], live, progress } = *tab; - let mark_w = if live.is_some() { 18.0 } else { 0.0 }; + let TabPaint { name, kind, rects: [r, close_r], width: w, state: [active, hovered, close_hovered, dirty], live, lost, progress } = *tab; + let mark_w = if live.is_some() || lost { 18.0 } else { 0.0 }; let paint_r = r.translate(vec2(-28.0 * (1.0 - progress).powi(2), 0.0)); let painter = ui.painter().with_clip_rect(r); if active { @@ -487,8 +491,10 @@ fn paint_tab(ui: &Ui, t: &Tokens, tab: &TabPaint) { t.text.linear_multiply(progress), (w - mark_w - if dirty { 72.0 } else { 60.0 }).max(0.0), ); - if let Some(shared) = live { - let at = Rect::from_center_size(pos2(paint_r.max.x - 38.0, paint_r.center().y), vec2(14.0, 14.0)); + let at = Rect::from_center_size(pos2(paint_r.max.x - 38.0, paint_r.center().y), vec2(14.0, 14.0)); + if lost { + icons::paint(&painter, at, Icon::LiveDisconnected, t.danger.linear_multiply(progress)); + } else if let Some(shared) = live { icons::paint(&painter, at, crate::online::mark(shared), if shared { t.success } else { t.icon_secondary }.linear_multiply(progress)); } if dirty { @@ -672,6 +678,12 @@ pub fn live_mark(app: &FigApp, uid: u64) -> Option { live.shared().filter(|_| live.doc_uid == Some(uid)) } +/// Document `uid` is a live session's whose connection is gone: what happens in it reaches no one. +pub fn live_lost(app: &FigApp, uid: u64) -> bool { + let live = app.session.live_of(uid); + live.disconnected() && live.doc_uid == Some(uid) +} + /// Minimise, maximise/restore and close at the right end of the title bar; returns their width. fn window_buttons(app: &mut FigApp, ui: &mut Ui, bar: Rect) -> f32 { const W: f32 = 46.0; diff --git a/docs/collab-protocol.md b/docs/collab-protocol.md index b4b459afb..9852a1686 100644 --- a/docs/collab-protocol.md +++ b/docs/collab-protocol.md @@ -535,9 +535,21 @@ resuming with its offline edits, and leaving the session going back to snapshot `ws:///` and sends the token only inside its `Hello`. Links are pasted into the Join dialog (its field has the keyboard when it opens; Escape closes it), or opened from anywhere: see *Opening invite links* below. -- Reconnect: a connection that drops after it was live reconnects with growing pauses (0.5 s, - doubling up to 10 s, 12 attempts); a refusal (wrong token, full) or the host ending the session - is final; a first join that can't reach the host fails at once. A reconnect resumes as the same +- Reconnect: a connection that drops after it was live reconnects until you leave, with growing + pauses (0.5 s, doubling up to 10 s, 12 attempts), then a minute's pause and as many again. A + server turning a live session away is taken as passing (busy, a limit, a file it couldn't read + just now) unless it says why in words that are for good: the link, the file or your place in it + is gone, or the versions can't talk (`FINAL_REFUSALS` in `crates/collab-net/src/join.rs`). Those + and the host ending the session are final; a first join that can't reach the host fails at once. + Messages queued while offline go out paced, so they stay within what a server takes from a + connection. +- Disconnected: while it reconnects, and after it can't go on, the tab shows a red mark and a strip + across the top of the canvas says so, with **Retry now** (`collab.reconnect`). The tab can still + be edited. A session you joined that ended stays on its tab with **Try again**, **Save a copy…** + and **Dismiss** (`collab.leave`). **Try again** joins it in the same tab as someone new; if the + tab changed since it lost the server, what changed opens beside it as an unsaved + "offline copy" and the tab shows the server's version. A session that gets out of step does the + same by itself. A reconnect resumes as the same client: the guest's connection keeps the session id, client id and last sequence number from the messages it read and sends them in its next `Hello` (see Offline and reconnect). - The browser build joins and publishes to a server (see *Built (the browser client)*); hosting diff --git a/docs/collaboration-parity.md b/docs/collaboration-parity.md index fb54a8d55..deab037ae 100644 --- a/docs/collaboration-parity.md +++ b/docs/collaboration-parity.md @@ -26,6 +26,7 @@ cursors, selections, text carets, cursor chat and spotlights shown live ([`colla | Date | Change | Summary | |---|---|---| +| 2026-11-04 | minor | M9: a session you joined no longer turns into a local file by itself. A dropped connection reconnects until you leave (a minute's pause after each 12 attempts, a pause Retry now skips); only refusals that are for good stop it; queued messages go out paced. While disconnected the tab shows a red mark and a strip across the canvas says so; a session that ended stays on its tab with Try again, Save a copy and Dismiss, and what changed offline opens beside it when it joins again | | 2026-10-31 | patch | M9, #1665: the document in a live session's welcome goes compressed (deflate) to clients whose hello announces `packed`, about seven times smaller on the wire (a synthetic file of 4,040 layers: 397 KB to 53 KB); older clients keep getting it uncompressed, and the client reads a packed document only up to the size an uncompressed welcome could have had. | | 2026-10-30 | minor | M9, part of #742: a team list on `figcraft-server`. With `team_files = "edit"` (or `comment`, `view`) everyone who reaches the server sees the files shared from it now (a link of theirs has a join code: Share ▸ Start sharing) and joins them by a team list link, `figcraft://join?server=…&file=…&team=1`, which carries no secret: the server hands out the file's team link while it stays shared (`ResolveTeam`, answered like a join code), and Stop sharing revokes it with the others, disconnecting whoever joined with it. Turning the list off or changing its access refuses the old team links at once. `collab.files {refresh}` asks the server in Preferences for its list (`Admin::Team`; not the public server FigCraft ships with), and Home ▸ Online shows it below your own files: a row joins its file, its menu copies the link for the app or, with the server's `web_url`, for the browser build (`#join=…`). A browser build served beside its server names it with ``, the page's default for sharing, codes and the team list. No accounts: the list is for a server only your team reaches. Tests: `team_tests` (the server and apps over loopback), `code_tests` (links and the browser's driver), `online_tests` (the Online section), `query_tests` (the page's server). Percentages and hours unchanged | | 2026-10-30 | minor | M9, #1130: `figcraft-server` compares the SHA-256 digests of the presented and the kept server key instead of the raw strings, as `docs/server.md` says; the key file, `figcraft-server key` and rotation are unchanged. Numbers unchanged. | diff --git a/docs/mcp.md b/docs/mcp.md index b5e49e749..aae5e5860 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -1478,7 +1478,10 @@ only: a host stops hosting it, a guest leaves it. see `FIGCRAFT_CA_FILE` in [`server.md`](server.md#behind-a-reverse-proxy-tls)). `collab.leave {}` leaves the active document's session (or ends a join started from it that is still on its way or failed), and `collab.leave {session}` the session with that id (from `sessions`), whichever - document is active; the synced document stays open as your copy. + document is active; the synced document stays open as your copy. `collab.reconnect {}` acts on + the active document's lost session: a reconnect that is waiting tries now, and a session you + joined that couldn't go on is joined again in its tab. A dropped connection keeps reconnecting + until you leave (see *Reconnect* in [`collab-protocol.md`](collab-protocol.md)). - `collab.publish {server, name?, key?}` puts the active document on a `figcraft-server` ([`server.md`](server.md)) and syncs it from there (as that document's session; others go on): `server` is the address `figcraft-server key` prints (`ws://host:47900/?key=…`; or give `key` separately), `name` the diff --git a/docs/server.md b/docs/server.md index 9d42cb0fc..6fb4056a5 100644 --- a/docs/server.md +++ b/docs/server.md @@ -43,7 +43,7 @@ Releases also ship it ready-built for Linux (`figcraft-server--linux-x8 tells everyone it is restarting. Every change it accepted is on disk before anyone is told about it, so even a hard kill loses nothing confirmed (a file's log is then folded into a new snapshot the next time everyone leaves it). Apps that were in a file keep editing and reconnect once the - server is back (for about a minute and a half), as the same participants. + server is back, as the same participants. ## Publishing a file From 436b893741fcafa3e88f5666ccc8d34e4674e4c0 Mon Sep 17 00:00:00 2001 From: fluid-chey Date: Wed, 4 Nov 2026 12:49:11 -0700 Subject: [PATCH 2/8] M9: pace queued sends without blocking, and address review comments Queued sync messages wait in a list the session loop drains, so Leave, a lost connection, presence and pings are never held up behind them. A text frame from the host stays final. A session that ended stays on its tab after a Save as: ending it there could turn a tab into a plain local file with no word about it. Co-Authored-By: Claude Opus 5.5 --- crates/collab-net/src/driver_tests.rs | 13 +++++++++++++ crates/collab-net/src/guest.rs | 19 ++++++++++++------- crates/collab-net/src/join.rs | 3 ++- crates/ui-egui/src/live.rs | 9 +++------ docs/collaboration-parity.md | 4 ++-- 5 files changed, 32 insertions(+), 16 deletions(-) diff --git a/crates/collab-net/src/driver_tests.rs b/crates/collab-net/src/driver_tests.rs index 57921df3d..df4d5c492 100644 --- a/crates/collab-net/src/driver_tests.rs +++ b/crates/collab-net/src/driver_tests.rs @@ -349,6 +349,19 @@ fn a_session_that_was_live_only_stops_for_a_refusal_that_is_for_good() { assert!(d.done(), "{reason}"); assert!(matches!(statuses(&inbox.drain()).last(), Some(Status::Failed { reason: r }) if r.contains(reason)), "{reason}"); } + // What this end finds it can't talk to: a text frame, or a message it can't read. + for text in [true, false] { + let (mut d, inbox) = start(Script::default(), t0); + d.opened(t0); + d.message(&welcome(), t0); + if text { + d.text(t0); + } else { + d.message(&[0xFF, 1, 2, 3], t0); + } + assert!(d.done(), "text: {text}"); + assert!(matches!(statuses(&inbox.drain()).last(), Some(Status::Failed { .. }))); + } } #[test] diff --git a/crates/collab-net/src/guest.rs b/crates/collab-net/src/guest.rs index 84572b884..02685339f 100644 --- a/crates/collab-net/src/guest.rs +++ b/crates/collab-net/src/guest.rs @@ -9,6 +9,7 @@ //! Publishing ([`Guest::publish`]) first uploads the document to a server over a managing //! connection, then joins the new file with its edit link on the same thread. +use std::collections::VecDeque; use std::io; use std::net::TcpStream; use std::sync::Arc; @@ -280,17 +281,17 @@ fn session( let mut welcomed = false; let mut throttle = Throttle::new(PRESENCE_INTERVAL); let mut pace = Pace::new(Instant::now()); + // Sync messages waiting their turn (a long queue from while offline), and when each may go. + let mut paced: VecDeque<(Instant, ClientMsg)> = VecDeque::new(); let outcome = if send(&mut writer, &hello) { loop { let ping_at = last_ping + cfg.limits.keepalive; - let wake = earliest([throttle.due(), Some(ping_at)]).unwrap_or(ping_at); + let wake = earliest([throttle.due(), Some(ping_at), paced.front().map(|(at, _)| *at)]).unwrap_or(ping_at); let mut out = None; - let mut now_send = Vec::new(); match rx.recv_timeout(wake.saturating_duration_since(Instant::now())) { Ok(Cmd::Send(m)) if welcomed => { - // A long queue from while offline doesn't go out faster than the server takes it. - std::thread::sleep(pace.wait(Instant::now())); - now_send.push(m); + let now = Instant::now(); + paced.push_back((now + pace.wait(now), m)); } // Made for the previous connection (the sync client sends again what the server // didn't get once it has caught up with this connection's welcome). @@ -312,10 +313,14 @@ fn session( Ok(Cmd::Welcomed(_) | Cmd::Lost(..) | Cmd::Retry) => {} Err(RecvTimeoutError::Timeout) => {} } - if !now_send.iter().all(|m| send(&mut writer, m)) { + let now = Instant::now(); + let mut sent = true; + while sent && let Some((_, m)) = paced.pop_front_if(|(at, _)| *at <= now) { + sent = send(&mut writer, &m); + } + if !sent { break Outcome::Retry("the connection dropped".into()); } - let now = Instant::now(); let out = out.or_else(|| throttle.take_due(now)); let ok = out.is_none_or(|p| send(&mut writer, &ClientMsg::Presence(p))); let ok = ok diff --git a/crates/collab-net/src/join.rs b/crates/collab-net/src/join.rs index 63f94bec2..aa1c008eb 100644 --- a/crates/collab-net/src/join.rs +++ b/crates/collab-net/src/join.rs @@ -158,7 +158,7 @@ pub(crate) enum Next { /// gone, or the two versions can't talk. Anything else a server refuses a session with (it is /// busy, a limit was hit, the file couldn't be read just now) may pass, and servers of any version /// say so in these words, so a reason not listed here means try again. -const FINAL_REFUSALS: [&str; 11] = [ +const FINAL_REFUSALS: [&str; 12] = [ "isn't valid", "isn't on the server", "was removed from the server", @@ -170,6 +170,7 @@ const FINAL_REFUSALS: [&str; 11] = [ "couldn't be read", "message too large", "this version can't read", + "sent a text message", ]; /// Whether `reason`, from a server refusing a session that was live, is for good. diff --git a/crates/ui-egui/src/live.rs b/crates/ui-egui/src/live.rs index c80548323..ac00b2c4e 100644 --- a/crates/ui-egui/src/live.rs +++ b/crates/ui-egui/src/live.rs @@ -441,7 +441,9 @@ pub fn chat_box(app: &mut FigApp, ctx: &egui::Context) { pub fn connection_banner(app: &mut FigApp, ctx: &egui::Context, canvas: Rect) -> f32 { const H: f32 = 30.0; let live = app.session.live(); - let Some(uid) = app.session.docs.get(app.session.active).map(|d| d.uid).filter(|u| crate::tabs::live_lost(app, *u)) else { return 0.0 }; + if !app.session.docs.get(app.session.active).is_some_and(|d| crate::tabs::live_lost(app, d.uid)) { + return 0.0; + } let (text, retry, gone) = match &live.status { Status::Failed { reason } => { (trf("Not connected: {reason}. Your work is still in this tab.", &[("reason", &failure_text(reason))]), Some(tr("Try again")), true) @@ -451,11 +453,6 @@ pub fn connection_banner(app: &mut FigApp, ctx: &egui::Context, canvas: Rect) -> } _ => (tr("Disconnected. Reconnecting…").to_string(), None, false), }; - // Saved as a file of its own: it is one now. - if gone && app.session.docs.iter().any(|d| d.uid == uid && d.path.is_some()) { - app.run("collab.leave", json!({})); - return 0.0; - } let t = Tokens::get(ctx); let ink = text_on(t.danger); let (mut try_now, mut save, mut dismiss) = (false, false, false); diff --git a/docs/collaboration-parity.md b/docs/collaboration-parity.md index deab037ae..6c4aa9340 100644 --- a/docs/collaboration-parity.md +++ b/docs/collaboration-parity.md @@ -1,6 +1,6 @@ # Collaboration parity -> **Last reviewed:** 2026-10-10 · **Last updated:** 2026-10-31 · **Change:** minor (a live session's welcome sends the document compressed to clients that read it, about seven times smaller (#1665); a server can list the files shared from it now for everyone who reaches it, Home ▸ Online shows them and their links carry no secret (`team_files`, part of #742); `figcraft-server` checks the server key by comparing SHA-256 digests, not the raw keys (#1130); joining and leaving a file nobody is editing reads and writes nothing on the server: it stays loaded a minute after everyone leaves (#1074); `comment.people` lists names that were only @mentioned, and `comment.next` and `comment.prompt` carry the mentions too (#1415); share links can comment: a third access level beside edit and view, whose holder sees the file live and adds, replies to, reacts to and resolves comments but can't change the design; `comment.list {mentions}` lists only the threads that @mention a name (#1415); the browser build publishes files to a server, so Share ▸ Online and Publish to server work there (#917); an AI agent attached over MCP shows where it works (#1416); one name for live sessions and comments: Preferences ▸ Your name, in the Join dialog too, kept in the browser, and a rename shows at once in the sessions you are in (#1380); the Join dialog shows a join from the click until its file is ready, with Cancel, and a failed join with Try again (#1381); requests that change a server's files have a budget per address, the server key's too, and join codes are found by their hash (#1075); comment.list reports each message's @mentions (#1415); the batches kept for reconnecting apps are bounded in bytes by default, and `figcraft-server` takes `max_log_mb` (#1069); sibling placement in live sessions searches instead of counting (#1065); an AI agent attached over MCP comments under its own name, with an Agent label (#1414); a refused batch's fixes take time linear in the batch (#1063); image entries are checked as they apply; branches in live sessions and branch history entries; server message limits follow `max_file_mb`; a damaged log generation field keeps flushed edits; the web build joins from a `#join=` link; removing someone from a live session; live-session revert guard; live sessions per document; a public figcraft-server: open hosting, owner tokens, per-address limits, idle expiry, moderation, install.sh; cursor chat and spotlight in live sessions; branch version history and finer list merges; comments and version history behaviour suite and fixes; hosting the server: `wss://` in the desktop app, server releases, deployment guide; team libraries on the server and swap library; branches: create, update from main, review and merge; wording: the standalone server with share links to edit or view; numbers unchanged) · **Target:** Figma Design (desktop app 126.10.7) +> **Last reviewed:** 2026-10-10 · **Last updated:** 2026-11-04 · **Change:** minor (M9: a joined live session reconnects until you leave and says when it is disconnected, with a red tab mark and a strip across the canvas; a live session's welcome sends the document compressed to clients that read it, about seven times smaller (#1665); a server can list the files shared from it now for everyone who reaches it, Home ▸ Online shows them and their links carry no secret (`team_files`, part of #742); `figcraft-server` checks the server key by comparing SHA-256 digests, not the raw keys (#1130); joining and leaving a file nobody is editing reads and writes nothing on the server: it stays loaded a minute after everyone leaves (#1074); `comment.people` lists names that were only @mentioned, and `comment.next` and `comment.prompt` carry the mentions too (#1415); share links can comment: a third access level beside edit and view, whose holder sees the file live and adds, replies to, reacts to and resolves comments but can't change the design; `comment.list {mentions}` lists only the threads that @mention a name (#1415); the browser build publishes files to a server, so Share ▸ Online and Publish to server work there (#917); an AI agent attached over MCP shows where it works (#1416); one name for live sessions and comments: Preferences ▸ Your name, in the Join dialog too, kept in the browser, and a rename shows at once in the sessions you are in (#1380); the Join dialog shows a join from the click until its file is ready, with Cancel, and a failed join with Try again (#1381); requests that change a server's files have a budget per address, the server key's too, and join codes are found by their hash (#1075); comment.list reports each message's @mentions (#1415); the batches kept for reconnecting apps are bounded in bytes by default, and `figcraft-server` takes `max_log_mb` (#1069); sibling placement in live sessions searches instead of counting (#1065); an AI agent attached over MCP comments under its own name, with an Agent label (#1414); a refused batch's fixes take time linear in the batch (#1063); image entries are checked as they apply; branches in live sessions and branch history entries; server message limits follow `max_file_mb`; a damaged log generation field keeps flushed edits; the web build joins from a `#join=` link; removing someone from a live session; live-session revert guard; live sessions per document; a public figcraft-server: open hosting, owner tokens, per-address limits, idle expiry, moderation, install.sh; cursor chat and spotlight in live sessions; branch version history and finer list merges; comments and version history behaviour suite and fixes; hosting the server: `wss://` in the desktop app, server releases, deployment guide; team libraries on the server and swap library; branches: create, update from main, review and merge; wording: the standalone server with share links to edit or view; numbers unchanged) · **Target:** Figma Design (desktop app 126.10.7) Figma is multiplayer first; FigCraft is local first. **Ready ~13% (estimated):** comments, version history and branches (review and merge) work inside one file; the real-time collaboration core landed in #103 (sync model, @@ -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). Image entries are checked as they apply: a key must be a plain name (letters, digits, `-`, `_`; it becomes `images/` in a saved file), bytes under a content-hash key must hash to it (lowercase sha256), a side is at most 65 535 px and the decoded bytes at most 256 MiB (what the importers read), a key's bytes can't be replaced and an image a layer shows can't be removed, so no collaborator can poison the shared image store or balloon memory; snapshots and server logs from before these checks open without the images they refuse (#1086). The batches kept for apps that reconnect are bounded in bytes by default (64 MiB; `max_log_mb` on `figcraft-server`), and an app further behind catches up with a snapshot (#1069). Placing a layer among its siblings is a binary search, so large pastes and reorders in one batch no longer take quadratic time (#1065). A refused batch's fixes take time linear in the batch: each deleted layer is restored once and each entry fix looked up once (#1063). 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, a live connection reconnecting until you leave), 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). Image entries are checked as they apply: a key must be a plain name (letters, digits, `-`, `_`; it becomes `images/` in a saved file), bytes under a content-hash key must hash to it (lowercase sha256), a side is at most 65 535 px and the decoded bytes at most 256 MiB (what the importers read), a key's bytes can't be replaced and an image a layer shows can't be removed, so no collaborator can poison the shared image store or balloon memory; snapshots and server logs from before these checks open without the images they refuse (#1086). The batches kept for apps that reconnect are bounded in bytes by default (64 MiB; `max_log_mb` on `figcraft-server`), and an app further behind catches up with a snapshot (#1069). Placing a layer among its siblings is a binary search, so large pastes and reorders in one batch no longer take quadratic time (#1065). A refused batch's fixes take time linear in the batch: each deleted layer is restored once and each entry fix looked up once (#1063). Left: server archives beyond Linux | 25–55 | | Live cursors, selections, viewports; follow / observation mode; spotlight | yes | **done, in live sessions**: named cursors (one name for live sessions and comments: Preferences ▸ Your name, also in the Share popover and the Join dialog, kept in the browser too; a rename shows at once in the sessions you're in, #1380), selection outlines, avatars (an AI agent attached over MCP also shows its avatar and an outline around the layers it last edited or selected while it works, in or out of a session, #1416), 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, comment 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, and the Join dialog then shows the join until its file is ready (Cancel), or why it failed (Try again, #1381); 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), and shares online and publishes to a server as the desktop app does, though not on the local network (#917; the owner token of a file it puts on an open server lasts as long as the page)); `figcraft-server` share links per file with edit, comment or view access (a viewer sees changes and presence, a "View only" badge and only the moving tools; a commenter also adds, replies to, reacts to and resolves comment threads, with a "Can comment" badge and the comment tool, #1525; the server refuses their other 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; a server only a team reaches can also list the files shared from it now for everyone (`team_files`): Home ▸ Online shows that team list for the server in Preferences, a row joins its file, and its links (`figcraft://join?server=…&file=…&team=1`, or the browser build's) carry no secret and stop working with the share. No accounts, teams or projects | 8–17 | From e87078a8ec2c591cde5bc164bda24cbef3c01dda Mon Sep 17 00:00:00 2001 From: fluid-chey Date: Wed, 4 Nov 2026 13:03:09 -0700 Subject: [PATCH 3/8] M9: leaving a live session sends what was waiting its turn Co-Authored-By: Claude Opus 5.5 --- crates/collab-net/src/driver.rs | 6 ++++++ crates/collab-net/src/driver_tests.rs | 15 +++++++++++++++ crates/collab-net/src/guest.rs | 2 ++ docs/collaboration-parity.md | 2 +- 4 files changed, 24 insertions(+), 1 deletion(-) diff --git a/crates/collab-net/src/driver.rs b/crates/collab-net/src/driver.rs index d96650371..622b6c52d 100644 --- a/crates/collab-net/src/driver.rs +++ b/crates/collab-net/src/driver.rs @@ -339,6 +339,12 @@ impl Driver { /// Leave the session. pub(crate) fn leave(&mut self, now: Instant) { + // What you did before leaving still reaches the others, whatever its turn. + while let Some((_, msg)) = self.paced.pop_front() { + if !self.write(&msg) { + break; + } + } self.end(Outcome::Left, now); } diff --git a/crates/collab-net/src/driver_tests.rs b/crates/collab-net/src/driver_tests.rs index df4d5c492..83684db7e 100644 --- a/crates/collab-net/src/driver_tests.rs +++ b/crates/collab-net/src/driver_tests.rs @@ -392,6 +392,21 @@ fn a_long_queue_goes_out_paced_and_in_order() { assert_eq!(d.deadline().unwrap() - now, ms(500), "nothing left to send but the reconnect"); } +#[test] +fn leaving_sends_what_was_waiting_its_turn() { + let t0 = Instant::now(); + let (mut d, _inbox) = start(Script::default(), t0); + d.opened(t0); + d.message(&welcome(), t0); + for seq in 0..300 { + d.send(ClientMsg::Ack { seq }, t0); + } + d.leave(t0); + let seqs: Vec = d.sock.sent.iter().filter_map(|m| if let ClientMsg::Ack { seq } = m { Some(*seq) } else { None }).collect(); + assert_eq!(seqs, (0..300).collect::>()); + assert!(d.done()); +} + #[test] fn a_first_join_that_cant_reach_the_host_fails_at_once() { let t0 = Instant::now(); diff --git a/crates/collab-net/src/guest.rs b/crates/collab-net/src/guest.rs index 02685339f..f338fcce2 100644 --- a/crates/collab-net/src/guest.rs +++ b/crates/collab-net/src/guest.rs @@ -306,6 +306,8 @@ fn session( } Ok(Cmd::Lost(e, o)) if e == epoch => break o, Ok(Cmd::Leave) | Err(RecvTimeoutError::Disconnected) => { + // What you did before leaving still reaches the others, whatever its turn. + while paced.pop_front().is_some_and(|(_, m)| send(&mut writer, &m)) {} let _ = writer.close(Some(CloseFrame { code: CloseCode::Normal, reason: "".into() })); break Outcome::Left; } diff --git a/docs/collaboration-parity.md b/docs/collaboration-parity.md index 6c4aa9340..6f1faad8e 100644 --- a/docs/collaboration-parity.md +++ b/docs/collaboration-parity.md @@ -26,7 +26,7 @@ cursors, selections, text carets, cursor chat and spotlights shown live ([`colla | Date | Change | Summary | |---|---|---| -| 2026-11-04 | minor | M9: a session you joined no longer turns into a local file by itself. A dropped connection reconnects until you leave (a minute's pause after each 12 attempts, a pause Retry now skips); only refusals that are for good stop it; queued messages go out paced. While disconnected the tab shows a red mark and a strip across the canvas says so; a session that ended stays on its tab with Try again, Save a copy and Dismiss, and what changed offline opens beside it when it joins again | +| 2026-11-04 | minor | M9: a session you joined no longer turns into a local file by itself. A dropped connection reconnects until you leave (a minute's pause after each 12 attempts, a pause Retry now skips); only the host ending the session and refusals that are for good stop it; queued messages go out paced. While disconnected the tab shows a red mark and a strip across the canvas says so; a session that ended stays on its tab with Try again, Save a copy and Dismiss, and what changed offline opens beside it when it joins again | | 2026-10-31 | patch | M9, #1665: the document in a live session's welcome goes compressed (deflate) to clients whose hello announces `packed`, about seven times smaller on the wire (a synthetic file of 4,040 layers: 397 KB to 53 KB); older clients keep getting it uncompressed, and the client reads a packed document only up to the size an uncompressed welcome could have had. | | 2026-10-30 | minor | M9, part of #742: a team list on `figcraft-server`. With `team_files = "edit"` (or `comment`, `view`) everyone who reaches the server sees the files shared from it now (a link of theirs has a join code: Share ▸ Start sharing) and joins them by a team list link, `figcraft://join?server=…&file=…&team=1`, which carries no secret: the server hands out the file's team link while it stays shared (`ResolveTeam`, answered like a join code), and Stop sharing revokes it with the others, disconnecting whoever joined with it. Turning the list off or changing its access refuses the old team links at once. `collab.files {refresh}` asks the server in Preferences for its list (`Admin::Team`; not the public server FigCraft ships with), and Home ▸ Online shows it below your own files: a row joins its file, its menu copies the link for the app or, with the server's `web_url`, for the browser build (`#join=…`). A browser build served beside its server names it with ``, the page's default for sharing, codes and the team list. No accounts: the list is for a server only your team reaches. Tests: `team_tests` (the server and apps over loopback), `code_tests` (links and the browser's driver), `online_tests` (the Online section), `query_tests` (the page's server). Percentages and hours unchanged | | 2026-10-30 | minor | M9, #1130: `figcraft-server` compares the SHA-256 digests of the presented and the kept server key instead of the raw strings, as `docs/server.md` says; the key file, `figcraft-server key` and rotation are unchanged. Numbers unchanged. | From 0ca48147e5b6cbb35181e042ac942f90c6120d40 Mon Sep 17 00:00:00 2001 From: fluid-chey Date: Wed, 4 Nov 2026 13:21:38 -0700 Subject: [PATCH 4/8] M9: a reconnect starts its pace afresh in the browser, as on the desktop Co-Authored-By: Claude Opus 5.5 --- crates/collab-net/src/driver.rs | 2 ++ crates/collab-net/src/driver_tests.rs | 18 ++++++++++++++++++ 2 files changed, 20 insertions(+) diff --git a/crates/collab-net/src/driver.rs b/crates/collab-net/src/driver.rs index 622b6c52d..a570836d3 100644 --- a/crates/collab-net/src/driver.rs +++ b/crates/collab-net/src/driver.rs @@ -165,6 +165,8 @@ impl Driver { fn connect(&mut self, now: Instant) { self.reconnect_at = None; + // A server counts per connection. + self.pace = Pace::new(now); self.conn = Some(Conn::new(now)); if let Err(e) = self.sock.open(&self.server().url()) { self.dropped(e, now); diff --git a/crates/collab-net/src/driver_tests.rs b/crates/collab-net/src/driver_tests.rs index 83684db7e..69cafb929 100644 --- a/crates/collab-net/src/driver_tests.rs +++ b/crates/collab-net/src/driver_tests.rs @@ -392,6 +392,24 @@ fn a_long_queue_goes_out_paced_and_in_order() { assert_eq!(d.deadline().unwrap() - now, ms(500), "nothing left to send but the reconnect"); } +#[test] +fn a_new_connection_starts_with_a_fresh_pace() { + let t0 = Instant::now(); + let (mut d, _inbox) = start(Script::default(), t0); + d.opened(t0); + d.message(&welcome(), t0); + for seq in 0..500 { + d.send(ClientMsg::Ack { seq }, t0); + } + d.closed(1006, "", t0); + let back = d.deadline().unwrap(); + d.tick(back); + d.opened(back); + d.message(&welcome(), back); + d.send(ClientMsg::Ack { seq: 900 }, back); + assert!(d.sock.sent.iter().any(|m| matches!(m, ClientMsg::Ack { seq: 900 })), "sent at once"); +} + #[test] fn leaving_sends_what_was_waiting_its_turn() { let t0 = Instant::now(); From c25088c58a9bbaf5ad6a47766868b14d615a7d5e Mon Sep 17 00:00:00 2001 From: fluid-chey Date: Wed, 4 Nov 2026 13:49:15 -0700 Subject: [PATCH 5/8] M9: count the pace of queued sends when they go, and say when leaving early The pace was set when a message was queued, so messages held up by a stalled connection left together afterwards and could trip the server's rate limit. One queue now counts it as they go, for the desktop and the browser alike. Leaving while changes are unconfirmed now says so: an undo waits for its edit, so what was still on its way may not all reach the others. Co-Authored-By: Claude Opus 5.5 --- crates/collab-net/src/driver.rs | 33 +++++++----------- crates/collab-net/src/driver_tests.rs | 16 +++++++++ crates/collab-net/src/guest.rs | 19 ++++------- crates/collab-net/src/throttle.rs | 47 ++++++++++++++++++++------ crates/engine/src/collab_live_tests.rs | 12 +++++++ crates/engine/src/sharing/live/join.rs | 9 +++-- 6 files changed, 92 insertions(+), 44 deletions(-) diff --git a/crates/collab-net/src/driver.rs b/crates/collab-net/src/driver.rs index a570836d3..049a60146 100644 --- a/crates/collab-net/src/driver.rs +++ b/crates/collab-net/src/driver.rs @@ -18,7 +18,6 @@ use figcraft_collab::message::{self, ClientMsg, Presence, ServerMsg}; use figcraft_engine::live::{Event, Status}; -use std::collections::VecDeque; use std::time::Duration; // `std::time::Instant` on native builds; the browser's clock in the web build. use web_time::Instant; @@ -29,7 +28,7 @@ use crate::invite::Endpoint; use crate::join::{GuestConfig, Next, Outcome, Resuming, Seen, Taken, after, ask, closed, hello, resolved, take, too_large, unresolved}; use crate::limits::NetLimits; use crate::requests::Publish; -use crate::throttle::{PRESENCE_INTERVAL, Pace, Throttle, earliest}; +use crate::throttle::{PRESENCE_INTERVAL, Paced, Throttle, earliest}; /// A WebSocket the driver opens, writes and closes; what it receives comes back through the /// driver's event methods. @@ -80,9 +79,8 @@ pub(crate) struct Driver { /// Your latest presence while not welcomed (sent with the welcome). pending: Option, throttle: Throttle, - /// Sync messages waiting their turn (a long queue from while offline), and when each may go. - pace: Pace, - paced: VecDeque<(Instant, ClientMsg)>, + /// Sync messages waiting their turn (a long queue from while offline). + paced: Paced, reconnect_at: Option, /// What this connection asks a server (a managing one), and what follows its answers. asking: Option<(Box, Then)>, @@ -101,8 +99,7 @@ impl Driver { resume: Resuming::default(), pending: None, throttle: Throttle::new(PRESENCE_INTERVAL), - pace: Pace::new(now), - paced: VecDeque::new(), + paced: Paced::new(now), reconnect_at: None, asking: None, done: false, @@ -165,8 +162,8 @@ impl Driver { fn connect(&mut self, now: Instant) { self.reconnect_at = None; - // A server counts per connection. - self.pace = Pace::new(now); + // A server counts per connection, and these were made for the last one. + self.paced = Paced::new(now); self.conn = Some(Conn::new(now)); if let Err(e) = self.sock.open(&self.server().url()) { self.dropped(e, now); @@ -311,19 +308,15 @@ impl Driver { if !self.conn.as_ref().is_some_and(|c| c.welcomed) { return; } - let wait = self.pace.wait(now); - if wait.is_zero() && self.paced.is_empty() { - if !self.write(&msg) { - self.end(Outcome::Retry("the connection dropped".into()), now); - } - } else { - self.paced.push_back((now + wait, msg)); + self.paced.push(msg); + if !self.send_paced(now) { + self.end(Outcome::Retry("the connection dropped".into()), now); } } /// Send the sync messages whose turn has come; false when one couldn't go. fn send_paced(&mut self, now: Instant) -> bool { - while let Some((_, msg)) = self.paced.pop_front_if(|(at, _)| *at <= now) { + while let Some(msg) = self.paced.take_due(now) { if !self.write(&msg) { return false; } @@ -342,7 +335,7 @@ impl Driver { /// Leave the session. pub(crate) fn leave(&mut self, now: Instant) { // What you did before leaving still reaches the others, whatever its turn. - while let Some((_, msg)) = self.paced.pop_front() { + while let Some(msg) = self.paced.take_now() { if !self.write(&msg) { break; } @@ -364,7 +357,7 @@ impl Driver { [Some(c.last_ping + l.keepalive), Some(c.last_heard + c.silence_limit(l)), if c.welcomed { self.throttle.due() } else { None }] } }); - earliest(conn.into_iter().flatten().chain([self.reconnect_at, self.paced.front().map(|(at, _)| *at)])) + earliest(conn.into_iter().flatten().chain([self.reconnect_at, self.paced.due()])) } /// Do what's due at `now`: reconnect, send held presence or a ping, give up on a connection @@ -426,7 +419,7 @@ impl Driver { } self.throttle = Throttle::new(PRESENCE_INTERVAL); // Made for this connection: the sync client sends again what the server didn't get. - self.paced.clear(); + self.paced = Paced::new(now); match after(outcome, &self.cfg, self.was_live, &mut self.attempt, &self.inbox) { Next::Stop(e) => { if let Some(e) = e { diff --git a/crates/collab-net/src/driver_tests.rs b/crates/collab-net/src/driver_tests.rs index 69cafb929..19c282615 100644 --- a/crates/collab-net/src/driver_tests.rs +++ b/crates/collab-net/src/driver_tests.rs @@ -392,6 +392,22 @@ fn a_long_queue_goes_out_paced_and_in_order() { assert_eq!(d.deadline().unwrap() - now, ms(500), "nothing left to send but the reconnect"); } +#[test] +fn messages_held_up_by_a_stall_dont_leave_together() { + let t0 = Instant::now(); + let (mut d, _inbox) = start(Script::default(), t0); + d.opened(t0); + d.message(&welcome(), t0); + for seq in 0..700 { + d.send(ClientMsg::Ack { seq }, t0); + } + let acks = |d: &Driver