diff --git a/crates/collab-net/src/inbox.rs b/crates/collab-net/src/inbox.rs index fb505a0a..1ec00e67 100644 --- a/crates/collab-net/src/inbox.rs +++ b/crates/collab-net/src/inbox.rs @@ -11,11 +11,18 @@ use figcraft_engine::live::{Event, Presence}; pub type Waker = Arc; /// Most discrete events kept between two drains (joins, leaves, status); the oldest go first. +/// Document sync events don't count and are never dropped: the sync client treats a missing +/// batch as a fatal sequence gap, so losing one would end the session. const MAX_EVENTS: usize = 10_000; #[derive(Default)] struct State { - events: VecDeque, + /// Arrival number of the next event, so [`Inbox::drain`] can merge the two queues in order. + next: u64, + /// [`Event::Sync`] events, never dropped. + syncs: VecDeque<(u64, Event)>, + /// Every other event, capped at [`MAX_EVENTS`]. + others: VecDeque<(u64, Event)>, presence: BTreeMap, } @@ -43,11 +50,17 @@ impl Inbox { if let Event::PeerLeft(id) = &e { s.presence.remove(id); } - if s.events.len() >= MAX_EVENTS { - log::warn!("live session: {MAX_EVENTS} events not taken; dropping the oldest"); - s.events.pop_front(); + let n = s.next; + s.next += 1; + if matches!(e, Event::Sync(_)) { + s.syncs.push_back((n, e)); + return; } - s.events.push_back(e); + if s.others.len() >= MAX_EVENTS { + log::warn!("live session: {MAX_EVENTS} events not taken; dropping the oldest"); + s.others.pop_front(); + } + s.others.push_back((n, e)); }); } @@ -62,8 +75,70 @@ impl Inbox { /// peer. pub fn drain(&self) -> Vec { let mut s = self.state.lock().unwrap_or_else(PoisonError::into_inner); - let mut out: Vec = s.events.drain(..).collect(); + let mut syncs = std::mem::take(&mut s.syncs).into_iter().peekable(); + let mut others = std::mem::take(&mut s.others).into_iter().peekable(); + let mut out = Vec::with_capacity(syncs.len() + others.len() + s.presence.len()); + loop { + let sync_first = match (syncs.peek(), others.peek()) { + (Some((a, _)), Some((b, _))) => a < b, + (Some(_), None) => true, + (None, Some(_)) => false, + (None, None) => break, + }; + let next = if sync_first { syncs.next() } else { others.next() }; + out.extend(next.map(|(_, e)| e)); + } out.extend(std::mem::take(&mut s.presence).into_iter().map(|(id, p)| Event::Presence(id, p))); out } } + +#[cfg(test)] +mod tests { + use super::*; + use figcraft_collab::message::ServerMsg; + + fn reject(batch_id: u64) -> Event { + Event::Sync(ServerMsg::Reject { batch_id, reason: String::new(), fixes: Vec::new() }) + } + + fn sync_ids(events: &[Event]) -> Vec { + events + .iter() + .filter_map(|e| match e { + Event::Sync(ServerMsg::Reject { batch_id, .. }) => Some(*batch_id), + _ => None, + }) + .collect() + } + + #[test] + fn a_full_inbox_never_drops_sync_events() { + let inbox = Inbox::new(Arc::new(|| {})); + inbox.push(reject(1)); + for _ in 0..MAX_EVENTS { + inbox.push(Event::PeerLeft(7)); + } + inbox.push(reject(2)); + for _ in 0..MAX_EVENTS { + inbox.push(Event::PeerLeft(7)); + } + let out = inbox.drain(); + assert_eq!(sync_ids(&out), vec![1, 2]); + assert!(out.len() <= MAX_EVENTS + 2, "{} events kept", out.len()); + } + + #[test] + fn sync_events_past_the_cap_are_all_kept_in_order() { + let inbox = Inbox::new(Arc::new(|| {})); + let n = MAX_EVENTS as u64 + 5; + for id in 0..n { + inbox.push(reject(id)); + } + inbox.push(Event::PeerLeft(7)); + let out = inbox.drain(); + let ids = sync_ids(&out); + assert_eq!(ids.len() as u64, n, "Sync events dropped"); + assert!(ids.iter().copied().eq(0..n), "Sync events out of order"); + } +}