M9: a slow join in a busy room isn't dropped for other people's cursors

A client taking in a large welcome holds one queue slot for the whole transfer,
while everyone's presence fills the rest. Five people moving their cursors fill
1024 slots in about 7 s, and the joiner is dropped with the same result on every
try. Presence now stays out of a queue that is half full: it is latest-wins, and
the other half stays free for edits, which can't be left out.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

crates/collab-net/src/hub.rs

@@ -737,6 +737,9 @@ impl<B: Backend> Hub<B> {
                         continue;
                     }
                     let Some(&conn) = room.ids.get(&id) else { continue };
+                    if matches!(msg, ServerMsg::Presence { .. }) && self.conns.get(&conn).is_some_and(|c| !c.out.takes_presence()) {
+                        continue;
+                    }
                     let Some(b) = bytes.get_or_insert_with(|| encode(&msg)).clone() else { continue };
                     if !deliver(&self.conns, conn, &b, &mut held, &mut stuck) {
                         stuck.push(conn);

crates/collab-net/src/hub/barred_tests.rs

@@ -74,6 +74,8 @@ fn a_kicked_connection_with_a_full_queue_is_shut() {
     let out = Line {
         tx,
         queued: Arc::default(),
+        waiting: Arc::default(),
+        max_presence: 32,
         block: Arc::new(AtomicUsize::new(0)),
         all: Arc::new(AtomicUsize::new(0)),
         max: 1 << 20,

crates/collab-net/src/hub/conn.rs

@@ -235,6 +235,8 @@ fn serve(mut stream: TcpStream, conn: u64, hub: &Sender<HubIn>, limits: NetLimit
     let out = Line {
         tx,
         queued: Arc::default(),
+        waiting: Arc::default(),
+        max_presence: limits.queue.max(4) / 2,
         block: block.queued.clone(),
         all: shared.queued.clone(),
         max: limits.max_queued_bytes,

crates/collab-net/src/hub/queue.rs

@@ -48,6 +48,7 @@ pub(crate) struct Charges {
 pub(super) struct Charge {
     n: usize,
     mine: Arc<AtomicUsize>,
+    waiting: Arc<AtomicUsize>,
     _all: Arc<Held>,
     _block: Option<Arc<Held>>,
 }
@@ -55,6 +56,7 @@ pub(super) struct Charge {
 impl Drop for Charge {
     fn drop(&mut self) {
         self.mine.fetch_sub(self.n, Ordering::Relaxed);
+        self.waiting.fetch_sub(1, Ordering::Relaxed);
     }
 }
 
@@ -76,6 +78,10 @@ pub(crate) enum Push {
 pub(crate) struct Line {
     pub(crate) tx: SyncSender<Out>,
     pub(crate) queued: Arc<AtomicUsize>,
+    /// Messages waiting in the queue.
+    pub(crate) waiting: Arc<AtomicUsize>,
+    /// With this many waiting, presence is left out (see [`Line::takes_presence`]).
+    pub(crate) max_presence: usize,
     pub(crate) block: Arc<AtomicUsize>,
     pub(crate) all: Arc<AtomicUsize>,
     pub(crate) max: usize,
@@ -84,6 +90,15 @@ pub(crate) struct Line {
 }
 
 impl Line {
+    /// Whether presence goes in the queue now. A client still taking in a long message (the welcome
+    /// of a large file on a slow link) would otherwise have its queue filled by everyone's cursors
+    /// and be dropped, again on every try. Presence is latest-wins, so what's left out is made up
+    /// for by the next update, and this keeps room for what can't be: other people's edits. A client
+    /// that reads nothing at all still ends at the write timeout.
+    pub(crate) fn takes_presence(&self) -> bool {
+        self.waiting.load(Ordering::Relaxed) < self.max_presence
+    }
+
     /// Queue `b`. One message always fits an empty queue (a welcome as large as the largest file
     /// gets out). `charges` is what the message counts against, shared by its recipients;
     /// `over_budget` lets it past the server's budget (the hub just dropped the biggest consumer
@@ -118,7 +133,8 @@ impl Line {
             }
         };
         self.queued.fetch_add(n, Ordering::Relaxed);
-        let charge = Charge { n, mine: self.queued.clone(), _all: all, _block: block };
+        self.waiting.fetch_add(1, Ordering::Relaxed);
+        let charge = Charge { n, mine: self.queued.clone(), waiting: self.waiting.clone(), _all: all, _block: block };
         self.tx.try_send(Out::Bytes(b.clone(), Some(charge))).map_err(|e| match e {
             TrySendError::Full(_) => Push::Full,
             TrySendError::Disconnected(_) => Push::Gone,

crates/collab-net/src/hub/queue_tests.rs

@@ -1,7 +1,8 @@
 //! The hub's send queues (no writers: nothing is ever sent, so what is queued stays queued): a
 //! broadcast counts once against the server's budget however many copies are queued, a block of
 //! addresses can't take more than its share, and when the server's budget is used up the
-//! connection holding the most is shed, not the next client that needs a little.
+//! connection holding the most is shed, not the next client that needs a little. Presence stays out
+//! of a busy queue.
 
 use std::collections::HashMap;
 use std::net::{IpAddr, TcpListener, TcpStream};
@@ -16,7 +17,20 @@ use super::{BlockUse, Blocks, Charges, Conn, Kind, Line, Out, Push, deliver};
 /// A line with its own queue counter, the given block counter and the server's.
 fn line(block: &Arc<AtomicUsize>, all: &Arc<AtomicUsize>, max_block: usize, max_all: usize) -> (Line, Receiver<Out>) {
     let (tx, rx) = sync_channel(64);
-    (Line { tx, queued: Arc::default(), block: block.clone(), all: all.clone(), max: 1 << 20, max_block, max_all }, rx)
+    (
+        Line {
+            tx,
+            queued: Arc::default(),
+            waiting: Arc::default(),
+            max_presence: 32,
+            block: block.clone(),
+            all: all.clone(),
+            max: 1 << 20,
+            max_block,
+            max_all,
+        },
+        rx,
+    )
 }
 
 fn socket() -> TcpStream {
@@ -120,3 +134,25 @@ fn blocks_nobody_uses_are_forgotten() {
     let _c = blocks.get("203.0.113.3".parse().unwrap());
     assert_eq!(blocks.len(), 2, "203.0.113.1 went");
 }
+
+#[test]
+fn a_busy_queue_leaves_presence_out_and_keeps_room_for_edits() {
+    let (l, rx) = line(&Arc::default(), &Arc::default(), usize::MAX, usize::MAX);
+    let msg = Bytes::from_static(b"x");
+    for _ in 0..32 {
+        assert!(l.takes_presence());
+        l.push(&msg, &mut Charges::default(), false).unwrap();
+    }
+    assert!(!l.takes_presence(), "half full");
+    // Edits still fit in the half kept for them.
+    for _ in 0..32 {
+        l.push(&msg, &mut Charges::default(), false).unwrap();
+    }
+    assert_eq!(l.push(&msg, &mut Charges::default(), false), Err(Push::Full), "a client that takes in nothing still ends");
+    // It catches up: presence goes in again.
+    for _ in 0..33 {
+        drop(rx.recv().unwrap());
+    }
+    assert!(l.takes_presence());
+    assert_eq!(l.waiting.load(Ordering::Relaxed), 31);
+}