acpmux: ACP session/list goes in pages when the whole list is too large to send

cmux-tui/crates/acpmux/src/hub/list_guard.rs

@@ -94,7 +94,33 @@ pub(crate) struct Snapshot {
     made: StdMutex<HashMap<&'static str, Arc<String>>>,
     /// The full rows are built only for trusted internal consumers (peer bootstrap and ACP).
     /// Ordinary census/watch replies keep using `whole`, so their retained bytes stay small.
-    full: StdMutex<Option<Arc<String>>>,
+    full: StdMutex<Option<Arc<FullRows>>>,
+}
+
+/// Every summary with its tools, serialized as [`Snapshot::whole`] is, with where each row is
+/// and its paging keys: a page of it parses only its own rows.
+pub(crate) struct FullRows {
+    whole: Arc<String>,
+    spans: Vec<Range<usize>>,
+    keys: Vec<Value>,
+}
+
+impl FullRows {
+    /// Each row's `sessionId` and `peer`, in the order of [`Self::row`].
+    pub(crate) fn keys(&self) -> &[Value] {
+        &self.keys
+    }
+
+    /// The bytes of every row together.
+    pub(crate) fn bytes(&self) -> usize {
+        self.whole.len()
+    }
+
+    /// Row `at`, parsed.
+    pub(crate) fn row(&self, at: usize) -> Value {
+        let text = self.spans.get(at).and_then(|span| self.whole.get(span.clone()));
+        text.and_then(|text| serde_json::from_str(text).ok()).unwrap_or_default()
+    }
 }
 
 type Built = (String, Vec<Range<usize>>, Vec<Value>);
@@ -124,11 +150,17 @@ impl Snapshot {
     }
 
     pub(crate) fn full_whole(&self, hub: &Hub) -> Arc<String> {
+        self.full_rows(hub).whole.clone()
+    }
+
+    /// [`Self::full_whole`] with where each row is, for ACP's pages.
+    pub(crate) fn full_rows(&self, hub: &Hub) -> Arc<FullRows> {
         let mut full = self.full.lock().unwrap_or_else(|e| e.into_inner());
         full.get_or_insert_with(|| {
             let mut rows = Rows::default();
             hub.visit_session_summaries(true, |row| rows.push(&row));
-            Arc::new(rows.finish().0)
+            let (whole, spans, keys) = rows.finish();
+            Arc::new(FullRows { whole: Arc::new(whole), spans, keys })
         })
         .clone()
     }
@@ -222,11 +254,15 @@ impl Rows {
     }
 
     fn finish(mut self) -> Built {
-        // Stable ties keep the original local-before-remote order, including duplicate IDs.
+        // The order `sort_session_summaries` gives: a chat's copies local first, then by peer,
+        // the order their cursors (`session_pages.rs`) read them in.
         self.keys.sort_by(|a, b| {
-            b.updated_at.cmp(&a.updated_at).then_with(|| {
-                a.session_id.as_deref().unwrap_or("").cmp(b.session_id.as_deref().unwrap_or(""))
-            })
+            b.updated_at
+                .cmp(&a.updated_at)
+                .then_with(|| {
+                    a.session_id.as_deref().unwrap_or("").cmp(b.session_id.as_deref().unwrap_or(""))
+                })
+                .then_with(|| a.peer.cmp(&b.peer))
         });
         let prefix = br#"{"sessions":["#;
         let mut whole = Vec::with_capacity(
@@ -440,6 +476,7 @@ mod tests {
             json!({"sessionId": "z", "updatedAt": 1, "status": "closed", "tools": [{"id": "tool", "description": "雪\n\""}]}),
             json!({"sessionId": "same", "updatedAt": 9, "origin": "local"}),
             json!({"sessionId": "same", "updatedAt": 9, "peer": "remote", "origin": "remote"}),
+            json!({"sessionId": "same", "updatedAt": 9, "peer": "other", "origin": "other"}),
             json!({"sessionId": "a", "updatedAt": 9, "unknown": {"array": [null, false, 42]}}),
             json!({"sessionId": null, "updatedAt": "invalid", "status": "closed"}),
             json!({"updatedAt": -1, "opaque": "preserved"}),
@@ -460,7 +497,8 @@ mod tests {
             );
         }
         assert_eq!(rows[1]["origin"], "local");
-        assert_eq!(rows[2]["origin"], "remote");
+        assert_eq!(rows[2]["origin"], "other");
+        assert_eq!(rows[3]["origin"], "remote");
     }
 
     #[test]

cmux-tui/crates/acpmux/src/hub/session_pages.rs

@@ -20,11 +20,15 @@
 
 use super::*;
 
-/// Sorts session summaries newest first, ties by `sessionId` ascending. The
-/// summaries are snapshots, so the order is total while sorting.
+/// Sorts session summaries newest first, ties by `sessionId` ascending, then by `peer` (this
+/// daemon's own first): a chat here and its copy on a peer keep one order. The summaries are
+/// snapshots, so the order is total while sorting.
 pub(crate) fn sort_session_summaries(list: &mut [Value]) {
     list.sort_by(|a, b| {
-        updated_ms(b).cmp(&updated_ms(a)).then_with(|| session_id(a).cmp(session_id(b)))
+        updated_ms(b)
+            .cmp(&updated_ms(a))
+            .then_with(|| session_id(a).cmp(session_id(b)))
+            .then_with(|| peer(a).cmp(&peer(b)))
     });
 }
 
@@ -95,6 +99,58 @@ pub(crate) fn session_list_reply(sorted: Vec<Value>, params: &Value) -> Result<V
     Ok(page(sessions, Some(next.map_or(Value::Null, Value::String))))
 }
 
+/// A page cut by bytes rather than by count, for a reply that has no `limit` to page by (ACP's
+/// `session/list`): of the rows whose paging `keys` are given, those after `cursor` (a
+/// `nextCursor` of this list) that `row` makes and whose sizes fit in `budget`, at least one,
+/// and the cursor after the last of them when rows are left. Only the page's rows are made.
+///
+/// Pages go in the order of [`page_key`], a hash of each row's `sessionId` and `peer`, not
+/// newest first, and the cursor is that hash, 33 bytes whatever the ids: a session that
+/// updates or leaves while a client pages moves no other, so none is skipped or repeated. A
+/// row listed twice (one id, one peer) comes once. A page never ends between rows of one hash,
+/// which two different rows share only if 128-bit hashes collide.
+pub(crate) fn byte_page(
+    keys: &[Value],
+    cursor: Option<&str>,
+    budget: usize,
+    row: impl Fn(usize) -> Value,
+) -> Result<(Vec<Value>, Option<String>), RpcError> {
+    let after = match cursor {
+        None => None,
+        Some(text) => Some(
+            text.strip_prefix('h')
+                .filter(|hex| hex.len() == 32)
+                .and_then(|hex| u128::from_str_radix(hex, 16).ok())
+                .ok_or_else(|| {
+                    RpcError::invalid_params(format!("cursor {text:?} is not a cursor"))
+                })?,
+        ),
+    };
+    let pair = |at: usize| (session_id(&keys[at]), peer(&keys[at]));
+    let mut order: Vec<(u128, usize)> = keys
+        .iter()
+        .enumerate()
+        .map(|(at, key)| (page_key(key), at))
+        .filter(|(hash, _)| after.is_none_or(|after| *hash > after))
+        .collect();
+    // Stable, so of a row listed twice the first is kept.
+    order.sort_by(|(a, at), (b, bt)| a.cmp(b).then_with(|| pair(*at).cmp(&pair(*bt))));
+    order.dedup_by(|(_, at), (_, bt)| pair(*at) == pair(*bt));
+    let (mut page, mut used, mut last) = (Vec::new(), 0, None);
+    for &(hash, at) in &order {
+        let value = row(at);
+        // One more byte for the comma between rows.
+        let bytes = value.to_string().len() + 1;
+        if last.is_some_and(|last| last != hash) && used + bytes > budget {
+            break;
+        }
+        (used, last) = (used + bytes, Some(hash));
+        page.push(value);
+    }
+    let next = (page.len() < order.len()).then(|| format!("h{:032x}", last.unwrap_or_default()));
+    Ok((page, next))
+}
+
 /// IDs are immutable: an unread session updating between pages cannot be lost.
 fn session_id_page(
     mut rows: Vec<Value>,
@@ -196,7 +252,7 @@ pub(crate) fn refuse_watch_filters(params: &Value) -> Result<(), RpcError> {
     Ok(())
 }
 
-/// After the session updated at `updated_ms` whose id hashes to `key`.
+/// After the session updated at `updated_ms` whose id and peer hash to `key` ([`row_key`]).
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 struct SessionCursor {
     updated_ms: u64,
@@ -205,7 +261,7 @@ struct SessionCursor {
 
 impl SessionCursor {
     fn after(session: &Value) -> Self {
-        Self { updated_ms: stamp(session), key: id_hash(session_id(session)) }
+        Self { updated_ms: stamp(session), key: row_key(session) }
     }
 
     /// `k<stamp hex>.<id hash hex>`: at most 30 characters (the app passes at most 32).
@@ -230,7 +286,7 @@ impl SessionCursor {
     fn start(self, sorted: &[Value]) -> usize {
         sorted
             .iter()
-            .position(|s| stamp(s) == self.updated_ms && id_hash(session_id(s)) == self.key)
+            .position(|s| stamp(s) == self.updated_ms && row_key(s) == self.key)
             .map(|at| at + 1)
             .unwrap_or_else(|| {
                 sorted.iter().position(|s| stamp(s) <= self.updated_ms).unwrap_or(sorted.len())
@@ -255,17 +311,131 @@ fn session_id(session: &Value) -> &str {
     session.get("sessionId").and_then(Value::as_str).unwrap_or("")
 }
 
-/// FNV-1a of the session id: the same in every daemon run.
-fn id_hash(id: &str) -> u64 {
-    id.bytes().fold(0xcbf2_9ce4_8422_2325, |hash, byte| {
-        (hash ^ u64::from(byte)).wrapping_mul(0x0100_0000_01b3)
-    })
+fn peer(session: &Value) -> Option<&str> {
+    session.get("peer").and_then(Value::as_str)
+}
+
+/// Where a row stands in [`byte_page`]'s order: two FNV-1a hashes of its `sessionId` and
+/// `peer` ([`row_key`]'s bytes), the same in every daemon run.
+fn page_key(session: &Value) -> u128 {
+    let high = fnv(0xcbf2_9ce4_8422_2325, row_key_bytes(session));
+    let low = fnv(0x6c62_272e_07bb_0142, row_key_bytes(session));
+    (u128::from(high) << 64) | u128::from(low)
+}
+
+/// FNV-1a of the session id, then, for a peer's row, a NUL and the peer's name: the same in
+/// every daemon run, and this daemon's own rows keep the key they had before peers were in it.
+fn row_key(session: &Value) -> u64 {
+    fnv(0xcbf2_9ce4_8422_2325, row_key_bytes(session))
+}
+
+fn row_key_bytes(session: &Value) -> impl Iterator<Item = u8> + '_ {
+    let peer = peer(session).map(|p| std::iter::once(0).chain(p.bytes()));
+    session_id(session).bytes().chain(peer.into_iter().flatten())
+}
+
+fn fnv(basis: u64, bytes: impl Iterator<Item = u8>) -> u64 {
+    bytes.fold(basis, |hash, byte| (hash ^ u64::from(byte)).wrapping_mul(0x0100_0000_01b3))
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    fn name(row: &Value) -> String {
+        format!("{}{}", session_id(row), peer(row).map_or(String::new(), |p| format!("@{p}")))
+    }
+
+    /// Every page `budget` cuts, read from the first, with `between` run
+    /// on the rows (and the names read so far) after each page.
+    fn read_pages(
+        rows: &mut Vec<Value>,
+        budget: usize,
+        mut between: impl FnMut(&mut Vec<Value>, &[String]),
+    ) -> Vec<String> {
+        let (mut read, mut cursor) = (Vec::new(), None::<String>);
+        loop {
+            let (page, next) =
+                byte_page(rows, cursor.as_deref(), budget, |at| json!({"name": name(&rows[at])}))
+                    .unwrap();
+            assert!(!page.is_empty());
+            read.extend(page.iter().map(|row| row["name"].as_str().unwrap().to_owned()));
+            assert!(read.len() <= rows.len() + 2, "{read:?}");
+            match next {
+                Some(next) => cursor = Some(next),
+                None => return read,
+            }
+            between(rows, &read);
+        }
+    }
+
+    fn sorted(mut names: Vec<String>) -> Vec<String> {
+        names.sort();
+        names
+    }
+
+    /// A chat here, its copies on two peers and a row listed twice: at every budget each
+    /// comes once.
+    #[test]
+    fn byte_pages_read_each_row_once() {
+        let rows = vec![
+            json!({"sessionId": "z", "updatedAt": 1}),
+            json!({"sessionId": "dup", "updatedAt": 5, "peer": "p2"}),
+            json!({"sessionId": "a", "updatedAt": 9}),
+            json!({"sessionId": "dup", "updatedAt": 5}),
+            json!({"sessionId": "dup", "updatedAt": 5, "peer": "p1"}),
+            json!({"sessionId": "a", "updatedAt": 9}),
+        ];
+        for budget in [1, 40, 80, 120, 10_000] {
+            let read = read_pages(&mut rows.clone(), budget, |_, _| {});
+            assert_eq!(sorted(read), ["a", "dup", "dup@p1", "dup@p2", "z"], "budget {budget}");
+        }
+        let bad = |cursor| byte_page(&rows, Some(cursor), 1, |_| json!({})).is_err();
+        assert!(bad("k5.0011") && bad("h") && bad("h12") && bad(&format!("h{}", "g".repeat(32))));
+    }
+
+    /// Between pages an unread session becomes the newest, and a read one and an unread one
+    /// leave: no row is skipped or repeated.
+    #[test]
+    fn byte_pages_hold_their_place_while_sessions_update_and_leave() {
+        let mut rows: Vec<Value> = ["a", "b", "c", "d", "e", "f"]
+            .iter()
+            .enumerate()
+            .map(|(i, id)| json!({"sessionId": id, "updatedAt": 10 - i}))
+            .collect();
+        let mut gone = None;
+        let read = read_pages(&mut rows, 1, |rows, read| {
+            let unread: Vec<String> = rows.iter().map(name).filter(|n| !read.contains(n)).collect();
+            match read.len() {
+                1 => rows.iter_mut().filter(|r| name(r) == unread[0]).for_each(|r| {
+                    r["updatedAt"] = json!(100);
+                }),
+                2 => {
+                    gone = Some(unread[0].clone());
+                    rows.retain(|r| name(r) != unread[0] && name(r) != read[0]);
+                }
+                _ => {}
+            }
+            sort_session_summaries(rows);
+        });
+        let mut all: Vec<String> =
+            ["a", "b", "c", "d", "e", "f"].map(String::from).into_iter().collect();
+        all.retain(|n| Some(n) != gone.as_ref());
+        assert_eq!(sorted(read), all);
+    }
+
+    /// The cursor is 33 bytes however long the ids are.
+    #[test]
+    fn a_byte_page_cursor_stays_small_for_a_huge_id() {
+        let rows = vec![
+            json!({"sessionId": "x".repeat(1 << 20), "peer": "p".repeat(1 << 20)}),
+            json!({"sessionId": "y".repeat(1 << 20)}),
+        ];
+        let (page, next) = byte_page(&rows, None, 1, |at| json!({"at": at})).unwrap();
+        assert_eq!(page.len(), 1);
+        assert_eq!(next.unwrap().len(), 33);
+    }
+
     #[test]
     fn stable_scan_keeps_an_unread_chief_when_it_updates_and_cursor_row_leaves() {
         let mut rows = vec![