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 <noreply@anthropic.com>

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<S: Socket> {
     /// Your latest presence while not welcomed (sent with the welcome).
     pending: Option<Presence>,
     throttle: Throttle<Presence>,
-    /// 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<ClientMsg>,
     reconnect_at: Option<Instant>,
     /// What this connection asks a server (a managing one), and what follows its answers.
     asking: Option<(Box<Exchange>, Then)>,
@@ -101,8 +99,7 @@ impl<S: Socket> Driver<S> {
             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<S: Socket> Driver<S> {
 
     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<S: Socket> Driver<S> {
         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<S: Socket> Driver<S> {
     /// 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<S: Socket> Driver<S> {
                 [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<S: Socket> Driver<S> {
         }
         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 {

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<Script>| d.sock.sent.iter().filter(|m| matches!(m, ClientMsg::Ack { .. })).count();
+    let first = acks(&d);
+    // Nothing ran for 8 s.
+    d.tick(t0 + Duration::from_secs(8));
+    assert!(acks(&d) - first <= 200, "{} at once: a server drops a connection past 400", acks(&d) - first);
+}
+
 #[test]
 fn a_new_connection_starts_with_a_fresh_pace() {
     let t0 = Instant::now();

crates/collab-net/src/guest.rs

@@ -9,7 +9,6 @@
 //! 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;
@@ -32,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, Pace, Throttle, earliest};
+use crate::throttle::{PRESENCE_INTERVAL, Paced, Throttle, earliest};
 
 /// A joined (or joining) session. Dropping it leaves.
 pub struct Guest {
@@ -280,19 +279,15 @@ 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());
-    // Sync messages waiting their turn (a long queue from while offline), and when each may go.
-    let mut paced: VecDeque<(Instant, ClientMsg)> = VecDeque::new();
+    // Sync messages waiting their turn (a long queue from while offline).
+    let mut paced: Paced<ClientMsg> = Paced::new(Instant::now());
     let outcome = if send(&mut writer, &hello) {
         loop {
             let ping_at = last_ping + cfg.limits.keepalive;
-            let wake = earliest([throttle.due(), Some(ping_at), paced.front().map(|(at, _)| *at)]).unwrap_or(ping_at);
+            let wake = earliest([throttle.due(), Some(ping_at), paced.due()]).unwrap_or(ping_at);
             let mut out = None;
             match rx.recv_timeout(wake.saturating_duration_since(Instant::now())) {
-                Ok(Cmd::Send(m)) if welcomed => {
-                    let now = Instant::now();
-                    paced.push_back((now + pace.wait(now), m));
-                }
+                Ok(Cmd::Send(m)) if welcomed => paced.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(_)) => {}
@@ -307,7 +302,7 @@ 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)) {}
+                    while paced.take_now().is_some_and(|m| send(&mut writer, &m)) {}
                     let _ = writer.close(Some(CloseFrame { code: CloseCode::Normal, reason: "".into() }));
                     break Outcome::Left;
                 }
@@ -317,7 +312,7 @@ fn session(
             }
             let now = Instant::now();
             let mut sent = true;
-            while sent && let Some((_, m)) = paced.pop_front_if(|(at, _)| *at <= now) {
+            while sent && let Some(m) = paced.take_due(now) {
                 sent = send(&mut writer, &m);
             }
             if !sent {

crates/collab-net/src/throttle.rs

@@ -2,6 +2,7 @@
 //! (replacing any older waiting value) until the interval has passed, so the last one is never
 //! lost. Used for presence on both ends (about 30 updates a second).
 
+use std::collections::VecDeque;
 use std::time::Duration;
 
 // `std::time::Instant` on native builds; the browser's clock in the web build.
@@ -15,25 +16,51 @@ pub const PRESENCE_INTERVAL: Duration = Duration::from_millis(33);
 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.
+/// Messages that must all go out, in order, at a pace (everything queued while offline leaves at
+/// once when the connection comes back). The pace is counted when they go, so messages held up by
+/// a stalled connection don't leave together afterwards.
 #[derive(Debug)]
-pub struct Pace {
+pub struct Paced<T> {
     tokens: f64,
     last: Instant,
+    queue: VecDeque<T>,
 }
 
-impl Pace {
+impl<T> Paced<T> {
     pub fn new(now: Instant) -> Self {
-        Self { tokens: PACE_BURST, last: now }
+        Self { tokens: PACE_BURST, last: now, queue: VecDeque::new() }
     }
 
-    /// 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();
+    pub fn push(&mut self, v: T) {
+        self.queue.push_back(v);
+    }
+
+    /// The next message, if its turn has come at `now`.
+    pub fn take_due(&mut self, now: Instant) -> Option<T> {
+        if self.queue.is_empty() {
+            return None;
+        }
+        self.tokens = (self.tokens + now.saturating_duration_since(self.last).as_secs_f64() * PACE_RATE).min(PACE_BURST);
         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) }
+        // Within rounding of a whole turn counts as one: the time `due` gave has come.
+        if self.tokens < 1.0 - 1e-6 {
+            return None;
+        }
+        self.tokens -= 1.0;
+        self.queue.pop_front()
+    }
+
+    /// When the next message's turn comes (`None`: there's none).
+    pub fn due(&self) -> Option<Instant> {
+        self.queue.front()?;
+        // Rounded up to whole microseconds, so a timer set for this always finds the turn come.
+        let wait = ((1.0 - self.tokens).max(0.0) / PACE_RATE * 1e6).ceil() as u64;
+        Some(self.last + Duration::from_micros(wait))
+    }
+
+    /// The next message, whatever its turn (leaving).
+    pub fn take_now(&mut self) -> Option<T> {
+        self.queue.pop_front()
     }
 }
 

crates/engine/src/collab_live_tests.rs

@@ -419,6 +419,18 @@ fn a_session_out_of_step_joins_again_by_itself() {
     assert_eq!(room.s(0).docs.len(), tabs);
 }
 
+#[test]
+fn leaving_with_changes_unconfirmed_says_so() {
+    let mut room = Room::new(base(), 1, 7);
+    let (_, r) = frame_and_rect(room.server.document());
+    room.s(0).execute("collab.leave", &json!({})).unwrap();
+    assert_eq!(room.s(0).message.as_deref(), Some("You left the session; your copy stays open here"));
+    let mut room = Room::new(base(), 1, 7);
+    room.s(0).execute("node.set", &json!({"id": r, "props": {"name": "Unconfirmed"}})).unwrap();
+    room.s(0).execute("collab.leave", &json!({})).unwrap();
+    assert!(room.s(0).message.as_deref().is_some_and(|m| m.contains("before your last changes were confirmed")));
+}
+
 #[test]
 fn leaving_a_lost_session_makes_the_tab_a_plain_copy() {
     let mut room = Room::new(base(), 1, 7);

crates/engine/src/sharing/live/join.rs

@@ -42,8 +42,13 @@ fn leave(s: &mut Session, p: &Value) -> Result<Value> {
         None => s.live.shown(s.tab()),
     };
     let ended = id.and_then(|id| s.live.end(id));
-    if ended.as_ref().is_some_and(|l| l.syncing() && l.role() == Some(Role::Guest)) {
-        s.message = Some("You left the session; your copy stays open here".into());
+    if let Some(l) = ended.as_ref().filter(|l| l.syncing() && l.role() == Some(Role::Guest)) {
+        // What was still on its way may be held back for what's before it (an undo waits for its edit).
+        s.message = Some(if l.client().is_some_and(|c| c.pending() > 0) {
+            "You left before your last changes were confirmed: others may not see them all, and your copy stays open here".into()
+        } else {
+            "You left the session; your copy stays open here".into()
+        });
     }
     Ok(json!({"ended": ended.is_some_and(|l| l.status.active())}))
 }