plugins/native/znippy-plugin-git/src/progress.rs
//! **Per-pack index progress, published through a cheap shared counter.**
//!
//! The absorber ([`crate::git_ops`]) owns one [`IndexProgress`] per store and
//! bumps its atomics as it goes — one `fetch_add` per resolved object, nothing
//! locked, nothing formatted. A reader (gunnar's dashboard tick, `Admin.Stats`)
//! takes a [`Snapshot`] of every live counter through [`snapshot_all`], which
//! is what the row `index alice/linux 34% 4.7M/13.9M resolve 2.1 GB/s` is
//! drawn from.
//!
//! # Surviving a restart
//!
//! The counter is process memory and does not survive one; the **work** does
//! (the journal re-queues an unabsorbed pack on open). A pack that is being
//! absorbed because the journal said so — rather than because a push in this
//! process queued it — is marked `resumed`, so after a reboot the row reads
//! `index alice/linux resumed 0%` and an operator knows what the box is doing
//! and why it started from zero.
//!
//! # The registry
//!
//! A process-wide list of `Weak<IndexProgress>`: a store registers on open and
//! its entry dies with it. Dead entries are swept on every snapshot, so the
//! list is O(open stores) and no store has to remember to deregister.
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicU8, Ordering};
use std::sync::{Arc, Mutex, OnceLock, Weak};
use std::time::{SystemTime, UNIX_EPOCH};
/// What the worker is doing to the pack right now.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum Phase {
Idle = 0,
/// Walking the pack's entries (sizes, chain shape) before a byte is inflated.
Walk = 1,
/// Inflate, delta-resolve, hash — objects tick up here.
Resolve = 2,
/// Paused by the memory watchdog ([`crate::memguard`]).
Paused = 3,
/// The exploded tier and index rows are being written down.
Flush = 4,
/// The derived tables (graph, generations) are being re-folded.
Fold = 5,
Done = 6,
}
impl Phase {
pub fn as_str(self) -> &'static str {
match self {
Phase::Idle => "idle",
Phase::Walk => "walk",
Phase::Resolve => "resolve",
Phase::Paused => "paused",
Phase::Flush => "flush",
Phase::Fold => "fold",
Phase::Done => "done",
}
}
fn from_u8(v: u8) -> Phase {
match v {
1 => Phase::Walk,
2 => Phase::Resolve,
3 => Phase::Paused,
4 => Phase::Flush,
5 => Phase::Fold,
6 => Phase::Done,
_ => Phase::Idle,
}
}
}
/// The counter. Every field is an atomic so the hot path pays one relaxed add.
pub struct IndexProgress {
/// The store's root directory — what a reader maps to a repository name.
root: String,
active: AtomicBool,
resumed: AtomicBool,
pack_id: AtomicU64,
objects_total: AtomicU64,
objects_done: AtomicU64,
/// Pack bytes covered by the resolved entries so far.
bytes_done: AtomicU64,
bytes_total: AtomicU64,
phase: AtomicU8,
started_at_ms: AtomicU64,
/// Bases re-inflated after an eviction, this pack.
reinflations: AtomicU64,
/// Watchdog pauses, this pack.
pauses: AtomicU64,
/// Workers that parked to give their core back on a short box (the
/// resolver's shed), since open. Not in [`Snapshot`], so adding it changed
/// no public struct: read it with [`IndexProgress::sheds`].
sheds: AtomicU64,
/// Packs finished by this counter since open.
packs_done: AtomicU64,
/// The process's `VmHWM` as of the last tick — see [`IndexProgress::sample_memory`].
peak_rss: AtomicU64,
/// `MemAvailable` as of the last tick.
mem_available: AtomicU64,
}
/// One reader's copy of a counter.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Snapshot {
pub root: String,
pub active: bool,
pub resumed: bool,
pub pack_id: u64,
pub objects_total: u64,
pub objects_done: u64,
pub bytes_done: u64,
pub bytes_total: u64,
pub phase: Phase,
pub started_at_ms: u64,
pub reinflations: u64,
pub pauses: u64,
pub packs_done: u64,
/// **Peak resident set of the whole process**, sampled on the progress
/// tick. It rode only on `index.finished` until 2026-08-30, which meant it
/// was unobservable in exactly the failure it was built for: lane P2's two
/// rig boxes never finished a pack, so the one number that would have said
/// how close to the ceiling they were was never printed. `0` = not sampled
/// yet.
pub peak_rss: u64,
/// `MemAvailable` at the same tick. The pair is what says "is it coming
/// back" without waiting for a row that may never come.
pub mem_available: u64,
}
impl Snapshot {
/// Whole percent, `0` when the total is unknown.
pub fn percent(&self) -> u64 {
if self.objects_total == 0 {
0
} else {
(self.objects_done * 100 / self.objects_total).min(100)
}
}
}
fn now_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0)
}
impl IndexProgress {
/// A counter for the store rooted at `root`, registered process-wide.
pub fn new(root: &str) -> Arc<Self> {
let p = Arc::new(Self {
root: root.to_string(),
active: AtomicBool::new(false),
resumed: AtomicBool::new(false),
pack_id: AtomicU64::new(0),
objects_total: AtomicU64::new(0),
objects_done: AtomicU64::new(0),
bytes_done: AtomicU64::new(0),
bytes_total: AtomicU64::new(0),
phase: AtomicU8::new(Phase::Idle as u8),
started_at_ms: AtomicU64::new(0),
reinflations: AtomicU64::new(0),
pauses: AtomicU64::new(0),
sheds: AtomicU64::new(0),
packs_done: AtomicU64::new(0),
peak_rss: AtomicU64::new(0),
mem_available: AtomicU64::new(0),
});
registry().lock().unwrap_or_else(|e| e.into_inner()).push(Arc::downgrade(&p));
p
}
/// A counter nobody reads, for a caller with no store behind it.
pub fn detached() -> Arc<Self> {
let p = Self::new("");
p
}
pub fn root(&self) -> &str {
&self.root
}
/// A pack begins. `resumed` = the journal queued it, not a push in this process.
pub fn begin(&self, pack_id: u64, bytes_total: u64, resumed: bool) {
self.pack_id.store(pack_id, Ordering::Relaxed);
self.bytes_total.store(bytes_total, Ordering::Relaxed);
self.bytes_done.store(0, Ordering::Relaxed);
self.objects_total.store(0, Ordering::Relaxed);
self.objects_done.store(0, Ordering::Relaxed);
self.reinflations.store(0, Ordering::Relaxed);
self.pauses.store(0, Ordering::Relaxed);
self.resumed.store(resumed, Ordering::Relaxed);
self.started_at_ms.store(now_ms(), Ordering::Relaxed);
self.phase.store(Phase::Walk as u8, Ordering::Relaxed);
self.active.store(true, Ordering::Release);
}
/// The walk is done and the object count is known.
pub fn set_total(&self, objects: u64) {
self.objects_total.store(objects, Ordering::Relaxed);
}
pub fn set_phase(&self, phase: Phase) {
self.phase.store(phase as u8, Ordering::Relaxed);
}
/// **The hot-path call**: one object resolved, covering `bytes` of the pack.
#[inline]
pub fn object_done(&self, bytes: u64) {
self.objects_done.fetch_add(1, Ordering::Relaxed);
self.bytes_done.fetch_add(bytes, Ordering::Relaxed);
}
pub fn add_reinflations(&self, n: u64) {
self.reinflations.fetch_add(n, Ordering::Relaxed);
}
pub fn add_pause(&self) {
self.pauses.fetch_add(1, Ordering::Relaxed);
self.sample_memory();
}
/// A worker parked to shed its core (see `resolve`'s checkpoint).
pub fn add_shed(&self) {
self.sheds.fetch_add(1, Ordering::Relaxed);
}
/// Sheds recorded since this counter was opened.
pub fn sheds(&self) -> u64 {
self.sheds.load(Ordering::Relaxed)
}
/// **Read `/proc` once and publish it on the counter.**
///
/// Two file reads, so it is a tick's work and never the hot path's: the
/// ticker calls it every [`TICK`], `add_pause` calls it when a pause opens,
/// and the absorber calls it when a pack ends. Everything else reads the
/// published value out of a [`Snapshot`].
pub fn sample_memory(&self) {
self.peak_rss
.store(crate::memguard::peak_rss_bytes(), Ordering::Relaxed);
self.mem_available.store(
crate::memguard::global().available().unwrap_or(0),
Ordering::Relaxed,
);
}
/// Objects per second since `begin`, or 0 before the clock has moved.
pub fn rate(&self) -> u64 {
let s = self.snapshot();
let started = s.started_at_ms;
let now = now_ms();
if started == 0 || now <= started {
return 0;
}
s.objects_done * 1000 / (now - started)
}
/// The pack is absorbed (or abandoned): the row goes idle.
pub fn finish(&self) {
self.packs_done.fetch_add(1, Ordering::Relaxed);
self.phase.store(Phase::Done as u8, Ordering::Relaxed);
self.active.store(false, Ordering::Release);
}
pub fn snapshot(&self) -> Snapshot {
Snapshot {
root: self.root.clone(),
active: self.active.load(Ordering::Acquire),
resumed: self.resumed.load(Ordering::Relaxed),
pack_id: self.pack_id.load(Ordering::Relaxed),
objects_total: self.objects_total.load(Ordering::Relaxed),
objects_done: self.objects_done.load(Ordering::Relaxed),
bytes_done: self.bytes_done.load(Ordering::Relaxed),
bytes_total: self.bytes_total.load(Ordering::Relaxed),
phase: Phase::from_u8(self.phase.load(Ordering::Relaxed)),
started_at_ms: self.started_at_ms.load(Ordering::Relaxed),
reinflations: self.reinflations.load(Ordering::Relaxed),
pauses: self.pauses.load(Ordering::Relaxed),
packs_done: self.packs_done.load(Ordering::Relaxed),
peak_rss: self.peak_rss.load(Ordering::Relaxed),
mem_available: self.mem_available.load(Ordering::Relaxed),
}
}
}
/// How often the live progress row is logged while a pack is being absorbed.
pub const TICK: std::time::Duration = std::time::Duration::from_secs(10);
/// **A live progress row on the log, for as long as a pack is being absorbed.**
///
/// # Why this exists
///
/// `peak_rss` and the re-inflation count rode only on `index.finished`, and a
/// pack that never finishes emits neither. Lane P2's two rig boxes ran for
/// 2h14m and 2h12m and produced no row carrying either number — the two figures
/// most wanted, missing from exactly the failure they were built for. The band
/// could show 13G/16G because a dashboard polls [`snapshot_all`]; a serial log
/// and a `journalctl` had nothing.
///
/// So the numbers ride the tick as well as the finish, and the tick is a thread
/// rather than a hook inside the resolver: it prints while the resolver is
/// PAUSED too, which is when an operator most needs to see it, and a hook in
/// the resolve loop by definition does not run then.
pub struct ProgressTicker {
stop: Arc<AtomicBool>,
handle: Option<std::thread::JoinHandle<()>>,
}
impl ProgressTicker {
/// Start ticking for `progress`. The thread ends when the returned value is
/// dropped, which is why it is a value and not a call.
pub fn start(progress: Arc<IndexProgress>) -> Self {
let stop = Arc::new(AtomicBool::new(false));
let s2 = stop.clone();
let handle = std::thread::Builder::new()
.name("znippy-index-tick".into())
.spawn(move || {
let started = std::time::Instant::now();
let mut next = std::time::Instant::now() + TICK;
while !s2.load(Ordering::Relaxed) {
// **Park until the next tick, wakeable.** `sleep` cannot be
// interrupted, so a `drop` that joined a ticker mid-sleep
// blocked the absorb for up to the sleep's length — a flat
// ~250 ms tax on *every* push, dwarfing the fold at these
// sizes (measured: a 24-object delta absorb was 250 ms, of
// which the fold was under 20 ms even at 8 k commits). Park
// instead: `drop` sets `stop` and unparks, so a finished
// pack's ticker goes away at once. `park_timeout` may wake
// spuriously; the loop re-parks the remaining time and the
// 10 s cadence below is unchanged.
let now = std::time::Instant::now();
if now < next {
std::thread::park_timeout(next - now);
continue;
}
next = now + TICK;
progress.sample_memory();
let s = progress.snapshot();
if !s.active {
continue;
}
let secs = started.elapsed().as_secs_f64().max(1e-3);
tracing::info!(
target: "gunnar::index",
event = "index.progress",
repo = %s.root,
pack_id = s.pack_id,
phase = s.phase.as_str(),
percent = s.percent(),
objects_done = s.objects_done,
objects_total = s.objects_total,
bytes_done = s.bytes_done,
bytes_total = s.bytes_total,
objects_per_s = (s.objects_done as f64 / secs) as u64,
reinflations = s.reinflations,
pauses = s.pauses,
peak_rss = s.peak_rss,
mem_available = s.mem_available,
elapsed_ms = started.elapsed().as_millis() as u64,
"index {} {}% {}/{} {} {}/s; {} re-inflations, {} pauses, peak rss {}, {} available",
s.root,
s.percent(),
s.objects_done,
s.objects_total,
s.phase.as_str(),
(s.objects_done as f64 / secs) as u64,
s.reinflations,
s.pauses,
crate::memguard::fmt_bytes(s.peak_rss),
crate::memguard::fmt_bytes(s.mem_available)
);
}
})
.ok();
ProgressTicker { stop, handle }
}
}
impl Drop for ProgressTicker {
fn drop(&mut self) {
self.stop.store(true, Ordering::Relaxed);
if let Some(h) = self.handle.take() {
// Wake a parked ticker so the join returns at once instead of
// waiting out its current `park_timeout`. `unpark` before the
// thread parks is not lost — the token is saved and the next park
// returns immediately — so `stop` is always observed.
h.thread().unpark();
let _ = h.join();
}
}
}
fn registry() -> &'static Mutex<Vec<Weak<IndexProgress>>> {
static R: OnceLock<Mutex<Vec<Weak<IndexProgress>>>> = OnceLock::new();
R.get_or_init(|| Mutex::new(Vec::new()))
}
/// Every live counter, dead entries swept. Cheap: O(open stores), a few atomics
/// each. `active_only` keeps the rows a dashboard shows; the full list is what
/// a test or a stats call wants.
pub fn snapshot_all(active_only: bool) -> Vec<Snapshot> {
let mut r = registry().lock().unwrap_or_else(|e| e.into_inner());
r.retain(|w| w.strong_count() > 0);
r.iter()
.filter_map(|w| w.upgrade())
.map(|p| p.snapshot())
.filter(|s| !active_only || s.active)
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
/// **Dropping a ticker must not block the caller.**
///
/// The ticker's wait used to be an uninterruptible `sleep(250 ms)`, so a
/// `drop` that joined it mid-sleep waited the sleep out — a flat ~250 ms tax
/// on *every* pack absorb (`absorb_one` wraps its work in a ticker), which a
/// 24-object push measured in full: 250 ms for the ticker, under 20 ms for
/// everything else. Seen RED by restoring the `sleep` loop: this asserts
/// well under one tick's slowest step (250 ms) that the caller is freed at
/// once, so the guard is not sitting on the identity value of a fast box.
#[test]
fn dropping_the_ticker_does_not_block_the_caller() {
let p = IndexProgress::new("/tmp/x/alice/tick-drop.git/gunnar-git.znippy");
p.begin(1, 1 << 20, false);
let ticker = ProgressTicker::start(p.clone());
// Let the thread reach its park.
std::thread::sleep(std::time::Duration::from_millis(20));
let t = std::time::Instant::now();
drop(ticker);
let waited = t.elapsed();
assert!(
waited < std::time::Duration::from_millis(50),
"drop of a ticker blocked the caller for {waited:?} — a parked ticker must be \
unparked and joined at once, not waited out"
);
}
#[test]
fn a_counter_reports_what_the_worker_did_and_percent_is_applied_output() {
let p = IndexProgress::new("/tmp/x/alice/linux.git/gunnar-git.znippy");
assert!(!p.snapshot().active);
p.begin(7, 8 << 30, true);
p.set_total(13_948_244);
p.set_phase(Phase::Resolve);
for _ in 0..4_700_000 {
p.object_done(600);
}
let s = p.snapshot();
assert!(s.active && s.resumed);
assert_eq!(s.pack_id, 7);
assert_eq!(s.objects_done, 4_700_000);
assert_eq!(s.percent(), 33);
assert_eq!(s.bytes_done, 4_700_000 * 600);
assert_eq!(s.phase, Phase::Resolve);
assert!(s.started_at_ms > 0);
p.finish();
let s = p.snapshot();
assert!(!s.active);
assert_eq!(s.phase, Phase::Done);
assert_eq!(s.packs_done, 1);
}
#[test]
fn the_registry_lists_live_counters_and_forgets_dropped_ones() {
let root = format!("/reg-test-{}-a", std::process::id());
let a = IndexProgress::new(&root);
a.begin(1, 10, false);
let b = IndexProgress::new(&format!("{root}-b"));
let all = snapshot_all(false);
assert!(all.iter().any(|s| s.root == root));
assert!(all.iter().any(|s| s.root == format!("{root}-b")));
let active = snapshot_all(true);
assert!(active.iter().any(|s| s.root == root));
assert!(!active.iter().any(|s| s.root == format!("{root}-b")), "b never began");
drop(b);
assert!(!snapshot_all(false).iter().any(|s| s.root == format!("{root}-b")), "swept on drop");
assert_eq!(Phase::from_u8(2), Phase::Resolve);
assert_eq!(Phase::Paused.as_str(), "paused");
}
}