server/lib/src/admin_view.rs
//! **The admin view** — `AdminService.AdminSnapshot`: what an administrator's
//! console panes are shown about THIS machine, and nobody else is.
//!
//! gunnar has the same idea (its `Admin.Capacity`, `Admin.Health` and
//! `Bootstrap.Status` behind the console's admin panes); this is holger's own
//! code for it (RULE 1: the pattern, never the code). The one product-neutral
//! piece — every mount, one `statvfs(3)` each, the floor, `/sys/block` — is
//! edda `disk-snapshot` (RULE 2); everything here reads holger's own sources:
//!
//! | pane | source |
//! |---|---|
//! | Disks | `disk_snapshot::take` over `/` (system), the data root (data) and every repository's storage place |
//! | Twin | [`crate::twin::TwinPosture`] (role, follow, completeness, the door's lane), the mTLS credentials on disk ([`crate::twin_tls::load`]), a primary's roster ([`crate::grow::GrowPlane::twins`]) |
//! | Sync | the twins above and every paired downstream ([`crate::peers::PeerBoard::rows`], `GrowService.Downstreams`) |
//! | Growth | [`crate::grow::GrowPlane::status`] (the phase machine and the boot journal = the history) and the capacity ladder ([`crate::grow::capacity_growth`]) |
//! | Clock | [`crate::clock`] (trusted, owed) and edda `estate_clock::report::{last, boot}` |
//! | Plugins / Monetize | the boot's module census, the meters, the metered build's entitlement store and self-registration |
//! | Audit | the configured sink's mode and its segment manifest |
//! | Logs | the installer's boot log and its door ([`set_log_door`], registered by the CLI that owns it) |
//!
//! **Who may read it is decided by the RPC, not here**: the handler in
//! `grpc/services.rs` asks `authorize_admin_identity` first (the Admin role
//! under RBAC; every other role and every invalid credential refused by name).
//! The console's backend adds its own session rung on top (`/api/admin/…` is an
//! admin's), because holger sees one credential for every browser session.
//!
//! Everything is measured at the call. A figure that could not be measured
//! carries its reason, never a zero that reads as a measurement.
use std::path::{Path, PathBuf};
use std::sync::OnceLock;
use crate::grpc::holger_proto as pb;
// The Disks pane and the disk speed test over edda `disk-snapshot`, and the twin's
// credential read over nordisk-twin: the estate's (`estate`, H2's estate cut). A build
// without it says so on the pane (`admin_view/disks_off.rs`).
#[cfg(feature = "estate")]
mod disks;
#[cfg(feature = "estate")]
pub use disks::*;
#[cfg(not(feature = "estate"))]
mod disks_off;
#[cfg(not(feature = "estate"))]
pub use disks_off::*;
/// Log lines returned when the caller asks for none in particular.
pub const DEFAULT_LOG_LINES: usize = 200;
/// The most log lines one snapshot carries.
pub const MAX_LOG_LINES: usize = 2000;
/// Where the Stripe webhook's last delivery is NOT measured, in words. The paid build's pane
/// only (A28 M13): an open build has no payments pane.
#[cfg(feature = "metered")]
pub const STRIPE_NOTE: &str = "not measured on this box: Stripe delivers its webhook to monetize on the front (/callback/stripe), and monetize keeps no last-ACCEPTED-delivery read a box or console can ask (its CallbackAdmin.ListRefusals lists refused callbacks only)";
/// The number of log lines a request asks for, defaulted and capped.
pub fn log_lines(asked: u32) -> usize {
match asked as usize {
0 => DEFAULT_LOG_LINES,
n => n.min(MAX_LOG_LINES),
}
}
// ── Twin ────────────────────────────────────────────────────────────────────
/// The Twin pane (blocking: the credentials are files). `now_unix` is the
/// estate clock's seconds.
pub fn twin(
posture: &crate::twin::TwinPosture,
data_root: Option<&Path>,
grow: Option<&crate::grow::GrowPlane>,
now_unix: i64,
) -> pb::AdminTwin {
let role = posture.role();
let is_twin = role.is_twin();
let (complete, why, last_whole) = if is_twin {
posture.completeness()
} else {
(false, "this box is not a twin".to_string(), 0)
};
let (door_refused, door_last_refusal, door_admitted) = posture.door_lane();
let credentials = match data_root {
Some(root) => [
crate::twin_tls::Purpose::Twin,
crate::twin_tls::Purpose::Recharge,
]
.into_iter()
.map(|p| credential(root, p, now_unix))
.collect(),
None => Vec::new(),
};
let (twins, twins_error) = match grow {
Some(g) if !is_twin => match g.twins() {
Ok(r) => (
r.twins
.into_iter()
.map(|t| pb::AdminTwinRow {
principal: t.principal,
address: t.address,
state: t.state,
last_seen_unix_ms: t.last_seen_unix_ms,
lag_entries: t.lag_entries,
disk_total_bytes: t.disk_total_bytes,
disk_used_bytes: t.disk_used_bytes,
fill_permille: t.fill_permille,
applied_seq: t.applied_seq,
head_seq: t.head_seq,
pulls_today: t.pulls_today,
last_pull_unix: t.last_pull_unix,
})
.collect(),
String::new(),
),
Err(e) => (Vec::new(), e),
},
Some(_) => (Vec::new(), String::new()),
None => (
Vec::new(),
"this process holds no growth plane, so no twin roster".to_string(),
),
};
pb::AdminTwin {
role: if is_twin { "twin" } else { "primary" }.into(),
primary: role.primary().unwrap_or_default().into(),
serving: posture.serving(),
accepts_writes: !is_twin && posture.serving(),
awaiting_recharge_from: posture.awaiting_from().unwrap_or_default(),
follow: posture.follow(),
complete,
why,
last_whole_unix: last_whole,
lag_seconds: lag_seconds(is_twin, last_whole, now_unix),
store_warning: posture.verdict().warning().unwrap_or_default(),
door_refused,
door_last_refusal,
door_admitted: door_admitted.into_iter().collect(),
credentials,
twins,
twins_error,
}
}
/// Pure: the Sync pane's downstream rows — `GrowService.Downstreams`' rows
/// ([`crate::peers::PeerBoard::rows`]), field for field; nothing is judged here.
pub fn downstreams(
rows: Vec<crate::grpc::holger_grow_proto::DownstreamRow>,
) -> Vec<pb::AdminDownstreamRow> {
rows.into_iter()
.map(|d| pb::AdminDownstreamRow {
name: d.name,
fingerprint: d.fingerprint,
state: d.state,
stalled: d.stalled,
lag_entries: d.lag_entries,
last_seen_unix_ms: d.last_seen_unix_ms,
seen_ago_ms: d.seen_ago_ms,
every_secs: d.every_secs,
repositories: d
.repositories
.into_iter()
.map(|r| pb::AdminDownstreamRepo {
repository: r.repository,
state: r.state,
lag_entries: r.lag_entries,
applied_seq: r.applied_seq,
head_seq: r.head_seq,
last_complete_unix_ms: r.last_complete_unix_ms,
})
.collect(),
requests_today: d.requests_today,
bytes_today: d.bytes_today,
last_pull_unix: d.last_pull_unix,
})
.collect()
}
/// Pure: a twin's lag — now minus the end of its last whole follow cycle;
/// `-1` when it was never whole; `0` on a primary (it follows nobody).
pub fn lag_seconds(is_twin: bool, last_whole_unix: i64, now_unix: i64) -> i64 {
match (is_twin, last_whole_unix) {
(false, _) => 0,
(true, 0) => -1,
(true, t) => now_unix.saturating_sub(t).max(0),
}
}
pub(crate) fn credential(
root: &Path,
purpose: crate::twin_tls::Purpose,
now_unix: i64,
) -> pb::AdminTwinCredential {
let name = match purpose {
crate::twin_tls::Purpose::Twin => "twin",
crate::twin_tls::Purpose::Recharge => "recharge",
};
let mut c = pb::AdminTwinCredential {
purpose: name.into(),
..Default::default()
};
match crate::twin_tls::load(root, purpose) {
Ok(None) => {}
Err(e) => c.error = e,
Ok(Some(cred)) => installed(&mut c, &cred, now_unix),
}
c
}
// ── Growth ──────────────────────────────────────────────────────────────────
/// The Growth pane (blocking: the capacity ladder measures the machine).
pub fn growth(grow: Option<&crate::grow::GrowPlane>, data_root: Option<&Path>) -> pb::AdminGrowth {
let mut g = match grow {
None => pb::AdminGrowth {
present: false,
why: "this process holds no growth plane (a bare service)".into(),
..Default::default()
},
Some(plane) => {
let s = plane.status();
let j = s.journal.unwrap_or_default();
pb::AdminGrowth {
present: true,
phase: s.phase,
why: s.why,
epoch: s.epoch,
serving: s.serving,
engine_present: s.engine_present,
hotplug: s.hotplug,
in_flight: s.in_flight,
since_unix_ms: s.since_unix_ms,
detail: s.detail,
set_uuid: s.set_uuid,
total_bytes: s.total_bytes,
members: s
.members
.into_iter()
.map(|m| pb::AdminGrowMember {
index: m.index,
device: m.device,
partition: m.partition,
disk_bytes: m.disk_bytes,
set_bytes: m.set_bytes,
offered_bytes: m.offered_bytes,
larger: m.larger,
})
.collect(),
members_verdict: s.members_verdict,
journal_phase: j.phase,
journal_steps: j.steps,
journal_partial_members: j.partial_members,
journal_detail: j.detail,
journal_path: j.journal_path,
..Default::default()
}
}
};
if let Some(root) = data_root {
let (growth, reason, data_set, hotplug) = crate::grow::capacity_growth(root);
g.capacity_growth = growth;
g.capacity_growth_reason = reason;
g.data_set = data_set;
g.hotplug |= hotplug;
}
g
}
// ── Clock ───────────────────────────────────────────────────────────────────
/// The Clock pane: whether a verified answer stands behind holger's clock, what
/// is owed in-process, the configuration in force, and the last and boot syncs.
pub fn clock() -> pb::AdminClock {
let last = estate_clock::report::last();
let boot = estate_clock::report::boot();
pb::AdminClock {
trusted: crate::clock::is_trusted(),
owed_ms: crate::clock::owed_ms(),
config: config_words(&crate::clock::config_now()),
estate_unix_ms: crate::clock::now_unix_ms(),
system_unix_ms: estate_clock::unix_ms(std::time::SystemTime::now()),
has_last: last.is_some(),
last: last.as_ref().map(sync),
has_boot: boot.is_some(),
boot: boot.as_ref().map(sync),
}
}
/// Pure: the configuration in words — the sources and whether a peer key is
/// pinned (the key itself is public, and still not repeated here).
pub fn config_words(c: &Result<estate_clock::config::Config, String>) -> String {
use estate_clock::config::Config;
match c {
Err(why) => format!("REFUSED: {why}"),
Ok(Config::NotConfigured) => {
"not configured: holger's clock is the system clock, unverified".into()
}
Ok(Config::Off) => "off: this box was told not to follow a clock".into(),
Ok(Config::Sources { sources, peer_key }) => format!(
"{} ({})",
sources
.iter()
.map(|s| s.label())
.collect::<Vec<_>>()
.join(", "),
if peer_key.is_some() {
"peer key pinned"
} else {
"no peer key"
}
),
}
}
fn sync(r: &estate_clock::report::Report) -> pb::AdminClockSync {
pb::AdminClockSync {
trigger: r.trigger.as_str().into(),
at_unix_ms: r.at_unix_ms,
outcome: r.outcome.as_str().into(),
vouches: r.outcome.vouches(),
offset_ms: r.offset_ms,
detail: r.detail.clone(),
sources: r
.sources
.iter()
.map(|s| pb::AdminClockSource {
source: s.source.clone(),
answered: s.offset_ms.is_some(),
offset_ms: s.offset_ms.unwrap_or(0),
error: s.error.clone().unwrap_or_default(),
})
.collect(),
}
}
// ── Plugins / Monetize ──────────────────────────────────────────────────────
/// The Plugins pane: the boot's census verbatim, the meters, and — in the
/// metered build — the entitlement store and the self-registration.
pub fn plugins(g: &crate::grpc::HolgerGrpc) -> pb::AdminPlugins {
let census = &g.module_census;
#[allow(unused_mut)]
let mut p = pb::AdminPlugins {
modules: census
.modules
.iter()
.map(|m| pb::LoadedModule {
name: m.name.clone(),
origin: m.origin.clone(),
kind: m.kind.clone(),
digest: m.digest.clone(),
receipt: m.receipt.clone(),
granted: m.granted.clone(),
})
.collect(),
granted_capabilities: census.granted_capabilities.clone(),
metered_build: cfg!(feature = "metered"),
..Default::default()
};
#[cfg(feature = "meter-guests")]
{
p.meters = g.meters.len() as u32;
}
#[cfg(feature = "metered")]
{
if let Some(store) = &g.entitlements {
p.entitlements_configured = true;
p.entitlements = store
.list()
.into_iter()
.map(|f| pb::AdminEntitlement {
repository: f.tenant.0.clone(),
caps: f.caps.iter().map(|(k, v)| format!("{k}={v}")).collect(),
state: format!("{:?}", f.state).to_lowercase(),
paid_until_unix_ms: f.paid_until_unix_ms.unwrap_or(0),
source: f.source.clone(),
})
.collect();
}
}
p
}
// ── Disk speed ──────────────────────────────────────────────────────────────
/// The env seam: `HOLGER_DISK_SPEED=1` runs the start-up disk speed test on a
/// box that is not PID 1 (a developer's, a test rig's).
pub const SPEED_ENV: &str = "HOLGER_DISK_SPEED";
/// Why there is no speed figure when the test was never started here.
pub const SPEED_NOT_RUN: &str = "not measured: this process runs no disk speed test (it runs at start on an appliance's PID 1, or with HOLGER_DISK_SPEED=1)";
/// Why there is no speed figure while the test runs.
pub const SPEED_RUNNING: &str = "not measured yet: the start-up disk speed test is still running";
/// Whether [`SPEED_ENV`] asks for the test.
pub fn disk_speed_asked() -> bool {
std::env::var(SPEED_ENV)
.ok()
.is_some_and(|v| !v.is_empty() && v != "0")
}
// ── Tenants: usage, cache, quota ────────────────────────────────────────────
/// Why a repository has no cap figure.
pub const NO_CAP_OPEN: &str = "no cap: this is the open build, which caps nothing";
pub const NO_CAP_UNCONFIGURED: &str =
"no cap known: the entitlement store is not configured on this box";
pub const NO_CAP_NONE: &str = "no cap: no entitlement for this repository";
/// Pure: the paid `disk_bytes` cap in one entitlement's `name=value` caps.
pub fn cap_of(caps: &[String]) -> Option<u64> {
caps.iter().find_map(|c| {
c.strip_prefix(crate::metering::METER_DISK_BYTES)
.and_then(|r| r.strip_prefix('='))
.and_then(|v| v.trim().parse().ok())
})
}
/// Every repository's line (blocking: the meter walks the storage places).
/// `upstreams` names each repository's upstreams (empty = not a cache);
/// `plugins` is the Plugins pane (its entitlement store is the quota).
pub fn tenants(
places: &crate::metering::StoragePlaces,
upstreams: &std::collections::BTreeMap<String, Vec<String>>,
plugins: &pb::AdminPlugins,
) -> Vec<pb::AdminTenant> {
let names: std::collections::BTreeSet<&String> =
places.keys().chain(upstreams.keys()).collect();
names
.into_iter()
.map(|repo| {
let used = match places.get(repo) {
Some(p) => crate::metering::disk_bytes(p).map(|(n, _)| n),
None => Err("no storage place is declared for it".to_string()),
};
tenant_line(
repo,
upstreams.get(repo).cloned().unwrap_or_default(),
used,
plugins,
)
})
.collect()
}
/// Pure: one repository's line from what was measured.
pub fn tenant_line(
repo: &str,
ups: Vec<String>,
used: Result<u64, String>,
plugins: &pb::AdminPlugins,
) -> pb::AdminTenant {
let mut t = pb::AdminTenant {
repository: repo.to_string(),
cache: !ups.is_empty(),
upstreams: ups,
..Default::default()
};
match used {
Ok(n) => {
t.used_measured = true;
t.used_bytes = n;
}
Err(why) => t.used_why = why,
}
if !plugins.metered_build {
t.cap_why = NO_CAP_OPEN.into();
} else if !plugins.entitlements_configured {
t.cap_why = NO_CAP_UNCONFIGURED.into();
} else {
match plugins.entitlements.iter().find(|e| e.repository == repo) {
None => t.cap_why = NO_CAP_NONE.into(),
Some(e) => {
t.state = e.state.clone();
match cap_of(&e.caps) {
Some(c) => {
t.has_cap = true;
t.cap_bytes = c;
}
None => {
t.cap_why = format!(
"no cap: the entitlement names no {}",
crate::metering::METER_DISK_BYTES
)
}
}
}
}
}
t
}
// ── The job queue ───────────────────────────────────────────────────────────
/// How many of the newest jobs the depth is counted over.
pub const QUEUE_SCAN: usize = 10_000;
/// The job queue's depth (blocking: the redb ledger is read).
#[cfg(feature = "jobs")]
pub fn queue(q: Option<&crate::jobs::JobQueue>) -> pb::AdminQueue {
let Some(q) = q else {
return pb::AdminQueue {
present: false,
why: "this server carries no job queue (built without `jobs`, or not configured)"
.into(),
..Default::default()
};
};
match q.list(QUEUE_SCAN) {
Ok(views) => queue_from(views.iter().map(|v| v.status.as_str()), q.workers() as u32),
Err(e) => pb::AdminQueue {
present: true,
why: format!("the job ledger could not be read: {e}"),
workers: q.workers() as u32,
..Default::default()
},
}
}
#[cfg(not(feature = "jobs"))]
pub fn queue_absent() -> pb::AdminQueue {
pb::AdminQueue {
present: false,
why: "this server is built without the job queue (feature `jobs`)".into(),
..Default::default()
}
}
/// Pure: statuses → the depth. Open = every status that is not terminal.
#[cfg(feature = "jobs")]
pub fn queue_from<'a>(statuses: impl Iterator<Item = &'a str>, workers: u32) -> pb::AdminQueue {
let mut a = pb::AdminQueue {
present: true,
workers,
..Default::default()
};
for s in statuses {
a.scanned += 1;
*a.by_status.entry(s.to_string()).or_insert(0) += 1;
if !crate::jobs::status::is_terminal(s) {
a.open += 1;
}
}
a
}
// ── Audit ───────────────────────────────────────────────────────────────────
/// The Audit pane (blocking: the segment directory is scanned).
pub fn audit(log: &dyn crate::audit::AuditLog) -> pb::AdminAudit {
let mut a = pb::AdminAudit {
mode: log.mode().as_str().into(),
..Default::default()
};
match log.query_source() {
None => a.source = "none".into(),
Some(crate::audit::QuerySource::Snapshot(events)) => {
a.source = "memory".into();
a.total_rows = events.len() as u64;
}
Some(crate::audit::QuerySource::Directory(dir)) => {
a.source = format!("directory {}", dir.display());
a.total_bytes = std::fs::metadata(dir.join(crate::audit::AUDIT_FILE))
.map(|m| m.len())
.unwrap_or(0);
match crate::audit::FileAuditLog::read_dir(&dir) {
Ok(rows) => {
a.segments = u64::from(!rows.is_empty());
a.total_rows = rows.len() as u64;
a.first_ts_nanos = rows.first().map(|r| r.ts_nanos).unwrap_or(0);
a.last_ts_nanos = rows.last().map(|r| r.ts_nanos).unwrap_or(0);
}
Err(e) => a.error = e.to_string(),
}
}
}
a
}
// ── Logs ────────────────────────────────────────────────────────────────────
type LogDoor = Box<dyn Fn(usize) -> pb::AdminLogs + Send + Sync>;
static LOG_DOOR: OnceLock<LogDoor> = OnceLock::new();
/// **The process that owns the installer's boot log registers its reader
/// here** (holger `paired`, once, at start): `n` newest lines and the
/// door's state. The library does not own that log, so it cannot read it any
/// other way.
pub fn set_log_door(f: impl Fn(usize) -> pb::AdminLogs + Send + Sync + 'static) {
let _ = LOG_DOOR.set(Box::new(f));
}
/// The Logs pane: the registered reader, or `present: false` with the reason.
pub fn logs(n: usize) -> pb::AdminLogs {
match LOG_DOOR.get() {
Some(f) => f(n),
None => pb::AdminLogs {
present: false,
door: "none: this process carries no installer log (it is not an installer-built box)"
.into(),
..Default::default()
},
}
}
// ── The header ──────────────────────────────────────────────────────────────
/// This machine's host name (the kernel's), or empty.
pub fn host() -> String {
std::fs::read_to_string("/proc/sys/kernel/hostname")
.map(|s| s.trim().to_string())
.unwrap_or_default()
}
/// The payments pane: where the Stripe webhook's last delivery is not measured. An open
/// build takes no payment, so it has no pane (A28 M13).
pub fn payments() -> Option<pb::AdminPayments> {
#[cfg(feature = "metered")]
return Some(pb::AdminPayments {
stripe_last_delivery_unix_ms: 0,
note: STRIPE_NOTE.into(),
});
#[cfg(not(feature = "metered"))]
None
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_log_window_is_defaulted_and_capped() {
assert_eq!(log_lines(0), DEFAULT_LOG_LINES);
assert_eq!(log_lines(7), 7);
assert_eq!(log_lines(1_000_000), MAX_LOG_LINES);
}
#[test]
fn a_twins_lag_is_since_its_last_whole_cycle_and_never_is_minus_one() {
assert_eq!(lag_seconds(true, 1_000, 1_090), 90);
assert_eq!(lag_seconds(true, 0, 1_090), -1, "never whole");
assert_eq!(lag_seconds(false, 0, 1_090), 0, "a primary follows nobody");
assert_eq!(
lag_seconds(true, 2_000, 1_000),
0,
"a clock step back is not negative lag"
);
}
#[test]
fn the_clock_configuration_is_said_in_words() {
use estate_clock::config::Config;
assert!(config_words(&Ok(Config::NotConfigured)).contains("unverified"));
assert!(config_words(&Ok(Config::Off)).starts_with("off"));
assert!(config_words(&Err("medium and door disagree".into())).starts_with("REFUSED"));
let c = Config::Sources {
sources: vec![estate_clock::config::Source::Peer(
"10.13.8.1:50123".parse().unwrap(),
)],
peer_key: Some("ssh-ed25519 AAAA".into()),
};
assert_eq!(
config_words(&Ok(c)),
"peer://10.13.8.1:50123 (peer key pinned)"
);
}
#[test]
fn with_no_log_door_registered_the_pane_says_why() {
let l = logs(10);
// A test process registers none (only the CLI's paired start does).
assert!(!l.present);
assert!(l.door.contains("not an installer-built box"));
}
/// ★ **Usage, cache and quota per repository**: a figure or its reason.
#[test]
fn a_tenant_line_says_its_bytes_whether_they_are_a_cache_and_its_cap_or_why_not() {
assert_eq!(
cap_of(&["disk_bytes=10737418240".into(), "artifacts=5".into()]),
Some(10_737_418_240)
);
assert_eq!(cap_of(&["artifacts=5".into()]), None);
let open = pb::AdminPlugins::default();
let t = tenant_line(
"crates",
vec!["https://index.crates.io".into()],
Ok(42),
&open,
);
assert!(t.cache && t.used_measured && !t.has_cap);
assert_eq!((t.used_bytes, t.cap_why.as_str()), (42, NO_CAP_OPEN));
let unmeasured = tenant_line("all", vec![], Err("a group owns no storage".into()), &open);
assert!(!unmeasured.cache && !unmeasured.used_measured);
assert_eq!(
(unmeasured.used_bytes, unmeasured.used_why.as_str()),
(0, "a group owns no storage")
);
let mut metered = pb::AdminPlugins {
metered_build: true,
..Default::default()
};
assert_eq!(
tenant_line("bundles", vec![], Ok(1), &metered).cap_why,
NO_CAP_UNCONFIGURED
);
metered.entitlements_configured = true;
assert_eq!(
tenant_line("bundles", vec![], Ok(1), &metered).cap_why,
NO_CAP_NONE
);
metered.entitlements.push(pb::AdminEntitlement {
repository: "bundles".into(),
caps: vec!["disk_bytes=5368709120".into()],
state: "active".into(),
..Default::default()
});
let paid = tenant_line("bundles", vec![], Ok(1), &metered);
assert!(paid.has_cap && paid.cap_why.is_empty());
assert_eq!(
(paid.cap_bytes, paid.state.as_str()),
(5_368_709_120, "active")
);
}
/// The queue's depth is counted by status; open = not terminal.
#[cfg(feature = "jobs")]
#[test]
fn the_queue_depth_counts_every_status_and_open_is_the_unfinished() {
let q = queue_from(
[
"queued",
"running",
"done",
"failed",
"retry_wait",
"queued",
"interrupted",
]
.into_iter(),
4,
);
assert!(q.present);
assert_eq!((q.scanned, q.open, q.workers), (7, 4, 4));
assert_eq!(q.by_status.get("queued"), Some(&2));
let none = queue(None);
assert!(!none.present && !none.why.is_empty());
}
}