znippy_plugin_git/
progress.rs

1//! **Per-pack index progress, published through a cheap shared counter.**
2//!
3//! The absorber ([`crate::git_ops`]) owns one [`IndexProgress`] per store and
4//! bumps its atomics as it goes — one `fetch_add` per resolved object, nothing
5//! locked, nothing formatted. A reader (gunnar's dashboard tick, `Admin.Stats`)
6//! takes a [`Snapshot`] of every live counter through [`snapshot_all`], which
7//! is what the row `index rickard/linux 34% 4.7M/13.9M resolve 2.1 GB/s` is
8//! drawn from.
9//!
10//! # Surviving a restart
11//!
12//! The counter is process memory and does not survive one; the **work** does
13//! (the journal re-queues an unabsorbed pack on open). A pack that is being
14//! absorbed because the journal said so — rather than because a push in this
15//! process queued it — is marked `resumed`, so after a reboot the row reads
16//! `index rickard/linux resumed 0%` and an operator knows what the box is doing
17//! and why it started from zero.
18//!
19//! # The registry
20//!
21//! A process-wide list of `Weak<IndexProgress>`: a store registers on open and
22//! its entry dies with it. Dead entries are swept on every snapshot, so the
23//! list is O(open stores) and no store has to remember to deregister.
24
25use std::sync::atomic::{AtomicBool, AtomicU64, AtomicU8, Ordering};
26use std::sync::{Arc, Mutex, OnceLock, Weak};
27use std::time::{SystemTime, UNIX_EPOCH};
28
29/// What the worker is doing to the pack right now.
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31#[repr(u8)]
32pub enum Phase {
33    Idle = 0,
34    /// Walking the pack's entries (sizes, chain shape) before a byte is inflated.
35    Walk = 1,
36    /// Inflate, delta-resolve, hash — objects tick up here.
37    Resolve = 2,
38    /// Paused by the memory watchdog ([`crate::memguard`]).
39    Paused = 3,
40    /// The exploded tier and index rows are being written down.
41    Flush = 4,
42    /// The derived tables (graph, generations) are being re-folded.
43    Fold = 5,
44    Done = 6,
45}
46
47impl Phase {
48    pub fn as_str(self) -> &'static str {
49        match self {
50            Phase::Idle => "idle",
51            Phase::Walk => "walk",
52            Phase::Resolve => "resolve",
53            Phase::Paused => "paused",
54            Phase::Flush => "flush",
55            Phase::Fold => "fold",
56            Phase::Done => "done",
57        }
58    }
59    fn from_u8(v: u8) -> Phase {
60        match v {
61            1 => Phase::Walk,
62            2 => Phase::Resolve,
63            3 => Phase::Paused,
64            4 => Phase::Flush,
65            5 => Phase::Fold,
66            6 => Phase::Done,
67            _ => Phase::Idle,
68        }
69    }
70}
71
72/// The counter. Every field is an atomic so the hot path pays one relaxed add.
73pub struct IndexProgress {
74    /// The store's root directory — what a reader maps to a repository name.
75    root: String,
76    active: AtomicBool,
77    resumed: AtomicBool,
78    pack_id: AtomicU64,
79    objects_total: AtomicU64,
80    objects_done: AtomicU64,
81    /// Pack bytes covered by the resolved entries so far.
82    bytes_done: AtomicU64,
83    bytes_total: AtomicU64,
84    phase: AtomicU8,
85    started_at_ms: AtomicU64,
86    /// Bases re-inflated after an eviction, this pack.
87    reinflations: AtomicU64,
88    /// Watchdog pauses, this pack.
89    pauses: AtomicU64,
90    /// Packs finished by this counter since open.
91    packs_done: AtomicU64,
92    /// The process's `VmHWM` as of the last tick — see [`IndexProgress::sample_memory`].
93    peak_rss: AtomicU64,
94    /// `MemAvailable` as of the last tick.
95    mem_available: AtomicU64,
96}
97
98/// One reader's copy of a counter.
99#[derive(Debug, Clone, PartialEq, Eq)]
100pub struct Snapshot {
101    pub root: String,
102    pub active: bool,
103    pub resumed: bool,
104    pub pack_id: u64,
105    pub objects_total: u64,
106    pub objects_done: u64,
107    pub bytes_done: u64,
108    pub bytes_total: u64,
109    pub phase: Phase,
110    pub started_at_ms: u64,
111    pub reinflations: u64,
112    pub pauses: u64,
113    pub packs_done: u64,
114    /// **Peak resident set of the whole process**, sampled on the progress
115    /// tick. It rode only on `index.finished` until 2026-08-30, which meant it
116    /// was unobservable in exactly the failure it was built for: lane P2's two
117    /// rig boxes never finished a pack, so the one number that would have said
118    /// how close to the ceiling they were was never printed. `0` = not sampled
119    /// yet.
120    pub peak_rss: u64,
121    /// `MemAvailable` at the same tick. The pair is what says "is it coming
122    /// back" without waiting for a row that may never come.
123    pub mem_available: u64,
124}
125
126impl Snapshot {
127    /// Whole percent, `0` when the total is unknown.
128    pub fn percent(&self) -> u64 {
129        if self.objects_total == 0 {
130            0
131        } else {
132            (self.objects_done * 100 / self.objects_total).min(100)
133        }
134    }
135}
136
137fn now_ms() -> u64 {
138    SystemTime::now()
139        .duration_since(UNIX_EPOCH)
140        .map(|d| d.as_millis() as u64)
141        .unwrap_or(0)
142}
143
144impl IndexProgress {
145    /// A counter for the store rooted at `root`, registered process-wide.
146    pub fn new(root: &str) -> Arc<Self> {
147        let p = Arc::new(Self {
148            root: root.to_string(),
149            active: AtomicBool::new(false),
150            resumed: AtomicBool::new(false),
151            pack_id: AtomicU64::new(0),
152            objects_total: AtomicU64::new(0),
153            objects_done: AtomicU64::new(0),
154            bytes_done: AtomicU64::new(0),
155            bytes_total: AtomicU64::new(0),
156            phase: AtomicU8::new(Phase::Idle as u8),
157            started_at_ms: AtomicU64::new(0),
158            reinflations: AtomicU64::new(0),
159            pauses: AtomicU64::new(0),
160            packs_done: AtomicU64::new(0),
161            peak_rss: AtomicU64::new(0),
162            mem_available: AtomicU64::new(0),
163        });
164        registry().lock().unwrap_or_else(|e| e.into_inner()).push(Arc::downgrade(&p));
165        p
166    }
167
168    /// A counter nobody reads, for a caller with no store behind it.
169    pub fn detached() -> Arc<Self> {
170        let p = Self::new("");
171        p
172    }
173
174    pub fn root(&self) -> &str {
175        &self.root
176    }
177
178    /// A pack begins. `resumed` = the journal queued it, not a push in this process.
179    pub fn begin(&self, pack_id: u64, bytes_total: u64, resumed: bool) {
180        self.pack_id.store(pack_id, Ordering::Relaxed);
181        self.bytes_total.store(bytes_total, Ordering::Relaxed);
182        self.bytes_done.store(0, Ordering::Relaxed);
183        self.objects_total.store(0, Ordering::Relaxed);
184        self.objects_done.store(0, Ordering::Relaxed);
185        self.reinflations.store(0, Ordering::Relaxed);
186        self.pauses.store(0, Ordering::Relaxed);
187        self.resumed.store(resumed, Ordering::Relaxed);
188        self.started_at_ms.store(now_ms(), Ordering::Relaxed);
189        self.phase.store(Phase::Walk as u8, Ordering::Relaxed);
190        self.active.store(true, Ordering::Release);
191    }
192
193    /// The walk is done and the object count is known.
194    pub fn set_total(&self, objects: u64) {
195        self.objects_total.store(objects, Ordering::Relaxed);
196    }
197
198    pub fn set_phase(&self, phase: Phase) {
199        self.phase.store(phase as u8, Ordering::Relaxed);
200    }
201
202    /// **The hot-path call**: one object resolved, covering `bytes` of the pack.
203    #[inline]
204    pub fn object_done(&self, bytes: u64) {
205        self.objects_done.fetch_add(1, Ordering::Relaxed);
206        self.bytes_done.fetch_add(bytes, Ordering::Relaxed);
207    }
208
209    pub fn add_reinflations(&self, n: u64) {
210        self.reinflations.fetch_add(n, Ordering::Relaxed);
211    }
212
213    pub fn add_pause(&self) {
214        self.pauses.fetch_add(1, Ordering::Relaxed);
215        self.sample_memory();
216    }
217
218    /// **Read `/proc` once and publish it on the counter.**
219    ///
220    /// Two file reads, so it is a tick's work and never the hot path's: the
221    /// ticker calls it every [`TICK`], `add_pause` calls it when a pause opens,
222    /// and the absorber calls it when a pack ends. Everything else reads the
223    /// published value out of a [`Snapshot`].
224    pub fn sample_memory(&self) {
225        self.peak_rss
226            .store(crate::memguard::peak_rss_bytes(), Ordering::Relaxed);
227        self.mem_available.store(
228            crate::memguard::global().available().unwrap_or(0),
229            Ordering::Relaxed,
230        );
231    }
232
233    /// Objects per second since `begin`, or 0 before the clock has moved.
234    pub fn rate(&self) -> u64 {
235        let s = self.snapshot();
236        let started = s.started_at_ms;
237        let now = now_ms();
238        if started == 0 || now <= started {
239            return 0;
240        }
241        s.objects_done * 1000 / (now - started)
242    }
243
244    /// The pack is absorbed (or abandoned): the row goes idle.
245    pub fn finish(&self) {
246        self.packs_done.fetch_add(1, Ordering::Relaxed);
247        self.phase.store(Phase::Done as u8, Ordering::Relaxed);
248        self.active.store(false, Ordering::Release);
249    }
250
251    pub fn snapshot(&self) -> Snapshot {
252        Snapshot {
253            root: self.root.clone(),
254            active: self.active.load(Ordering::Acquire),
255            resumed: self.resumed.load(Ordering::Relaxed),
256            pack_id: self.pack_id.load(Ordering::Relaxed),
257            objects_total: self.objects_total.load(Ordering::Relaxed),
258            objects_done: self.objects_done.load(Ordering::Relaxed),
259            bytes_done: self.bytes_done.load(Ordering::Relaxed),
260            bytes_total: self.bytes_total.load(Ordering::Relaxed),
261            phase: Phase::from_u8(self.phase.load(Ordering::Relaxed)),
262            started_at_ms: self.started_at_ms.load(Ordering::Relaxed),
263            reinflations: self.reinflations.load(Ordering::Relaxed),
264            pauses: self.pauses.load(Ordering::Relaxed),
265            packs_done: self.packs_done.load(Ordering::Relaxed),
266            peak_rss: self.peak_rss.load(Ordering::Relaxed),
267            mem_available: self.mem_available.load(Ordering::Relaxed),
268        }
269    }
270}
271
272/// How often the live progress row is logged while a pack is being absorbed.
273pub const TICK: std::time::Duration = std::time::Duration::from_secs(10);
274
275/// **A live progress row on the log, for as long as a pack is being absorbed.**
276///
277/// # Why this exists
278///
279/// `peak_rss` and the re-inflation count rode only on `index.finished`, and a
280/// pack that never finishes emits neither. Lane P2's two rig boxes ran for
281/// 2h14m and 2h12m and produced no row carrying either number — the two figures
282/// most wanted, missing from exactly the failure they were built for. The band
283/// could show 13G/16G because a dashboard polls [`snapshot_all`]; a serial log
284/// and a `journalctl` had nothing.
285///
286/// So the numbers ride the tick as well as the finish, and the tick is a thread
287/// rather than a hook inside the resolver: it prints while the resolver is
288/// PAUSED too, which is when an operator most needs to see it, and a hook in
289/// the resolve loop by definition does not run then.
290pub struct ProgressTicker {
291    stop: Arc<AtomicBool>,
292    handle: Option<std::thread::JoinHandle<()>>,
293}
294
295impl ProgressTicker {
296    /// Start ticking for `progress`. The thread ends when the returned value is
297    /// dropped, which is why it is a value and not a call.
298    pub fn start(progress: Arc<IndexProgress>) -> Self {
299        let stop = Arc::new(AtomicBool::new(false));
300        let s2 = stop.clone();
301        let handle = std::thread::Builder::new()
302            .name("znippy-index-tick".into())
303            .spawn(move || {
304                let started = std::time::Instant::now();
305                let mut next = std::time::Instant::now() + TICK;
306                while !s2.load(Ordering::Relaxed) {
307                    // **Park until the next tick, wakeable.** `sleep` cannot be
308                    // interrupted, so a `drop` that joined a ticker mid-sleep
309                    // blocked the absorb for up to the sleep's length — a flat
310                    // ~250 ms tax on *every* push, dwarfing the fold at these
311                    // sizes (measured: a 24-object delta absorb was 250 ms, of
312                    // which the fold was under 20 ms even at 8 k commits). Park
313                    // instead: `drop` sets `stop` and unparks, so a finished
314                    // pack's ticker goes away at once. `park_timeout` may wake
315                    // spuriously; the loop re-parks the remaining time and the
316                    // 10 s cadence below is unchanged.
317                    let now = std::time::Instant::now();
318                    if now < next {
319                        std::thread::park_timeout(next - now);
320                        continue;
321                    }
322                    next = now + TICK;
323                    progress.sample_memory();
324                    let s = progress.snapshot();
325                    if !s.active {
326                        continue;
327                    }
328                    let secs = started.elapsed().as_secs_f64().max(1e-3);
329                    tracing::info!(
330                        target: "gunnar::index",
331                        event = "index.progress",
332                        repo = %s.root,
333                        pack_id = s.pack_id,
334                        phase = s.phase.as_str(),
335                        percent = s.percent(),
336                        objects_done = s.objects_done,
337                        objects_total = s.objects_total,
338                        bytes_done = s.bytes_done,
339                        bytes_total = s.bytes_total,
340                        objects_per_s = (s.objects_done as f64 / secs) as u64,
341                        reinflations = s.reinflations,
342                        pauses = s.pauses,
343                        peak_rss = s.peak_rss,
344                        mem_available = s.mem_available,
345                        elapsed_ms = started.elapsed().as_millis() as u64,
346                        "index {} {}% {}/{} {} {}/s; {} re-inflations, {} pauses, peak rss {}, {} available",
347                        s.root,
348                        s.percent(),
349                        s.objects_done,
350                        s.objects_total,
351                        s.phase.as_str(),
352                        (s.objects_done as f64 / secs) as u64,
353                        s.reinflations,
354                        s.pauses,
355                        crate::memguard::fmt_bytes(s.peak_rss),
356                        crate::memguard::fmt_bytes(s.mem_available)
357                    );
358                }
359            })
360            .ok();
361        ProgressTicker { stop, handle }
362    }
363}
364
365impl Drop for ProgressTicker {
366    fn drop(&mut self) {
367        self.stop.store(true, Ordering::Relaxed);
368        if let Some(h) = self.handle.take() {
369            // Wake a parked ticker so the join returns at once instead of
370            // waiting out its current `park_timeout`. `unpark` before the
371            // thread parks is not lost — the token is saved and the next park
372            // returns immediately — so `stop` is always observed.
373            h.thread().unpark();
374            let _ = h.join();
375        }
376    }
377}
378
379fn registry() -> &'static Mutex<Vec<Weak<IndexProgress>>> {
380    static R: OnceLock<Mutex<Vec<Weak<IndexProgress>>>> = OnceLock::new();
381    R.get_or_init(|| Mutex::new(Vec::new()))
382}
383
384/// Every live counter, dead entries swept. Cheap: O(open stores), a few atomics
385/// each. `active_only` keeps the rows a dashboard shows; the full list is what
386/// a test or a stats call wants.
387pub fn snapshot_all(active_only: bool) -> Vec<Snapshot> {
388    let mut r = registry().lock().unwrap_or_else(|e| e.into_inner());
389    r.retain(|w| w.strong_count() > 0);
390    r.iter()
391        .filter_map(|w| w.upgrade())
392        .map(|p| p.snapshot())
393        .filter(|s| !active_only || s.active)
394        .collect()
395}
396
397#[cfg(test)]
398mod tests {
399    use super::*;
400
401    /// **Dropping a ticker must not block the caller.**
402    ///
403    /// The ticker's wait used to be an uninterruptible `sleep(250 ms)`, so a
404    /// `drop` that joined it mid-sleep waited the sleep out — a flat ~250 ms tax
405    /// on *every* pack absorb (`absorb_one` wraps its work in a ticker), which a
406    /// 24-object push measured in full: 250 ms for the ticker, under 20 ms for
407    /// everything else. Seen RED by restoring the `sleep` loop: this asserts
408    /// well under one tick's slowest step (250 ms) that the caller is freed at
409    /// once, so the guard is not sitting on the identity value of a fast box.
410    #[test]
411    fn dropping_the_ticker_does_not_block_the_caller() {
412        let p = IndexProgress::new("/tmp/x/rickard/tick-drop.git/gunnar-git.znippy");
413        p.begin(1, 1 << 20, false);
414        let ticker = ProgressTicker::start(p.clone());
415        // Let the thread reach its park.
416        std::thread::sleep(std::time::Duration::from_millis(20));
417        let t = std::time::Instant::now();
418        drop(ticker);
419        let waited = t.elapsed();
420        assert!(
421            waited < std::time::Duration::from_millis(50),
422            "drop of a ticker blocked the caller for {waited:?} — a parked ticker must be \
423             unparked and joined at once, not waited out"
424        );
425    }
426
427    #[test]
428    fn a_counter_reports_what_the_worker_did_and_percent_is_applied_output() {
429        let p = IndexProgress::new("/tmp/x/rickard/linux.git/gunnar-git.znippy");
430        assert!(!p.snapshot().active);
431        p.begin(7, 8 << 30, true);
432        p.set_total(13_948_244);
433        p.set_phase(Phase::Resolve);
434        for _ in 0..4_700_000 {
435            p.object_done(600);
436        }
437        let s = p.snapshot();
438        assert!(s.active && s.resumed);
439        assert_eq!(s.pack_id, 7);
440        assert_eq!(s.objects_done, 4_700_000);
441        assert_eq!(s.percent(), 33);
442        assert_eq!(s.bytes_done, 4_700_000 * 600);
443        assert_eq!(s.phase, Phase::Resolve);
444        assert!(s.started_at_ms > 0);
445        p.finish();
446        let s = p.snapshot();
447        assert!(!s.active);
448        assert_eq!(s.phase, Phase::Done);
449        assert_eq!(s.packs_done, 1);
450    }
451
452    #[test]
453    fn the_registry_lists_live_counters_and_forgets_dropped_ones() {
454        let root = format!("/reg-test-{}-a", std::process::id());
455        let a = IndexProgress::new(&root);
456        a.begin(1, 10, false);
457        let b = IndexProgress::new(&format!("{root}-b"));
458        let all = snapshot_all(false);
459        assert!(all.iter().any(|s| s.root == root));
460        assert!(all.iter().any(|s| s.root == format!("{root}-b")));
461        let active = snapshot_all(true);
462        assert!(active.iter().any(|s| s.root == root));
463        assert!(!active.iter().any(|s| s.root == format!("{root}-b")), "b never began");
464        drop(b);
465        assert!(!snapshot_all(false).iter().any(|s| s.root == format!("{root}-b")), "swept on drop");
466        assert_eq!(Phase::from_u8(2), Phase::Resolve);
467        assert_eq!(Phase::Paused.as_str(), "paused");
468    }
469}
znippy_plugin_git/progress.rs

znippy_plugin_git/
progress.rs
