crates/gunnar-server/tests/repos_list_parallel.rs
//! **Concurrent `Repos.List` calls run side by side** (round 5.3.1).
//!
//! MEASURED on test5 2026-10-08 (the LIMIT bench, ~/runs/2026-10-08/bench/test5/LIMIT.md): the login page's public list
//! (`/login/repos`, one `Repos.List` as the console's guest per visit) reached only 96 rps at 160 offered, p99 7.8 s, and
//! main's own `Repos.List` p50 was 2.7 s while main used 2.85 of its 4 cores — queued, not CPU-bound. The listing asked
//! `may` once per row, and every ask read the catalog's and the identity table's pointers from the control store again
//! (two store read transactions per row) and took the documents' shared cache locks; the catalog walk itself cloned
//! every row while holding the one shard-cache lock every reader of the catalog waits for.
//!
//! The rows are now asked about together (`ControlPlane::may_each_repo`: one read of the catalog, one of the identity
//! table) and the shard walk clones outside the lock. Proved here: the answers are the per-row answers exactly, and
//! N concurrent lists of the same catalog take about the time of one — not N times it.
use std::sync::Arc;
use std::time::{Duration, Instant};
use gunnar_grpc::{
Caller, ControlPlane, EventBus, Identity as _, ReposService, ServerConfig, ServiceContext,
};
use gunnar_proto::repos_server::Repos as _;
use gunnar_proto::*;
use gunnar_server::control_plane::CoreControlPlane;
use gunnar_server::stores::ServerStores;
use tonic::Request;
/// Accounts and their repositories: a public and a private one each (the bench's 700 accounts made 3 each).
const ACCOUNTS: usize = 150;
const PER_ACCOUNT: usize = 4;
fn caller(id: &str) -> Caller {
Caller {
principal_id: id.to_owned(),
fingerprint: format!("SHA256:{id}"),
machine: false,
authenticated: true,
}
}
fn authed<T>(caller: &Caller, body: T) -> Request<T> {
let mut r = Request::new(body);
r.extensions_mut().insert(caller.clone());
r
}
struct World {
_dir: tempfile::TempDir,
plane: Arc<CoreControlPlane>,
ctx: ServiceContext,
}
fn world() -> World {
let dir = tempfile::tempdir().expect("tempdir");
let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
let plane = Arc::new(CoreControlPlane::new(Arc::clone(&stores)));
let control: Arc<dyn ControlPlane> = plane.clone();
let ctx = ServiceContext::new(
control,
Arc::new(EventBus::new(64)),
ServerConfig::default(),
);
// The console's guest: enrolled with NO grant (gunnar-ui `public_repos`).
plane
.add_principal("guest", "guest", PrincipalKind::User)
.expect("the guest");
for a in 0..ACCOUNTS {
let account = format!("acct{a:04}");
plane
.add_principal(&account, &account, PrincipalKind::User)
.expect("an account");
for r in 0..PER_ACCOUNT {
plane
.create_repo(gunnar_grpc::NewRepo {
path: RepoPath::new(&account, format!("repo{r}")),
hash_kind: HashKind::Unspecified,
default_branch: String::new(),
visibility: if r % 2 == 0 {
Visibility::Public
} else {
Visibility::Private
},
description: String::new(),
adopt: false,
})
.expect("a repository");
}
}
World {
_dir: dir,
plane,
ctx,
}
}
fn list(ctx: &ServiceContext, who: &str) -> Vec<String> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let svc = ReposService::new(ctx.clone());
let answer = rt
.block_on(svc.list(authed(&caller(who), ListReposRequest::default())))
.expect("Repos.List");
answer
.into_inner()
.repos
.into_iter()
.filter_map(|r| r.path.map(|p| format!("{}/{}", p.namespace, p.repo)))
.collect()
}
/// **The listing's ACL is the per-row ACL, exactly**: the guest sees every public row and no private one, an account
/// its own private rows too; `may_each_repo` answers what `may` answers row by row, and `false` for a row that does
/// not exist.
#[test]
fn a_listing_shows_exactly_what_each_row_may_show() {
let w = world();
let guest = list(&w.ctx, "guest");
assert_eq!(
guest.len(),
ACCOUNTS * PER_ACCOUNT / 2,
"every public row, no private one"
);
assert!(
guest
.iter()
.all(|p| p.ends_with("repo0") || p.ends_with("repo2")),
"{:?}",
&guest[..4]
);
let mine = list(&w.ctx, "acct0007");
assert!(
mine.contains(&"acct0007/repo1".to_owned()) && mine.contains(&"acct0007/repo3".to_owned()),
"an account lists its own private rows"
);
assert!(
!mine.contains(&"acct0008/repo1".to_owned()),
"never another account's private row"
);
assert_eq!(mine.len(), guest.len() + PER_ACCOUNT / 2);
let mut paths: Vec<RepoPath> = (0..20)
.flat_map(|a| {
(0..PER_ACCOUNT).map(move |r| RepoPath::new(format!("acct{a:04}"), format!("repo{r}")))
})
.collect();
paths.push(RepoPath::new("acct0001", "nosuch"));
for who in ["guest", "acct0003", "nobody"] {
let each = w
.plane
.may_each_repo(who, &paths, GrantLevel::Read)
.expect("may_each_repo");
let one: Vec<bool> = paths
.iter()
.map(|p| {
w.plane
.may(who, &gunnar_grpc::Access::repo(p), GrantLevel::Read)
.expect("may")
})
.collect();
assert_eq!(
each, one,
"{who}: the listing's answer is the row-by-row answer"
);
assert!(!each.last().unwrap(), "a row that does not exist is false");
}
}
/// Per-thread `(on-CPU ns, runnable-but-waiting ns)` from `/proc/thread-self/schedstat`.
fn schedstat() -> Option<(u64, u64)> {
let s = std::fs::read_to_string("/proc/thread-self/schedstat").ok()?;
let mut f = s.split_whitespace().map(|x| x.parse::<u64>().ok());
Some((f.next()??, f.next()??))
}
/// ★ **N concurrent lists do not queue on each other** (test5's 96 rps ceiling): 8 guest listings of the same catalog
/// at once, on 8 threads. Before the fix they queued on the documents' shared locks and per-row store reads.
///
/// 🔴 It asserted on WALL time (8 at once < 4× one), and wall time is the box's as much as the code's: on dwarf
/// (6 physical cores, load 25-30 from other builds, 2026-10-16) 8 lists took 5.05× one because there was no free
/// core to run them on — they were runnable, not queued. What the fix removed is time spent BLOCKED (waiting on a lock
/// or a store read), so that is what is measured: per thread, wall − on-CPU − runnable-waiting (the kernel's
/// `schedstat`), i.e. time asleep. Serialised on one lock, the k-th list sleeps through the k lists before it — a
/// mean of (n−1)/2 lists (measured with a test-side mutex around each list: 4.9× one); side by side it sleeps for a
/// fraction of one (measured 0.1-0.3× typical, 1.03× once at load 20). The bound is half the serialised figure,
/// (n−1)/4 lists. The wall ratio is still printed for the record.
#[test]
fn concurrent_lists_run_side_by_side_not_one_after_another() {
let w = world();
let n = 8usize
.min(
std::thread::available_parallelism()
.map(|p| p.get())
.unwrap_or(4),
)
.max(2);
// Warm the caches (the shards are read once per process).
let _ = list(&w.ctx, "guest");
let one = (0..5)
.map(|_| {
let t = Instant::now();
assert_eq!(list(&w.ctx, "guest").len(), ACCOUNTS * PER_ACCOUNT / 2);
t.elapsed()
})
.min()
.unwrap();
assert!(
schedstat().is_some(),
"/proc/thread-self/schedstat is not readable here; this guard needs it"
);
let rounds = 5;
let mut together = Duration::MAX;
let mut blocked_mean = Duration::MAX;
for _ in 0..rounds {
let barrier = Arc::new(std::sync::Barrier::new(n));
let t = Instant::now();
let handles: Vec<_> = (0..n)
.map(|_| {
let ctx = w.ctx.clone();
let barrier = Arc::clone(&barrier);
std::thread::spawn(move || {
barrier.wait();
let (cpu0, wait0) = schedstat().unwrap();
let t0 = Instant::now();
let len = list(&ctx, "guest").len();
let wall = t0.elapsed().as_nanos() as u64;
let (cpu1, wait1) = schedstat().unwrap();
let asleep = wall.saturating_sub((cpu1 - cpu0) + (wait1 - wait0));
(len, asleep)
})
})
.collect();
let mut asleep = 0u64;
for h in handles {
let (len, a) = h.join().unwrap();
assert_eq!(len, ACCOUNTS * PER_ACCOUNT / 2);
asleep += a;
}
together = together.min(t.elapsed());
blocked_mean = blocked_mean.min(Duration::from_nanos(asleep / n as u64));
}
let ratio = together.as_secs_f64() / one.as_secs_f64().max(1e-6);
println!(
"Repos.List over {} rows as the guest: one {:?}, {n} at once {:?} — {ratio:.2}× one wall (serialised would be {n}×); \
mean time asleep per list {:?} (serialised would be ~{:.1}× one)",
ACCOUNTS * PER_ACCOUNT,
one,
together,
blocked_mean,
(n as f64 - 1.0) / 2.0
);
let bound = one.mul_f64((n as f64 - 1.0) / 4.0);
assert!(
blocked_mean < bound,
"{n} concurrent lists each slept {blocked_mean:?} on average — over {bound:?}, half of what queueing one after \
another costs ({one:?} a list): they queue on each other"
);
}