collab: refuse sequence numbers a host cannot advance

The welcome's seq and processed.last_batch are the host's word and were used
unchecked: accepted computed self.seq + 1 and queue bumped next_batch, so a
host sending u64::MAX overflowed the client (panic in debug, wrap in release).
Both counters now step with checked_add/saturating_add, and a sequence that
cannot advance is refused as a protocol error (the established rejoin path).

Test: hostile_sequence_numbers_do_not_overflow_the_client.

crates/collab/src/client.rs

@@ -307,7 +307,8 @@ impl Client {
     /// Queue `changes` as a new batch (already shown in the view).
     fn queue(&mut self, changes: Vec<Change>, kind: BatchKind, step: Option<u64>) {
         let batch_id = self.next_batch;
-        self.next_batch += 1;
+        // `next_batch` starts from the welcome's `last_batch`, the host's word: saturate.
+        self.next_batch = self.next_batch.saturating_add(1);
         self.pending.push_back(Pending { batch_id, base_seq: self.seq, changes, kind, step, rebased: false, sent: false });
     }
 
@@ -484,7 +485,11 @@ impl Client {
 
     /// A batch the server accepted (in order). Follow-ups it causes are queued.
     fn accepted(&mut self, acc: Accepted, recompute: &dyn Recompute) -> Result<(), Error> {
-        if acc.seq != self.seq + 1 {
+        // The host's sequence number is untrusted: at `u64::MAX` it can't step past it.
+        let Some(next) = self.seq.checked_add(1) else {
+            return Err(Error::Protocol(format!("sequence {} after {}", acc.seq, self.seq)));
+        };
+        if acc.seq != next {
             return Err(Error::Protocol(format!("sequence {} after {}", acc.seq, self.seq)));
         }
         if acc.author != self.id {

crates/collab/src/client_tests.rs

@@ -0,0 +1,43 @@
+//! A client's handling of the host's messages: sequence numbers and batch ids come from the host,
+//! so they are treated as hostile input (they must never overflow).
+
+use figcraft_doc::NodeKind;
+
+use crate::message::{Accepted, ClientMsg, Hello, ServerMsg};
+use crate::sim_tests::base;
+use crate::{Client, Limits, NoRecompute, Outgoing, Server};
+
+fn hello() -> Hello {
+    let ClientMsg::Hello(h) = Client::hello("f", "tok", "Ada", "#FF8800") else { panic!("not a hello") };
+    h
+}
+
+/// A welcome whose sequence numbers are what a lying host would send.
+fn hostile_welcome(seq: u64, last_batch: u64) -> ServerMsg {
+    let mut s = Server::new(base(), "tok", Limits::default());
+    let (_, out) = s.join(hello()).unwrap();
+    let mut welcome = out
+        .into_iter()
+        .find_map(|o| match o {
+            Outgoing::To(_, m) => Some(m),
+            _ => None,
+        })
+        .unwrap();
+    let ServerMsg::Welcome { seq: seq0, processed, .. } = &mut welcome else { panic!("no welcome") };
+    *seq0 = seq;
+    processed.last_batch = last_batch;
+    welcome
+}
+
+#[test]
+fn hostile_sequence_numbers_do_not_overflow_the_client() {
+    let mut c = Client::join(hostile_welcome(u64::MAX, u64::MAX)).unwrap();
+    // This client's own first batch must not overflow its batch counter.
+    let mut d = c.document().clone();
+    let page = d.first_page().unwrap();
+    d.create(page, NodeKind::Rectangle, "R", 0.0, 0.0, 10.0, 10.0).unwrap();
+    assert!(c.edit(&d).unwrap().is_some(), "the edit is batched");
+    // A batch the host claims follows `u64::MAX` can't exist: refuse it, don't overflow.
+    let acc = Accepted { seq: 0, author: c.id().wrapping_add(1), batch_id: 0, changes: Vec::new() };
+    assert!(c.receive(ServerMsg::Changes(acc), &NoRecompute).is_err());
+}

crates/collab/src/lib.rs

@@ -123,6 +123,8 @@ impl Recompute for NoRecompute {
     fn recompute(&self, _: &mut Document, _: &Document, _: &BTreeSet<NodeId>) {}
 }
 
+#[cfg(test)]
+mod client_tests;
 #[cfg(test)]
 mod codec_tests;
 #[cfg(test)]