From 6b73c590f8be91e2364b98e806edef106f241aab Mon Sep 17 00:00:00 2001 From: samhyun Date: Tue, 27 Oct 2026 19:43:45 +0900 Subject: [PATCH] M9: a ping every keep-alive interval, whatever else the client sends A viewer moving the pointer sends presence about 30 times a second, and every message sent put the next ping off by the keep-alive interval, so no ping went out while presence did. The host answers pings and nothing else to a client alone in a session (presence goes to the other peers), so nothing was heard for the idle timeout (45 s): the client dropped a healthy connection and reconnected, over and over. Seen in the browser build; the desktop app's guest has the same timing (its reader times out the same way) and gets the same change. Pings are now timed from the last ping, not from the last message sent: one goes out every 15 s whatever else is on the wire. Regression test: 90 s of presence at 10 a second with the host answering only the pings keeps the connection (it closed before the change). Co-Authored-By: Claude Fable 5.1 --- crates/collab-net/src/driver.rs | 41 +++++++++++++++------------ crates/collab-net/src/driver_tests.rs | 36 +++++++++++++++++++++-- crates/collab-net/src/guest.rs | 27 ++++++++++-------- crates/collab-net/src/limits.rs | 2 +- 4 files changed, 73 insertions(+), 33 deletions(-) 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,