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
23 changes: 20 additions & 3 deletions cmux-tui/crates/acpmux/src/cli/run/ls.rs
Original file line number Diff line number Diff line change
Expand Up @@ -125,9 +125,7 @@ pub(super) async fn run(args: LsArgs, json_out: bool) -> Result<()> {
// Only a listing that shows usage has the daemon read it first (`hub/usage_view.rs`); the
// compact one shows the rows as they are. Every page asks; the daemon reads at most once
// in 5 s, so the pages after the first take the rows that read left.
let usage = wide
|| json_out
|| fields.as_ref().is_some_and(|f| f.iter().any(|f| f.starts_with("usageStats")));
let usage = shows_usage(wide, json_out, fields.as_deref());
if usage {
params.set("usage", json!(true));
}
Expand Down Expand Up @@ -315,6 +313,14 @@ pub(super) fn parse_duration_ms(text: &str) -> Result<u64> {
Ok(total)
}

/// A projected listing reads usage only if that projection includes it.
fn shows_usage(wide: bool, json_out: bool, fields: Option<&[String]>) -> bool {
match fields {
Some(fields) => fields.iter().any(|f| f == "usageStats" || f.starts_with("usageStats.")),
None => wide || json_out,
}
}

/// The rows `--live` and `--host` keep.
pub(super) struct RowFilter {
live: bool,
Expand Down Expand Up @@ -379,6 +385,17 @@ mod tests {
.collect()
}

#[test]
fn projected_listings_only_refresh_usage_when_they_show_it() {
let plain = vec!["sessionId".to_owned(), "name".to_owned(), "status".to_owned()];
assert!(!shows_usage(false, true, Some(&plain)));
assert!(!shows_usage(true, false, Some(&plain)));
assert!(shows_usage(false, false, Some(&["usageStats.cost".to_owned()])));
assert!(shows_usage(false, true, None));
assert!(shows_usage(true, false, None));
assert!(!shows_usage(false, false, None));
}

#[test]
fn ls_selection_parses_duration_and_projects_fields() {
assert_eq!(parse_duration_ms("1h30m").unwrap(), 5_400_000);
Expand Down
38 changes: 36 additions & 2 deletions cmux-tui/crates/acpmux/src/cli/run/ls_pages.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,15 +12,16 @@

use std::collections::HashSet;
use std::future::Future;
use std::io::{self, Write};

use anyhow::Result;
use cmux_json::JsonSet as _;
use serde_json::Value;

use crate::cli::errors::{AppError, Code};

/// Rows per page: about 4 MB at the 3 to 4 KB a row takes on a busy host, a quarter of the
/// message limit, and four requests for 3,700 sessions.
/// Initial rows per page. A complete census learns a fuller page size from the first
/// successful reply; explicit one-page reads keep the requested count.
pub(super) const PAGE_ROWS: u64 = 1000;

/// The most times in a row a read takes a cursor it had already followed, each with a page twice
Expand Down Expand Up @@ -78,6 +79,10 @@ where
// way past it.
let mut cursors: HashSet<String> = cursor.iter().cloned().collect();
let mut cursor = cursor;
// Learn once, from a successful ordinary first page. A size refusal fixes the
// smaller fallback for the rest of this read, so heterogeneous rows cannot
// repeatedly make us grow and retry the same oversized page.
let mut learn_size = plan.size == PAGE_ROWS && !plan.one && plan.want.is_none();
loop {
let left = plan.want.map(|want| want.saturating_sub(rows.len()));
let ask = match left {
Expand All @@ -94,13 +99,18 @@ where
Ok(reply) => reply,
Err(e) => match too_large(&e) {
Some(_) if ask > 1 => {
learn_size = false;
(size, widen) = (ask / 2, 0);
continue;
}
Some(bytes) => return Err(one_row_too_large(bytes).into()),
None => return Err(e),
},
};
if learn_size && reply.get("nextCursor").is_some_and(Value::is_string) {
size = fuller_page(&reply).max(size);
}
learn_size = false;
// A daemon that pages answers a `limit` with `nextCursor` (null on the last page); one
// too old for pages sends its whole list, with no cursor and from the start.
let pages = reply.get("nextCursor").is_some();
Expand Down Expand Up @@ -167,6 +177,30 @@ where
}
}

/// Size the next page from measured serialized bytes, leaving the protocol's existing
/// near-cap headroom for the reply envelope and variation between rows. Count bytes
/// without allocating another copy of the page. The daemon's byte bound and the
/// halve-on-error path remain authoritative when later rows are larger.
fn fuller_page(reply: &Value) -> u64 {
struct Bytes(usize);
impl Write for Bytes {
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
self.0 = self.0.saturating_add(bytes.len());
Ok(bytes.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
let rows = reply.get("sessions").and_then(Value::as_array).map_or(0, Vec::len);
let mut bytes = Bytes(0);
if serde_json::to_writer(&mut bytes, reply).is_err() || bytes.0 == 0 {
return PAGE_ROWS;
}
let count = rows.saturating_mul(crate::wire_size::NEAR_CAP_BYTES) / bytes.0;
u64::try_from(count).unwrap_or(PAGE_ROWS)
}

/// One session on one host: a chat's copies on peers are different rows.
fn row_key(row: &Value) -> (Option<String>, Option<String>) {
let text = |k: &str| row.get(k).and_then(Value::as_str).map(str::to_owned);
Expand Down
60 changes: 56 additions & 4 deletions cmux-tui/crates/acpmux/src/cli/run/ls_pages_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -93,12 +93,12 @@ async fn a_list_larger_than_a_page_is_read_whole_each_row_once() {
let seen = Seen::default();
let got = read_all(&list, plan(PAGE_ROWS), usize::MAX, &seen).await.unwrap();
assert_eq!(ids(&got.rows), ids(&list.borrow()));
assert_eq!(seen.requests.get(), 3);
assert_eq!(seen.requests.get(), 2);
assert_eq!(got.next, Some(Value::Null));
}

/// A fleet host, 2026-10-30: 3,700 sessions of about 4 KB came to 15.7 MB, past the 15 MiB a
/// reply may hold. Its pages hold a quarter of that each.
/// reply may hold. Later pages use the measured available space.
#[tokio::test]
async fn a_busy_hosts_list_goes_in_pages_under_the_message_limit() {
let list = RefCell::new(sessions(3_700, 4_300));
Expand All @@ -107,9 +107,9 @@ async fn a_busy_hosts_list_goes_in_pages_under_the_message_limit() {
let seen = Seen::default();
let got = read_all(&list, plan(PAGE_ROWS), crate::rpc::MAX_MESSAGE_BYTES, &seen).await.unwrap();
assert_eq!(got.rows.len(), 3_700);
assert_eq!(seen.requests.get(), 4);
assert_eq!(seen.requests.get(), 3);
let largest = seen.largest.get();
assert!(largest < crate::rpc::MAX_MESSAGE_BYTES / 3, "{largest}");
assert!(largest < crate::rpc::MAX_MESSAGE_BYTES, "{largest}");
}

#[tokio::test]
Expand Down Expand Up @@ -364,3 +364,55 @@ async fn many_peer_copies_are_read_past_one_after_another() {
rows.iter().map(key).collect::<Vec<_>>()
);
}

/// The reported host mirrors about 8,000 rows. Full pages must enumerate every
/// local/peer identity without eight separately paced snapshot requests.
#[tokio::test]
async fn an_eight_thousand_row_peer_census_uses_fuller_pages() {
let mut rows = sessions(8_000, 3_000);
for (n, row) in rows.iter_mut().enumerate().skip(445) {
row["peer"] = json!(format!("peer-{}", n % 20));
}
let list = RefCell::new(rows);
// Same fake daemon, rows and size cap; the baseline enforces the old fixed
// request size while still exercising the production cursor/dedup reader.
let baseline = Seen::default();
let mut fixed = counted(&list, crate::rpc::MAX_MESSAGE_BYTES, Some(&baseline));
let started = std::time::Instant::now();
let before = read(json!({}), None, plan(PAGE_ROWS), keep, |mut params| {
params["limit"] = json!(PAGE_ROWS);
fixed(params)
})
.await
.unwrap();
let before_ms = started.elapsed().as_secs_f64() * 1000.0;
let seen = Seen::default();
let started = std::time::Instant::now();
let got = read_all(&list, plan(PAGE_ROWS), crate::rpc::MAX_MESSAGE_BYTES, &seen).await.unwrap();
let after_ms = started.elapsed().as_secs_f64() * 1000.0;
assert_eq!(before.rows, *list.borrow());
assert_eq!(got.rows, before.rows);
assert_eq!(baseline.requests.get(), 8);
assert_eq!(seen.requests.get(), 3);
assert!(seen.largest.get() < crate::rpc::MAX_MESSAGE_BYTES);
eprintln!(
"8000-row census: fixed={} requests, {before_ms:.3} ms; adaptive={} requests, {after_ms:.3} ms; largest={} bytes",
baseline.requests.get(),
seen.requests.get(),
seen.largest.get()
);
}

#[tokio::test]
async fn larger_later_rows_shrink_the_learned_page_without_repeated_growth() {
let mut rows = sessions(4_000, 10);
for row in rows.iter_mut().skip(1_000) {
row["title"] = json!("x".repeat(8_000));
}
let list = RefCell::new(rows);
let seen = Seen::default();
let got = read_all(&list, plan(PAGE_ROWS), crate::rpc::MAX_MESSAGE_BYTES, &seen).await.unwrap();
assert_eq!(got.rows, *list.borrow());
assert!(seen.requests.get() < 20, "{}", seen.requests.get());
assert!(seen.largest.get() < crate::rpc::MAX_MESSAGE_BYTES);
}
Loading