diff --git a/crates/collab-net/src/driver.rs b/crates/collab-net/src/driver.rs index 4ebbec15..84801c4d 100644 --- a/crates/collab-net/src/driver.rs +++ b/crates/collab-net/src/driver.rs @@ -2,9 +2,10 @@ //! target: a [`Driver`] reacts to socket events and to the clock the way the native guest's //! threads do — hello on open (resuming as the same client after a drop), presence held until the //! welcome and changes made for an earlier connection dropped (the sync client sends again what -//! the server didn't get), presence throttled to about 30 a second (latest wins), a ping when quiet, a connection that goes silent counted as -//! dropped, and reconnects with growing pauses after a live connection drops. The browser layer -//! (`web.rs`) feeds it the WebSocket's events and wakes it at [`Driver::deadline`]. +//! the server didn't get), presence throttled to about 30 a second (latest wins), a ping every +//! keep-alive interval, a connection that goes silent counted as dropped, and reconnects with +//! growing pauses after a live connection drops. The browser layer (`web.rs`) feeds it the +//! WebSocket's events and wakes it at [`Driver::deadline`]. //! //! A join by code ([`GuestConfig::code`]) first asks the server which file the code opens on a //! connection of its own, then joins that file as with a link; a server that doesn't know the @@ -36,13 +37,15 @@ struct Conn { opened: bool, welcomed: bool, seen: Seen, - last_sent: Instant, + /// The last ping (or the connection's start): one goes out every keep-alive interval, whatever + /// else is sent, since presence gets no answer and only what the host sends counts as heard. + last_ping: Instant, last_heard: Instant, } impl Conn { fn new(now: Instant) -> Self { - Self { started: now, opened: false, welcomed: false, seen: Seen::default(), last_sent: now, last_heard: now } + Self { started: now, opened: false, welcomed: false, seen: Seen::default(), last_ping: now, last_heard: now } } } @@ -106,7 +109,7 @@ impl Driver { Some(code) => ClientMsg::Resolve(Resolve { protocol: PROTOCOL, client_version: message::client_version(), code: code.clone() }), None => hello(&self.cfg, &self.resume), }; - if !self.write(&hello, now) { + if !self.write(&hello) { self.end(Outcome::Retry("the connection dropped".into()), now); } } @@ -173,7 +176,7 @@ impl Driver { return; } if let Some(p) = self.throttle.offer(p, now) - && !self.write(&ClientMsg::Presence(p), now) + && !self.write(&ClientMsg::Presence(p)) { self.end(Outcome::Retry("the connection dropped".into()), now); } @@ -186,7 +189,7 @@ impl Driver { if !self.conn.as_ref().is_some_and(|c| c.welcomed) { return; } - if !self.write(&msg, now) { + if !self.write(&msg) { self.end(Outcome::Retry("the connection dropped".into()), now); } } @@ -204,7 +207,7 @@ impl Driver { let l = &self.cfg.limits; let conn = self.conn.as_ref().map(|c| { if c.opened { - [Some(c.last_sent + l.keepalive), Some(c.last_heard + l.idle_timeout), if c.welcomed { self.throttle.due() } else { None }] + [Some(c.last_ping + l.keepalive), Some(c.last_heard + l.idle_timeout), if c.welcomed { self.throttle.due() } else { None }] } else { [Some(c.started + self.cfg.connect_timeout), None, None] } @@ -223,7 +226,7 @@ impl Driver { } let l = self.cfg.limits; let Some(c) = self.conn.as_ref() else { return }; - let (opened, started, last_heard, last_sent) = (c.opened, c.started, c.last_heard, c.last_sent); + let (opened, started, last_heard, last_ping) = (c.opened, c.started, c.last_heard, c.last_ping); if !opened { if now >= started + self.cfg.connect_timeout { self.end(Outcome::Retry(format!("no answer from {}", self.cfg.invite.server.display())), now); @@ -233,21 +236,23 @@ impl Driver { if now >= last_heard + l.idle_timeout { return self.end(Outcome::Retry("the host stopped answering".into()), now); } - let ok = match self.throttle.take_due(now) { - Some(p) => self.write(&ClientMsg::Presence(p), now), - None if now >= last_sent + l.keepalive => self.write(&ClientMsg::Ping, now), - None => true, - }; + let ok = 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); } } - /// Send one message; false when it couldn't go. - fn write(&mut self, msg: &ClientMsg, now: Instant) -> bool { + /// Send a ping; false when it couldn't go. + fn ping(&mut self, now: Instant) -> bool { if let Some(c) = self.conn.as_mut() { - c.last_sent = now; + c.last_ping = now; } + self.write(&ClientMsg::Ping) + } + + /// Send one message; false when it couldn't go. + fn write(&mut self, msg: &ClientMsg) -> bool { message::encode(msg).is_ok_and(|b| self.sock.send(&b)) } diff --git a/crates/collab-net/src/driver_tests.rs b/crates/collab-net/src/driver_tests.rs index 5e395b98..a94ff064 100644 --- a/crates/collab-net/src/driver_tests.rs +++ b/crates/collab-net/src/driver_tests.rs @@ -142,9 +142,10 @@ fn presence_is_throttled_and_quiet_connections_ping() { d.tick(due); assert_eq!(d.sock.sent.last(), Some(&ClientMsg::Presence(cursor(3.0)))); let n = d.sock.sent.len(); - // Quiet: a ping after the keep-alive interval (the host answers, which counts as traffic). + // A ping every keep-alive interval from the connection's start, presence or not (the host + // answers, which counts as traffic). let ping_at = d.deadline().unwrap(); - assert_eq!(ping_at, due + Duration::from_secs(15)); + assert_eq!(ping_at, t0 + Duration::from_secs(15)); d.tick(ping_at - ms(1)); assert_eq!(d.sock.sent.len(), n, "not yet"); d.message(&encode(&ServerMsg::Pong), ping_at - ms(1)); @@ -152,6 +153,37 @@ fn presence_is_throttled_and_quiet_connections_ping() { assert_eq!(d.sock.sent.last(), Some(&ClientMsg::Ping)); } +/// A viewer moving the pointer sends presence all the time, which the host doesn't answer: the +/// pings still go out, and the host's pongs keep the connection (before, each presence put the +/// next ping off, nothing was heard for the idle timeout, and the client dropped the connection). +#[test] +fn pings_go_out_while_presence_does_and_the_pongs_keep_the_connection() { + let t0 = Instant::now(); + let (mut d, inbox) = start(Script::default(), t0); + d.opened(t0); + d.message(&welcome(), t0); + inbox.drain(); + let mut answered = 0; + for i in 1..=900u32 { + let now = t0 + ms(100 * u64::from(i)); + d.presence(cursor(f64::from(i)), now); + while let Some(due) = d.deadline().filter(|t| *t <= now) { + d.tick(due); + } + // The host answers every ping (and nothing else) at once. + let pinged = d.sock.sent[answered..].contains(&ClientMsg::Ping); + answered = d.sock.sent.len(); + if pinged { + d.message(&encode(&ServerMsg::Pong), now); + } + } + assert!(!d.done()); + assert_eq!(d.sock.closes, 0, "90 s of presence and pongs: still connected"); + assert!(statuses(&inbox.drain()).is_empty()); + let pings = d.sock.sent.iter().filter(|m| **m == ClientMsg::Ping).count(); + assert!(pings >= 5, "a ping every 15 s ({pings})"); +} + #[test] fn a_silent_host_counts_as_dropped_and_the_client_reconnects() { let t0 = Instant::now(); diff --git a/crates/collab-net/src/guest.rs b/crates/collab-net/src/guest.rs index 2e1d5046..91e72cf6 100644 --- a/crates/collab-net/src/guest.rs +++ b/crates/collab-net/src/guest.rs @@ -1,7 +1,8 @@ //! Joining a session: the client end of the transport, with reconnect. //! //! A supervisor thread connects, says hello, then writes (presence throttled to about 30 a -//! second, latest wins; a ping when quiet) while a reader thread takes in what the host sends. +//! second, latest wins; a ping every keep-alive interval) while a reader thread takes in what the +//! host sends. //! When the connection drops after having been live, it reconnects with growing pauses; a //! refusal or the host ending the session is final. //! @@ -280,13 +281,15 @@ fn session( } } }; - let mut last_sent = Instant::now(); + // A ping every keep-alive interval, whatever else is sent: presence gets no answer, and the + // reader counts only what the host sends. + let mut last_ping = Instant::now(); let hello = hello(cfg, resume); let mut welcomed = false; let mut throttle = Throttle::new(PRESENCE_INTERVAL); - let outcome = if send(&mut writer, &hello, &mut last_sent) { + let outcome = if send(&mut writer, &hello) { loop { - let ping_at = last_sent + cfg.limits.keepalive; + let ping_at = last_ping + cfg.limits.keepalive; let wake = earliest([throttle.due(), Some(ping_at)]).unwrap_or(ping_at); let mut out = None; let mut now_send = Vec::new(); @@ -312,16 +315,17 @@ fn session( Ok(Cmd::Welcomed(_) | Cmd::Lost(..)) => {} Err(RecvTimeoutError::Timeout) => {} } - if !now_send.iter().all(|m| send(&mut writer, m, &mut last_sent)) { + if !now_send.iter().all(|m| send(&mut writer, m)) { break Outcome::Retry("the connection dropped".into()); } let now = Instant::now(); let out = out.or_else(|| throttle.take_due(now)); - let ok = match out { - Some(p) => send(&mut writer, &ClientMsg::Presence(p), &mut last_sent), - None if now >= last_sent + cfg.limits.keepalive => send(&mut writer, &ClientMsg::Ping, &mut last_sent), - None => true, - }; + let ok = out.is_none_or(|p| send(&mut writer, &ClientMsg::Presence(p))); + let ok = ok + && (now < last_ping + cfg.limits.keepalive || { + last_ping = now; + send(&mut writer, &ClientMsg::Ping) + }); if !ok { break Outcome::Retry("the connection dropped".into()); } @@ -335,8 +339,7 @@ fn session( } /// Send one message; false when the connection is gone. -fn send(ws: &mut WebSocket, msg: &ClientMsg, last_sent: &mut Instant) -> bool { - *last_sent = Instant::now(); +fn send(ws: &mut WebSocket, msg: &ClientMsg) -> bool { message::encode(msg).is_ok_and(|b| ws.send(Message::Binary(Bytes::from(b))).is_ok()) } diff --git a/crates/collab-net/src/limits.rs b/crates/collab-net/src/limits.rs index 0321b36c..301c4e78 100644 --- a/crates/collab-net/src/limits.rs +++ b/crates/collab-net/src/limits.rs @@ -30,7 +30,7 @@ pub struct NetLimits { /// answer, so this is longer than `idle_timeout` (a large project on a slow connection can /// easily need several minutes). pub request_reply_timeout: Duration, - /// A quiet client pings this often (keeps it under `idle_timeout`). + /// A client pings this often, whatever else it sends (keeps it under `idle_timeout`). pub keepalive: Duration, /// Messages queued for one client before it counts as stuck and is disconnected. pub queue: usize,