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

znippy_plugin_git/
progress.rs
