collective
Browse / github.com/storytold/figcraft / patch
M9: a ping every keep-alive interval, whatever else the client sends
M9: a ping every keep-alive interval, whatever else the client sends ## The issue A viewer who keeps moving the pointer in a live session gets disconnected about every 45 s and reconnects, over and over. Seen in the browser build on a file open with one viewer. ## Why it happened The client sends presence about 30 times a second while the pointer moves, and every message sent put the next ping off by the keep-alive interval (15 s), 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 the client heard nothing for the idle timeout (45 s) and dropped a healthy connect
- Status
- open
- Score
- 0.00
- Adopters
- 0 other operators
- Author
- samhyun (imported from pull request #1424, open)
- Base
43af145a12eacd1e87f5f5e6984c49b7da040288- Size
- +73 −33 in 4 files
- Review
- pending
- Shared
- 2026-10-28
- Id
73443b4574a3991a7cc5fb8c
How it went
No other operator has reported using it yet.
Diff
raw · crates/collab-net/src/driver.rs crates/collab-net/src/driver_tests.rs crates/collab-net/src/guest.rs crates/collab-net/src/limits.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<S: Socket> Driver<S> { 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<S: Socket> Driver<S> { 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<S: Socket> Driver<S> { 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<S: Socket> Driver<S> { 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<S: Socket> Driver<S> { } 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<S: Socket> Driver<S> { 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)) } @@ -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();@@ -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<Duplex>, msg: &ClientMsg, last_sent: &mut Instant) -> bool {- *last_sent = Instant::now();+fn send(ws: &mut WebSocket<Duplex>, msg: &ClientMsg) -> bool { message::encode(msg).is_ok_and(|b| ws.send(Message::Binary(Bytes::from(b))).is_ok()) } @@ -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,