Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 57 additions & 2 deletions cmux-tui/crates/acpmux/src/hub/list_guard.rs
Original file line number Diff line number Diff line change
Expand Up @@ -231,7 +231,7 @@ impl Snapshot {
None => p.remove("cursor"),
};
}
let picked = super::session_pages::session_list_reply(keys.clone(), &filters)?;
let picked = super::session_pages::session_list_reply_borrowed(keys, &filters)?;
let rows =
picked.get("sessions").and_then(Value::as_array).map(Vec::as_slice).unwrap_or_default();
let row = |key: &Value| row_text(text, spans, key, budget);
Expand Down Expand Up @@ -291,7 +291,7 @@ impl Snapshot {
Some(full) => (full.whole.as_str(), &full.spans, &full.keys),
None => (self.whole.as_str(), &self.spans, &self.keys),
};
let picked = super::session_pages::session_list_reply(keys.clone(), params)?;
let picked = super::session_pages::session_list_reply_borrowed(keys, params)?;
let rows =
picked.get("sessions").and_then(Value::as_array).map(Vec::as_slice).unwrap_or_default();
let mut next = picked.get("nextCursor").cloned();
Expand Down Expand Up @@ -704,6 +704,61 @@ mod tests {
assert!(list.page(None, &json!({"cursor": 3}), usize::MAX).is_err());
}



/// A many-session daemon keeps the full keyset in its snapshot, but a one-row
/// request copies and serializes only its selected row, with the same cursors
/// and exact filters as the old owned path.
#[test]
fn three_thousand_key_snapshot_pages_only_selected_rows() {
let rows: Vec<Value> = (0..3_000)
.map(|i| json!({
"sessionId": format!("s{i:04}"),
"updatedAt": 3_000 - i,
"peer": (i % 5 == 0).then_some("machine"),
"status": if i % 3 == 0 { "closed" } else { "idle" },
}))
.collect();
let list = Snapshot::new(
(false, 0, 0, Vec::new()),
0,
Duration::ZERO,
false,
Snapshot::serialize(&rows),
);
let first: Value =
serde_json::from_str(&list.page(None, &json!({"limit": 1}), usize::MAX).unwrap()).unwrap();
assert_eq!(first["sessions"].as_array().unwrap().len(), 1);
assert_eq!(first["sessions"][0]["sessionId"], "s0000");
let second: Value = serde_json::from_str(
&list.page(
None,
&json!({"limit": 1, "cursor": first["nextCursor"]}),
usize::MAX,
)
.unwrap(),
)
.unwrap();
assert_eq!(second["sessions"][0]["sessionId"], "s0001");
let filter = json!({"localOnly": true, "excludeStatus": ["closed"],
"orderBy": "sessionId", "limit": 1});
let picked: Value =
serde_json::from_str(&list.page(None, &filter, usize::MAX).unwrap()).unwrap();
assert_eq!(picked["sessions"][0]["sessionId"], "s0001");
assert_eq!(picked["nextCursor"], "s:s0001");
// The existing owned API and the borrowed one remain byte-for-byte identical.
for params in [
json!({}),
json!({"limit": 1}),
json!({"status": "closed", "peer": "machine", "limit": 2}),
filter,
] {
let owned = super::super::session_pages::session_list_reply(rows.clone(), &params).unwrap();
let borrowed =
super::super::session_pages::session_list_reply_borrowed(&rows, &params).unwrap();
assert_eq!(owned, borrowed, "{params}");
}
}
/// A list read from a session list cursor (`k...`) with no `limit`, too large for one page,
/// keeps that cursor's cutoff on every page: the rows after the session it names, each once.
#[tokio::test]
Expand Down
50 changes: 25 additions & 25 deletions cmux-tui/crates/acpmux/src/hub/session_pages.rs
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,14 @@ pub(crate) fn sort_session_summaries(list: &mut [Value]) {

/// The session list reply for `params`: `{sessions}` without `limit`,
/// `{sessions, nextCursor}` with it.
#[cfg(test)]
pub(crate) fn session_list_reply(sorted: Vec<Value>, params: &Value) -> Result<Value, RpcError> {
session_list_reply_borrowed(&sorted, params)
}

/// Pick rows by reference from the shared session-list snapshot, then clone only rows
/// actually returned. A list page of one must not copy all 3,000 session keys first.
pub(crate) fn session_list_reply_borrowed(sorted: &[Value], params: &Value) -> Result<Value, RpcError> {
let limit = match params.get("limit") {
None | Some(Value::Null) => None,
Some(v) => Some(
Expand Down Expand Up @@ -67,29 +74,22 @@ pub(crate) fn session_list_reply(sorted: Vec<Value>, params: &Value) -> Result<V
return Err(RpcError::invalid_params("orderBy sessionId requires localOnly"));
}
let excluded = status_filter(params, "excludeStatus")?;
let sorted: Vec<Value> =
if peer.is_none() && status.is_none() && !local_only && excluded.is_empty() {
sorted
} else {
sorted
.into_iter()
.filter(|s| peer.is_none_or(|p| s.get("peer").and_then(Value::as_str) == Some(p)))
.filter(|s| {
status.is_none_or(|st| s.get("status").and_then(Value::as_str) == Some(st))
})
.filter(|s| !local_only || s.get("peer").and_then(Value::as_str).is_none())
.filter(|s| {
s.get("status").and_then(Value::as_str).is_none_or(|st| !excluded.contains(&st))
})
.collect()
};
let selected: Vec<&Value> = sorted
.iter()
.filter(|s| peer.is_none_or(|p| s.get("peer").and_then(Value::as_str) == Some(p)))
.filter(|s| status.is_none_or(|st| s.get("status").and_then(Value::as_str) == Some(st)))
.filter(|s| !local_only || s.get("peer").and_then(Value::as_str).is_none())
.filter(|s| {
s.get("status").and_then(Value::as_str).is_none_or(|st| !excluded.contains(&st))
})
.collect();
if order.is_some() {
return session_id_page(sorted, params, limit);
return session_id_page(selected, params, limit);
}
let start = cursor.map_or(0, |c| c.start(&sorted));
let mut sessions: Vec<Value> = sorted.into_iter().skip(start).collect();
let start = cursor.map_or(0, |c| c.start(&selected));
let mut sessions: Vec<&Value> = selected.into_iter().skip(start).collect();
let Some(limit) = limit else {
return Ok(page(sessions, None));
return Ok(page(sessions.into_iter().cloned().collect(), None));
};
let limit = usize::try_from(limit).unwrap_or(usize::MAX);
let next = if sessions.len() > limit {
Expand All @@ -98,7 +98,7 @@ pub(crate) fn session_list_reply(sorted: Vec<Value>, params: &Value) -> Result<V
} else {
None
};
Ok(page(sessions, Some(next.map_or(Value::Null, Value::String))))
Ok(page(sessions.into_iter().cloned().collect(), 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
Expand Down Expand Up @@ -157,7 +157,7 @@ pub(crate) fn byte_page<R>(

/// IDs are immutable: an unread session updating between pages cannot be lost.
fn session_id_page(
mut rows: Vec<Value>,
mut rows: Vec<&Value>,
params: &Value,
limit: Option<u64>,
) -> Result<Value, RpcError> {
Expand All @@ -173,7 +173,7 @@ fn session_id_page(
rows.retain(|row| after.is_none_or(|id| session_id(row) > id));
rows.sort_by(|a, b| session_id(a).cmp(session_id(b)));
let Some(limit) = limit else {
return Ok(page(rows, None));
return Ok(page(rows.into_iter().cloned().collect(), None));
};
let limit = usize::try_from(limit).unwrap_or(usize::MAX);
let next = if rows.len() > limit {
Expand All @@ -182,7 +182,7 @@ fn session_id_page(
} else {
None
};
Ok(page(rows, Some(next.unwrap_or(Value::Null))))
Ok(page(rows.into_iter().cloned().collect(), Some(next.unwrap_or(Value::Null))))
}

/// `{sessions, nextCursor?}`, moving the sessions in: `json!` would deep-copy the whole list.
Expand Down Expand Up @@ -295,7 +295,7 @@ impl SessionCursor {
/// Where the page starts in `sorted`. The named session gone or moved, the
/// page starts at the first session no newer than it was: one updated in
/// that same millisecond may repeat, none is skipped.
fn start(self, sorted: &[Value]) -> usize {
fn start(self, sorted: &[&Value]) -> usize {
sorted
.iter()
.position(|s| stamp(s) == self.updated_ms && row_key(s) == self.key)
Expand Down
Loading