Merge branch 'fix/live-reconnect' into fix/live-offline-merge

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

@@ -485,6 +485,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())}))
 }