znippy-plugin-git 0.1.5

//! **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 rickard/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 rickard/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,
    /// 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),
            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();
    }

    /// **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/rickard/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/rickard/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");
    }
}