src/admin_queue.rs
//! **The admin view's Queue and Admission panes — their decisions**, pure and
//! host-tested; `pages/admin_queue.rs` renders them.
//!
//! * **Queue** reads `GET /api/admin/jobs[?state=…]`: holger's
//! `JobService.ListJobs`, relayed by this console's server to an ADMIN's
//! session only. Every row says what it is doing (stage), how often it was
//! tried (attempts / max), why it failed, when it runs again (an absolute UTC
//! time AND "in N s" — the first is what a log line is compared with, the
//! second what a person waits for) and its priority class.
//! * **Admission** reads `GET /api/admin/admission`: holger's
//! `AdmissionService.ListUsage` plus one `AdmitPlan` per named consumer,
//! composed by the server. A server on `pass-through` enforces no limit, so
//! used / limit reads **"N/A (pass-through)"** — never `0`, which would read
//! as a measurement of an empty budget.
//!
//! Who may see either is the SERVER's decision (`ADMIN_READS` in the console
//! server's middleware, and the handler again); nothing here hides data.
#![cfg_attr(not(target_arch = "wasm32"), allow(dead_code))]
use serde::Deserialize;
use crate::admin::human;
#[cfg(target_arch = "wasm32")]
use crate::grpc::pb;
/// holger's messages, natively: `crate::grpc` is wasm-only (its transport is the
/// browser's), but these decisions are host-tested over the SAME generated
/// types the page reads, never a hand-kept mirror.
#[cfg(not(target_arch = "wasm32"))]
pub mod pb {
tonic::include_proto!("holger.v1");
}
/// The most jobs one Queue read asks for.
pub const DEFAULT_JOBS: i32 = 500;
/// The most consumers one Admission read asks for.
pub const MAX_CONSUMERS: i32 = 200;
/// The state filter, in order: `(wire word, label)`; `""` = every state.
pub const JOB_STATES: [(&str, &str); 5] = [
("", "All"),
("waiting", "Waiting"),
("running", "Running"),
("failed", "Failed"),
("done", "Done"),
];
/// holger's word for the policy that enforces nothing.
pub const PASS_THROUGH: &str = "pass-through";
/// What used / limit reads under `pass-through`.
pub const NOT_APPLICABLE: &str = "N/A (pass-through)";
/// Pure: the state filter holger is asked for (only the four words holger
/// knows; any other word is the empty filter).
pub fn job_state(state: &str) -> String {
if JOB_STATES.iter().any(|(w, _)| !w.is_empty() && *w == state) {
state.to_owned()
} else {
String::new()
}
}
/// Pure: which consumers to ask `AdmitPlan` about — every named one
/// (`principal:` / `peer:`), plus whether to ask for a fresh anonymous day
/// token. Nothing under `pass-through`, which plans nobody.
pub fn plans_to_ask(usage: &pb::ListUsageResponse) -> (Vec<String>, bool) {
if usage.policy == PASS_THROUGH {
return (Vec::new(), false);
}
let named = usage
.consumers
.iter()
.map(|c| c.consumer.clone())
.filter(|k| k.starts_with("principal:") || k.starts_with("peer:"))
.collect();
let anon = usage.consumers.iter().any(|c| c.consumer.starts_with("anon:"));
(named, anon)
}
/// The Admission pane's read: holger's usage book, then one plan per
/// consumer, each asked of holger as the person. `Ok(None)`: this holger
/// serves no admission door.
#[cfg(target_arch = "wasm32")]
pub async fn admission_view() -> Result<Option<AdmissionView>, String> {
use crate::grpc::pb as gpb;
let Some(usage) = crate::grpc::answer_or_none(crate::grpc::admission().list_usage(gpb::ListUsageRequest { limit: MAX_CONSUMERS }).await)? else {
return Ok(None);
};
let (named, anon) = plans_to_ask(&usage);
let mut view = AdmissionView { usage, ..AdmissionView::default() };
for consumer in named.into_iter().map(Some).chain(anon.then_some(None)) {
let asked = consumer.clone().unwrap_or_else(|| "anonymous".to_owned());
let req = gpb::AdmitPlanRequest { consumer: asked.clone(), work: "download".to_owned(), size_hint: 0 };
match crate::grpc::admission().admit_plan(req).await {
Ok(r) if consumer.is_some() => view.plans.push(r.into_inner()),
Ok(r) => view.anonymous = Some(r.into_inner()),
Err(status) => view.plan_errors.push(format!("{asked}: {}", status.message())),
}
}
Ok(Some(view))
}
/// Pure: the pending-list priority as its admission class (holger's
/// `JobRow.priority`: 0 Full, 1 Reduced, 2 Floor, 3 Queued).
pub fn priority_class(p: u32) -> String {
match p {
0 => "full".into(),
1 => "reduced".into(),
2 => "floor".into(),
3 => "queued".into(),
n => format!("priority {n}"),
}
}
/// Pure: `attempts / max` (`max == 0` is an older holger that does not count).
pub fn attempts_words(attempts: u32, max_attempts: u32) -> String {
if max_attempts == 0 {
attempts.to_string()
} else {
format!("{attempts}/{max_attempts}")
}
}
/// Pure: the status with the stage beside it when the stage says more.
pub fn stage_words(status: &str, stage: &str) -> String {
if stage.is_empty() || stage == status {
status.to_owned()
} else {
format!("{status} · {stage}")
}
}
/// Pure: days since the epoch → (year, month, day), proleptic Gregorian (H.
/// Hinnant's `civil_from_days`).
fn civil(days: i64) -> (i64, u32, u32) {
let z = days + 719_468;
let era = z.div_euclid(146_097);
let doe = z.rem_euclid(146_097);
let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
let d = (doy - (153 * mp + 2) / 5 + 1) as u32;
let m = if mp < 10 { mp + 3 } else { mp - 9 } as u32;
let y = yoe + era * 400 + i64::from(m <= 2);
(y, m, d)
}
/// Pure: µs since the epoch as `YYYY-MM-DD HH:MM:SS UTC`; `—` for none.
pub fn utc_micros(micros: i64) -> String {
if micros <= 0 {
return "—".into();
}
let secs = micros.div_euclid(1_000_000);
let (y, m, d) = civil(secs.div_euclid(86_400));
let s = secs.rem_euclid(86_400);
format!("{y:04}-{m:02}-{d:02} {:02}:{:02}:{:02} UTC", s / 3_600, (s % 3_600) / 60, s % 60)
}
/// Pure: when a waiting retry runs again — the absolute time AND the distance
/// from `now` in seconds (rounded up, so a retry 0.4 s away is not "now").
pub fn next_attempt_words(next_micros: i64, now_micros: i64) -> String {
if next_micros <= 0 {
return "—".into();
}
let delta = next_micros - now_micros;
let rel = if delta > 0 {
format!("in {} s", (delta + 999_999) / 1_000_000)
} else if delta > -1_000_000 {
"now".to_owned()
} else {
format!("due {} s ago", (-delta) / 1_000_000)
};
format!("{} ({rel})", utc_micros(next_micros))
}
/// The Admission pane's composed read ([`admission_view`]): holger's book and
/// holger's plans, nothing decided here.
#[derive(Debug, Clone, Default, Deserialize)]
#[serde(default)]
pub struct AdmissionView {
pub usage: pb::ListUsageResponse,
/// One `AdmitPlan` per named consumer (empty under pass-through).
pub plans: Vec<pb::AdmitPlanResponse>,
/// The plan a fresh anonymous day token is given (its per-day budget).
pub anonymous: Option<pb::AdmitPlanResponse>,
/// A plan the server could not get, in holger's words.
pub plan_errors: Vec<String>,
}
/// One consumer, as the Admission table draws it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConsumerLine {
/// `principal` | `paired peer` | `anonymous`.
pub kind: &'static str,
/// The name, or `anonymous (today)`.
pub label: String,
/// The raw book key, for the detail cell.
pub key: String,
pub used_limit: String,
pub class: String,
pub why: String,
pub in_flight: u32,
}
/// Pure: bytes that crossed a wire — what a share is sold in (holger's
/// `CostTally::bytes`: net in + net out + upstream fetch).
pub fn wire_bytes(c: Option<&pb::CostRow>) -> u64 {
c.map(|c| c.net_in.saturating_add(c.net_out).saturating_add(c.upstream_fetch)).unwrap_or(0)
}
/// Pure: a book key in words. `anon:<day>:<hex>` is a per-day token that
/// cannot be turned back into a client — it reads `anonymous (today)` (or its
/// day, should the book still hold an older one).
pub fn consumer_words(key: &str, today_day: i64) -> (&'static str, String) {
if let Some(sub) = key.strip_prefix("principal:") {
return ("principal", sub.to_owned());
}
if let Some(name) = key.strip_prefix("peer:") {
return ("paired peer", name.to_owned());
}
if let Some(rest) = key.strip_prefix("anon:") {
let day = rest.split(':').next().and_then(|d| d.parse::<i64>().ok());
let label = match day {
Some(d) if d != today_day => format!("anonymous ({})", utc_micros(d * 86_400_000_000).get(..10).unwrap_or("?")),
_ => "anonymous (today)".to_owned(),
};
return ("anonymous", label);
}
("unknown", key.to_owned())
}
/// Pure: used / limit over the window.
///
/// `plan_left` is the plan's `budget_left`: `None` = no plan was had,
/// `Some(None)` = unlimited, `Some(Some(n))` = n bytes left. For an anonymous
/// token the probe plan's `budget_left` IS the day budget (`day_budget`).
pub fn used_limit_words(policy: &str, used: u64, plan_left: Option<Option<u64>>, day_budget: bool) -> String {
if policy == PASS_THROUGH {
return NOT_APPLICABLE.to_owned();
}
match plan_left {
None => format!("{} / ? (no admission plan)", human(used)),
Some(None) => format!("{} / unlimited", human(used)),
Some(Some(left)) if day_budget => format!("{} / {} per day", human(used), human(left)),
Some(Some(left)) => format!("{} / {} ({} left)", human(used), human(used.saturating_add(left)), human(left)),
}
}
/// Pure: every consumer row the Admission table draws, in the server's order
/// (heaviest window first).
pub fn consumer_lines(v: &AdmissionView, today_day: i64) -> Vec<ConsumerLine> {
let policy = v.usage.policy.as_str();
v.usage
.consumers
.iter()
.map(|row| {
let (kind, label) = consumer_words(&row.consumer, today_day);
let used = wire_bytes(row.window.as_ref());
let anon = kind == "anonymous";
let plan = if anon { v.anonymous.as_ref() } else { v.plans.iter().find(|p| p.consumer == row.consumer) };
let used_limit = used_limit_words(policy, used, plan.map(|p| p.budget_left), anon);
// The plan asked NOW is the sentence to show; an anonymous row's own
// last plan is its own (the probe speaks for a fresh token).
let (class, why) = match plan {
Some(p) if !anon => (p.class.clone(), p.why.clone()),
_ => (row.class.clone(), row.why.clone()),
};
ConsumerLine {
kind,
label,
key: row.consumer.clone(),
used_limit,
class: if class.is_empty() { "—".into() } else { class },
why: if why.is_empty() { "—".into() } else { why },
in_flight: row.in_flight,
}
})
.collect()
}
/// Pure: the door line — policy, in-flight, ceiling and shares.
pub fn door_words(u: &pb::ListUsageResponse) -> String {
let opt = |o: Option<u32>| o.map_or_else(|| "none".to_owned(), |n| n.to_string());
format!(
"policy {} · in flight {} of {} · shares full {} / reduced {} / floor {} · pull-through on the job queue: {}",
if u.policy.is_empty() { "—" } else { &u.policy },
u.in_flight_total,
opt(u.max_in_flight),
opt(u.share_full),
opt(u.share_reduced),
opt(u.share_floor),
if u.pull_through_queue { "yes" } else { "no" }
)
}
/// Pure: one work kind's numbers for the detail row.
pub fn work_words(w: &pb::WorkUsageRow) -> String {
format!(
"{}: {} requests, {} refused · estimated {} · actual {}",
w.work,
w.requests,
w.refused,
human(wire_bytes(w.estimated.as_ref())),
human(wire_bytes(w.actual.as_ref()))
)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_state_filter_is_holgers_four_words_and_all() {
let words: Vec<_> = JOB_STATES.iter().map(|(w, _)| *w).collect();
assert_eq!(words, ["", "waiting", "running", "failed", "done"]);
assert_eq!(job_state(""), "");
assert_eq!(job_state("waiting"), "waiting");
assert_eq!(job_state("done"), "done");
assert_eq!(job_state("x&kind=y"), "", "no word but holger's reaches the request");
}
#[test]
fn priority_attempts_and_stage_in_words() {
assert_eq!([0, 1, 2, 3].map(priority_class), ["full", "reduced", "floor", "queued"]);
assert_eq!(priority_class(9), "priority 9");
assert_eq!(attempts_words(2, 5), "2/5");
assert_eq!(attempts_words(1, 0), "1");
assert_eq!(stage_words("retry_wait", "failed"), "retry_wait · failed");
assert_eq!(stage_words("running", "fetching"), "running · fetching");
assert_eq!(stage_words("done", "done"), "done");
assert_eq!(stage_words("queued", ""), "queued");
}
#[test]
fn times_are_utc_and_the_next_attempt_says_how_far() {
assert_eq!(utc_micros(0), "—");
assert_eq!(utc_micros(1_000_000), "1970-01-01 00:00:01 UTC");
// 2026-09-28 12:00:05 UTC = 1 790 596 805 s.
let t = 1_790_596_805_000_000;
assert_eq!(utc_micros(t), "2026-09-28 12:00:05 UTC");
assert_eq!(utc_micros(951_782_400_000_000), "2000-02-29 00:00:00 UTC", "a leap day");
assert_eq!(next_attempt_words(t, t - 42_000_000), "2026-09-28 12:00:05 UTC (in 42 s)");
assert_eq!(next_attempt_words(t, t - 400_000), "2026-09-28 12:00:05 UTC (in 1 s)", "rounded up");
assert_eq!(next_attempt_words(t, t + 300_000), "2026-09-28 12:00:05 UTC (now)");
assert_eq!(next_attempt_words(t, t + 7_000_000), "2026-09-28 12:00:05 UTC (due 7 s ago)");
assert_eq!(next_attempt_words(0, t), "—", "not waiting");
}
#[test]
fn consumers_are_named_and_an_anonymous_token_reads_today() {
let today = 20_724; // 2026-09-28
assert_eq!(consumer_words("principal:ada", today), ("principal", "ada".into()));
assert_eq!(consumer_words("peer:vm3", today), ("paired peer", "vm3".into()));
assert_eq!(consumer_words("anon:20724:0a1b2c3d", today), ("anonymous", "anonymous (today)".into()));
assert_eq!(consumer_words("anon:20723:0a1b2c3d", today), ("anonymous", "anonymous (2026-09-27)".into()));
assert_eq!(consumer_words("what", today), ("unknown", "what".into()));
}
fn cost(net_out: u64) -> Option<pb::CostRow> {
Some(pb::CostRow { net_out, ..Default::default() })
}
fn view(policy: &str) -> AdmissionView {
AdmissionView {
usage: pb::ListUsageResponse {
policy: policy.into(),
consumers: vec![
pb::UsageRow { consumer: "principal:ada".into(), window: cost(3_000_000_000), class: "full".into(), why: "old".into(), ..Default::default() },
pb::UsageRow { consumer: "peer:vm3".into(), window: cost(0), ..Default::default() },
pb::UsageRow { consumer: "anon:20724:0a1b2c3d".into(), window: cost(1_000), class: "floor".into(), why: "anonymous: floor".into(), ..Default::default() },
],
..Default::default()
},
..Default::default()
}
}
/// ★ **Pass-through enforces no limit, so no row reads a number** — least
/// of all `0`, which would read as a budget measured empty.
#[test]
fn pass_through_reads_not_applicable_and_never_zero() {
let lines = consumer_lines(&view(PASS_THROUGH), 20_724);
assert_eq!(lines.len(), 3);
for l in &lines {
assert_eq!(l.used_limit, "N/A (pass-through)", "{l:?}");
assert!(!l.used_limit.contains('0'), "{l:?}");
}
assert_eq!(used_limit_words(PASS_THROUGH, 0, Some(Some(0)), false), NOT_APPLICABLE);
}
#[test]
fn usage_share_reads_used_of_limit_with_the_plan_sentence() {
let mut v = view("usage-share");
v.plans = vec![
pb::AdmitPlanResponse { consumer: "principal:ada".into(), class: "reduced".into(), why: "past its window budget".into(), budget_left: Some(0), ..Default::default() },
pb::AdmitPlanResponse { consumer: "peer:vm3".into(), class: "full".into(), why: "a paired peer".into(), budget_left: None, ..Default::default() },
];
v.anonymous = Some(pb::AdmitPlanResponse { class: "floor".into(), budget_left: Some(5_000_000_000), ..Default::default() });
let lines = consumer_lines(&v, 20_724);
assert_eq!(lines[0].kind, "principal");
assert_eq!(lines[0].used_limit, "3.0 GB / 3.0 GB (0 B left)");
assert_eq!((lines[0].class.as_str(), lines[0].why.as_str()), ("reduced", "past its window budget"), "the plan asked now, not the last one");
assert_eq!(lines[1].kind, "paired peer");
assert_eq!(lines[1].used_limit, "0 B / unlimited");
assert_eq!(lines[2].label, "anonymous (today)");
assert_eq!(lines[2].used_limit, "1.0 kB / 5.0 GB per day");
assert_eq!((lines[2].class.as_str(), lines[2].why.as_str()), ("floor", "anonymous: floor"));
v.plans.clear();
assert_eq!(consumer_lines(&v, 20_724)[0].used_limit, "3.0 GB / ? (no admission plan)");
}
/// One plan per named consumer, one for a fresh anonymous token, none at
/// all under pass-through (it plans nobody).
#[test]
fn the_plans_asked_are_the_named_consumers_and_one_anonymous() {
let (named, anon) = plans_to_ask(&view("usage-share").usage);
assert_eq!(named, ["principal:ada", "peer:vm3"]);
assert!(anon);
assert_eq!(plans_to_ask(&view(PASS_THROUGH).usage), (Vec::new(), false));
}
#[test]
fn the_door_line_says_none_rather_than_zero() {
let u = pb::ListUsageResponse { policy: PASS_THROUGH.into(), in_flight_total: 2, ..Default::default() };
assert_eq!(door_words(&u), "policy pass-through · in flight 2 of none · shares full none / reduced none / floor none · pull-through on the job queue: no");
}
#[test]
fn the_server_answer_is_read_with_every_field_defaulted() {
let v: AdmissionView = serde_json::from_str(r#"{"usage":{"policy":"pass-through","consumers":[{"consumer":"peer:vm3"}]}}"#).expect("reads");
assert_eq!(v.usage.consumers[0].consumer, "peer:vm3");
assert!(v.plans.is_empty() && v.anonymous.is_none());
let r: pb::ListJobsResponse = serde_json::from_str(r#"{"jobs":[{"job_id":"j1","stage":"fetching","attempts":2,"max_attempts":5,"next_attempt_micros":7,"priority":1}]}"#).expect("reads");
assert_eq!((r.jobs[0].attempts, r.jobs[0].max_attempts, r.jobs[0].priority), (2, 5, 1));
}
}