fdu_core/

counters.rs

//! Low-distortion performance instrumentation for fdu.
//!
//! The subsystem has two complementary tiers:
//!
//! - application counters attribute logical work to the filesystem, memory, and index
//!   layers;
//! - [`process`] samples kernel counters at phase boundaries and exposes only metrics
//!   supported by the current platform.
//!
//! Recording is compiled in, off by default, and enabled with `FDU_COUNTERS=1` by the
//! shipped CLI and the repository performance probe. Per-event updates stay thread-local;
//! workers fold their counts on exit, while a drop fallback preserves counts from callers
//! that do not install an explicit worker guard.

use std::cell::Cell;
use std::ffi::OsStr;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

pub mod alloc;
pub mod process;

static ENABLED: AtomicBool = AtomicBool::new(false);

/// Per-layer tallies for a walk, index build, and content pass.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct Counts {
    /// Logical directory-open operations.
    pub dir_opens: u64,
    /// Enumeration syscalls in complete successful macOS bulk listings, including the
    /// empty terminator.
    ///
    /// One successful directory is typically two calls (one data, one EOF), the same
    /// multiplicity the Linux portable path pays in `getdents64`. The portable `read_dir`
    /// path leaves this at 0 because the standard library hides that split. Abandoned
    /// bulk attempts before portable fallback are not counted. A later Linux or `openat`
    /// campaign can compare this to `dir_opens` without another sample.
    pub dir_enumeration_calls: u64,
    /// Directory entries yielded by enumeration.
    pub dir_entries: u64,
    /// Logical metadata-stat operations.
    pub stats: u64,
    /// File-open operations performed by content analysis.
    pub file_opens: u64,
    /// File read calls performed by content analysis.
    pub file_reads: u64,
    /// Bytes read from files by content analysis.
    pub bytes_read: u64,
    /// Allocation requests observed by the installed counting allocator.
    pub allocs: u64,
    /// Reallocation requests observed by the installed counting allocator.
    pub reallocs: u64,
    /// Deallocations observed by the installed counting allocator.
    pub frees: u64,
    /// Bytes requested by allocations and allocation growth.
    pub bytes_allocated: u64,
    /// Index upserts applied.
    pub upserts: u64,
    /// Parent-path resolutions performed for index updates.
    pub parent_resolutions: u64,
    /// Parent resolutions served by the adjacent-parent memo.
    pub parent_memo_hits: u64,
    /// Roll-up merge operations.
    pub rollup_merges: u64,
    /// Control files read from the filesystem.
    pub control_reads: u64,
    /// Control sources a control table refused for one of its limits.
    ///
    /// Counted where the table decides, so a re-read of a file it has already refused is
    /// not: that batch cannot change the table and is never projected. A warm revalidate
    /// of a refused tree therefore reports its reads against no refusals, which is the
    /// projection being skipped rather than a refusal being lifted.
    pub control_refused: u64,
    /// Control sources that joined a retained identical content instead of parsing again.
    pub control_sources_shared: u64,
    /// Index entries allocated.
    pub entries_allocated: u64,
    /// Successful detached baseline batches arbitrated by the index.
    pub baseline_batches: u64,
    /// Operations accepted from successful detached baseline batches.
    pub baseline_accepted_ops: u64,
    /// Successful opened-root batches arbitrated by the index.
    pub opened_batches: u64,
    /// Operations accepted from successful opened-root batches.
    pub opened_accepted_ops: u64,
    /// Successful arbitrary public batches arbitrated by the index.
    pub public_batches: u64,
    /// Operations accepted from successful arbitrary public batches.
    pub public_accepted_ops: u64,
    /// Path-keyed structural-overlay inserts during ancestry preflight.
    pub ancestry_overlay_inserts: u64,
    /// Same-parent path comparisons during ancestry preflight.
    pub ancestry_path_comparisons: u64,
    /// Parent chains proved during ancestry preflight, including memo hits.
    pub ancestry_parent_proofs: u64,
    /// Wall microseconds spent preparing trusted scanner batches.
    pub scanner_prepare_us: u64,
    /// Wall microseconds spent projecting controls for trusted scanner batches.
    pub scanner_control_projection_us: u64,
    /// Wall microseconds spent reducing prepared scanner batches into the index.
    pub scanner_reduce_us: u64,
    /// Wall microseconds spent reading the content sidecar image.
    pub content_sidecar_read_us: u64,
    /// Wall microseconds spent on sidecar integrity and record decode.
    pub content_sidecar_parse_us: u64,
    /// Wall microseconds spent building the live candidate map for a sidecar restore.
    pub content_sidecar_candidates_us: u64,
    /// Wall microseconds spent applying restored records and rebuilding content roll-ups.
    pub content_sidecar_apply_us: u64,
    /// Detached cold scans that selected the directory-group builder.
    pub detached_builds: u64,
    /// Entries consumed by that builder.
    pub detached_entries: u64,
    /// Wall microseconds spent walking while the builder consumed directory groups.
    pub detached_walk_us: u64,
    /// Wall microseconds spent in the final directories-only roll-up pass.
    pub detached_finish_us: u64,
    /// Owned paths retained in exact effective changes.
    pub effect_paths: u64,
    /// Native encoded bytes retained by exact effective-change paths.
    pub effect_path_bytes: u64,
    /// Change and state paths considered for bounded impact derivation.
    pub impact_candidates: u64,
    /// Path ancestors visited while deriving bounded impact.
    pub impact_ancestor_visits: u64,
    /// Dirty paths retained in completed bounded impacts.
    pub impact_retained_dirty_paths: u64,
    /// Impact derivations that crossed the dirty-path bound.
    pub impact_all_dirty: u64,
    /// Exact commits retained in the bounded journal.
    pub journal_retained_commits: u64,
    /// Exact commits cloned for attempted journal retention.
    pub journal_cloned_commits: u64,
    /// Exact commits rejected because one commit exceeded journal capacity.
    pub journal_oversized_commits: u64,
    /// Older exact commits dropped to admit a new retained commit.
    pub journal_dropped_commits: u64,
    /// Chunk releases folded into an automatic scan's worker-scaling calibration.
    pub adaptive_calibration_chunks: u64,
    /// Entries folded into that calibration before it reached a decision.
    pub adaptive_calibration_entries: u64,
    /// Worker time, in microseconds, folded into that calibration.
    ///
    /// Microseconds rather than nanoseconds because the decision threshold is expressed
    /// per entry in microseconds, and a nanosecond total over a multi-million-entry walk
    /// is large enough to make the row hard to read.
    pub adaptive_calibration_work_us: u64,
    /// Reserve-pool expansions an automatic scan actually performed.
    pub adaptive_scale_ups: u64,
    /// Walks that ended while their scaling calibration was still undecided.
    ///
    /// A walk shorter than the calibration window never observes enough entries to
    /// decide, so its worker policy is unobservable rather than merely unchanged. That
    /// is the distinction [`crate::scan`] fails closed on, and it must not read as a
    /// decision to hold.
    pub adaptive_policy_undecided: u64,
}

impl Counts {
    const ZERO: Self = Self {
        dir_opens: 0,
        dir_enumeration_calls: 0,
        dir_entries: 0,
        stats: 0,
        file_opens: 0,
        file_reads: 0,
        bytes_read: 0,
        allocs: 0,
        reallocs: 0,
        frees: 0,
        bytes_allocated: 0,
        upserts: 0,
        parent_resolutions: 0,
        parent_memo_hits: 0,
        rollup_merges: 0,
        control_reads: 0,
        control_refused: 0,
        control_sources_shared: 0,
        entries_allocated: 0,
        baseline_batches: 0,
        baseline_accepted_ops: 0,
        opened_batches: 0,
        opened_accepted_ops: 0,
        public_batches: 0,
        public_accepted_ops: 0,
        ancestry_overlay_inserts: 0,
        ancestry_path_comparisons: 0,
        ancestry_parent_proofs: 0,
        scanner_prepare_us: 0,
        scanner_control_projection_us: 0,
        scanner_reduce_us: 0,
        content_sidecar_read_us: 0,
        content_sidecar_parse_us: 0,
        content_sidecar_candidates_us: 0,
        content_sidecar_apply_us: 0,
        detached_builds: 0,
        detached_entries: 0,
        detached_walk_us: 0,
        detached_finish_us: 0,
        effect_paths: 0,
        effect_path_bytes: 0,
        impact_candidates: 0,
        impact_ancestor_visits: 0,
        impact_retained_dirty_paths: 0,
        impact_all_dirty: 0,
        journal_retained_commits: 0,
        journal_cloned_commits: 0,
        journal_oversized_commits: 0,
        journal_dropped_commits: 0,
        adaptive_calibration_chunks: 0,
        adaptive_calibration_entries: 0,
        adaptive_calibration_work_us: 0,
        adaptive_scale_ups: 0,
        adaptive_policy_undecided: 0,
    };

    /// Every counter as `(group, label, value)`, in report order.
    #[must_use]
    pub fn rows(&self) -> Vec<(&'static str, &'static str, u64)> {
        vec![
            ("filesystem operations", "directory opens", self.dir_opens),
            ("filesystem operations", "directory enumeration calls", self.dir_enumeration_calls),
            ("filesystem operations", "directory entries enumerated", self.dir_entries),
            ("filesystem operations", "metadata stats", self.stats),
            ("filesystem operations", "file opens", self.file_opens),
            ("filesystem operations", "file read calls", self.file_reads),
            ("filesystem operations", "bytes read from files", self.bytes_read),
            ("memory", "allocations", self.allocs),
            ("memory", "reallocations", self.reallocs),
            ("memory", "frees", self.frees),
            ("memory", "bytes allocated", self.bytes_allocated),
            ("index", "upserts applied", self.upserts),
            ("index", "parent path resolutions", self.parent_resolutions),
            ("index", "parent memo hits", self.parent_memo_hits),
            ("index", "roll-up merges", self.rollup_merges),
            ("control state", "control files read", self.control_reads),
            ("control state", "control sources refused", self.control_refused),
            ("control state", "control sources shared", self.control_sources_shared),
            ("index", "index entries allocated", self.entries_allocated),
            ("mutation provenance", "baseline batches", self.baseline_batches),
            ("mutation provenance", "baseline accepted ops", self.baseline_accepted_ops),
            ("mutation provenance", "opened batches", self.opened_batches),
            ("mutation provenance", "opened accepted ops", self.opened_accepted_ops),
            ("mutation provenance", "public batches", self.public_batches),
            ("mutation provenance", "public accepted ops", self.public_accepted_ops),
            ("mutation preflight", "ancestry overlay inserts", self.ancestry_overlay_inserts),
            ("mutation preflight", "same-parent path comparisons", self.ancestry_path_comparisons),
            ("mutation preflight", "parent chains proved", self.ancestry_parent_proofs),
            ("mutation timing", "scanner preparation microseconds", self.scanner_prepare_us),
            (
                "mutation timing",
                "scanner control projection microseconds",
                self.scanner_control_projection_us,
            ),
            ("mutation timing", "scanner reduction microseconds", self.scanner_reduce_us),
            (
                "content sidecar timing",
                "sidecar image read microseconds",
                self.content_sidecar_read_us,
            ),
            ("content sidecar timing", "sidecar parse microseconds", self.content_sidecar_parse_us),
            (
                "content sidecar timing",
                "sidecar candidate-map microseconds",
                self.content_sidecar_candidates_us,
            ),
            ("content sidecar timing", "sidecar apply microseconds", self.content_sidecar_apply_us),
            ("detached builder", "builds", self.detached_builds),
            ("detached builder", "entries", self.detached_entries),
            ("detached builder", "walk microseconds", self.detached_walk_us),
            ("detached builder", "finish microseconds", self.detached_finish_us),
            ("mutation consequences", "effective paths retained", self.effect_paths),
            ("mutation consequences", "effective path bytes", self.effect_path_bytes),
            ("mutation consequences", "impact candidates", self.impact_candidates),
            ("mutation consequences", "impact ancestor visits", self.impact_ancestor_visits),
            (
                "mutation consequences",
                "impact dirty paths retained",
                self.impact_retained_dirty_paths,
            ),
            ("mutation consequences", "impact all-dirty transitions", self.impact_all_dirty),
            ("mutation journal", "commits retained", self.journal_retained_commits),
            ("mutation journal", "commits cloned", self.journal_cloned_commits),
            ("mutation journal", "oversized commits", self.journal_oversized_commits),
            ("mutation journal", "older commits dropped", self.journal_dropped_commits),
            ("adaptive scan policy", "calibration chunks", self.adaptive_calibration_chunks),
            ("adaptive scan policy", "calibration entries", self.adaptive_calibration_entries),
            (
                "adaptive scan policy",
                "calibration worker microseconds",
                self.adaptive_calibration_work_us,
            ),
            ("adaptive scan policy", "reserve expansions", self.adaptive_scale_ups),
            ("adaptive scan policy", "walks left undecided", self.adaptive_policy_undecided),
        ]
    }

    /// Whether every counter is zero.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        *self == Self::default()
    }

    /// Fold another set into this one without panicking on overflow.
    fn add(&mut self, other: &Self) {
        self.dir_opens = self.dir_opens.saturating_add(other.dir_opens);
        self.dir_enumeration_calls =
            self.dir_enumeration_calls.saturating_add(other.dir_enumeration_calls);
        self.dir_entries = self.dir_entries.saturating_add(other.dir_entries);
        self.stats = self.stats.saturating_add(other.stats);
        self.file_opens = self.file_opens.saturating_add(other.file_opens);
        self.file_reads = self.file_reads.saturating_add(other.file_reads);
        self.bytes_read = self.bytes_read.saturating_add(other.bytes_read);
        self.allocs = self.allocs.saturating_add(other.allocs);
        self.reallocs = self.reallocs.saturating_add(other.reallocs);
        self.frees = self.frees.saturating_add(other.frees);
        self.bytes_allocated = self.bytes_allocated.saturating_add(other.bytes_allocated);
        self.upserts = self.upserts.saturating_add(other.upserts);
        self.parent_resolutions = self.parent_resolutions.saturating_add(other.parent_resolutions);
        self.parent_memo_hits = self.parent_memo_hits.saturating_add(other.parent_memo_hits);
        self.rollup_merges = self.rollup_merges.saturating_add(other.rollup_merges);
        self.control_reads = self.control_reads.saturating_add(other.control_reads);
        self.control_refused = self.control_refused.saturating_add(other.control_refused);
        self.control_sources_shared =
            self.control_sources_shared.saturating_add(other.control_sources_shared);
        self.entries_allocated = self.entries_allocated.saturating_add(other.entries_allocated);
        self.baseline_batches = self.baseline_batches.saturating_add(other.baseline_batches);
        self.baseline_accepted_ops =
            self.baseline_accepted_ops.saturating_add(other.baseline_accepted_ops);
        self.opened_batches = self.opened_batches.saturating_add(other.opened_batches);
        self.opened_accepted_ops =
            self.opened_accepted_ops.saturating_add(other.opened_accepted_ops);
        self.public_batches = self.public_batches.saturating_add(other.public_batches);
        self.public_accepted_ops =
            self.public_accepted_ops.saturating_add(other.public_accepted_ops);
        self.ancestry_overlay_inserts =
            self.ancestry_overlay_inserts.saturating_add(other.ancestry_overlay_inserts);
        self.ancestry_path_comparisons =
            self.ancestry_path_comparisons.saturating_add(other.ancestry_path_comparisons);
        self.ancestry_parent_proofs =
            self.ancestry_parent_proofs.saturating_add(other.ancestry_parent_proofs);
        self.scanner_prepare_us = self.scanner_prepare_us.saturating_add(other.scanner_prepare_us);
        self.scanner_control_projection_us =
            self.scanner_control_projection_us.saturating_add(other.scanner_control_projection_us);
        self.scanner_reduce_us = self.scanner_reduce_us.saturating_add(other.scanner_reduce_us);
        self.content_sidecar_read_us =
            self.content_sidecar_read_us.saturating_add(other.content_sidecar_read_us);
        self.content_sidecar_parse_us =
            self.content_sidecar_parse_us.saturating_add(other.content_sidecar_parse_us);
        self.content_sidecar_candidates_us =
            self.content_sidecar_candidates_us.saturating_add(other.content_sidecar_candidates_us);
        self.content_sidecar_apply_us =
            self.content_sidecar_apply_us.saturating_add(other.content_sidecar_apply_us);
        self.detached_builds = self.detached_builds.saturating_add(other.detached_builds);
        self.detached_entries = self.detached_entries.saturating_add(other.detached_entries);
        self.detached_walk_us = self.detached_walk_us.saturating_add(other.detached_walk_us);
        self.detached_finish_us = self.detached_finish_us.saturating_add(other.detached_finish_us);
        self.effect_paths = self.effect_paths.saturating_add(other.effect_paths);
        self.effect_path_bytes = self.effect_path_bytes.saturating_add(other.effect_path_bytes);
        self.impact_candidates = self.impact_candidates.saturating_add(other.impact_candidates);
        self.impact_ancestor_visits =
            self.impact_ancestor_visits.saturating_add(other.impact_ancestor_visits);
        self.impact_retained_dirty_paths =
            self.impact_retained_dirty_paths.saturating_add(other.impact_retained_dirty_paths);
        self.impact_all_dirty = self.impact_all_dirty.saturating_add(other.impact_all_dirty);
        self.journal_retained_commits =
            self.journal_retained_commits.saturating_add(other.journal_retained_commits);
        self.journal_cloned_commits =
            self.journal_cloned_commits.saturating_add(other.journal_cloned_commits);
        self.journal_oversized_commits =
            self.journal_oversized_commits.saturating_add(other.journal_oversized_commits);
        self.journal_dropped_commits =
            self.journal_dropped_commits.saturating_add(other.journal_dropped_commits);
        self.adaptive_calibration_chunks =
            self.adaptive_calibration_chunks.saturating_add(other.adaptive_calibration_chunks);
        self.adaptive_calibration_entries =
            self.adaptive_calibration_entries.saturating_add(other.adaptive_calibration_entries);
        self.adaptive_calibration_work_us =
            self.adaptive_calibration_work_us.saturating_add(other.adaptive_calibration_work_us);
        self.adaptive_scale_ups = self.adaptive_scale_ups.saturating_add(other.adaptive_scale_ups);
        self.adaptive_policy_undecided =
            self.adaptive_policy_undecided.saturating_add(other.adaptive_policy_undecided);
    }

    /// Per-event ratios against a denominator, in report order.
    #[must_use]
    pub fn per(&self, denominator: u64) -> Vec<(&'static str, &'static str, f64)> {
        self.rows()
            .into_iter()
            .map(|(group, label, value)| (group, label, ratio(value, denominator)))
            .collect()
    }
}

struct GlobalCounts {
    dir_opens: AtomicU64,
    dir_enumeration_calls: AtomicU64,
    dir_entries: AtomicU64,
    stats: AtomicU64,
    file_opens: AtomicU64,
    file_reads: AtomicU64,
    bytes_read: AtomicU64,
    allocs: AtomicU64,
    reallocs: AtomicU64,
    frees: AtomicU64,
    bytes_allocated: AtomicU64,
    upserts: AtomicU64,
    parent_resolutions: AtomicU64,
    parent_memo_hits: AtomicU64,
    rollup_merges: AtomicU64,
    control_reads: AtomicU64,
    control_refused: AtomicU64,
    control_sources_shared: AtomicU64,
    entries_allocated: AtomicU64,
    baseline_batches: AtomicU64,
    baseline_accepted_ops: AtomicU64,
    opened_batches: AtomicU64,
    opened_accepted_ops: AtomicU64,
    public_batches: AtomicU64,
    public_accepted_ops: AtomicU64,
    ancestry_overlay_inserts: AtomicU64,
    ancestry_path_comparisons: AtomicU64,
    ancestry_parent_proofs: AtomicU64,
    scanner_prepare_us: AtomicU64,
    scanner_control_projection_us: AtomicU64,
    scanner_reduce_us: AtomicU64,
    content_sidecar_read_us: AtomicU64,
    content_sidecar_parse_us: AtomicU64,
    content_sidecar_candidates_us: AtomicU64,
    content_sidecar_apply_us: AtomicU64,
    detached_builds: AtomicU64,
    detached_entries: AtomicU64,
    detached_walk_us: AtomicU64,
    detached_finish_us: AtomicU64,
    effect_paths: AtomicU64,
    effect_path_bytes: AtomicU64,
    impact_candidates: AtomicU64,
    impact_ancestor_visits: AtomicU64,
    impact_retained_dirty_paths: AtomicU64,
    impact_all_dirty: AtomicU64,
    journal_retained_commits: AtomicU64,
    journal_cloned_commits: AtomicU64,
    journal_oversized_commits: AtomicU64,
    journal_dropped_commits: AtomicU64,
    adaptive_calibration_chunks: AtomicU64,
    adaptive_calibration_entries: AtomicU64,
    adaptive_calibration_work_us: AtomicU64,
    adaptive_scale_ups: AtomicU64,
    adaptive_policy_undecided: AtomicU64,
}

impl GlobalCounts {
    const fn new() -> Self {
        Self {
            dir_opens: AtomicU64::new(0),
            dir_enumeration_calls: AtomicU64::new(0),
            dir_entries: AtomicU64::new(0),
            stats: AtomicU64::new(0),
            file_opens: AtomicU64::new(0),
            file_reads: AtomicU64::new(0),
            bytes_read: AtomicU64::new(0),
            allocs: AtomicU64::new(0),
            reallocs: AtomicU64::new(0),
            frees: AtomicU64::new(0),
            bytes_allocated: AtomicU64::new(0),
            upserts: AtomicU64::new(0),
            parent_resolutions: AtomicU64::new(0),
            parent_memo_hits: AtomicU64::new(0),
            rollup_merges: AtomicU64::new(0),
            control_reads: AtomicU64::new(0),
            control_refused: AtomicU64::new(0),
            control_sources_shared: AtomicU64::new(0),
            entries_allocated: AtomicU64::new(0),
            baseline_batches: AtomicU64::new(0),
            baseline_accepted_ops: AtomicU64::new(0),
            opened_batches: AtomicU64::new(0),
            opened_accepted_ops: AtomicU64::new(0),
            public_batches: AtomicU64::new(0),
            public_accepted_ops: AtomicU64::new(0),
            ancestry_overlay_inserts: AtomicU64::new(0),
            ancestry_path_comparisons: AtomicU64::new(0),
            ancestry_parent_proofs: AtomicU64::new(0),
            scanner_prepare_us: AtomicU64::new(0),
            scanner_control_projection_us: AtomicU64::new(0),
            scanner_reduce_us: AtomicU64::new(0),
            content_sidecar_read_us: AtomicU64::new(0),
            content_sidecar_parse_us: AtomicU64::new(0),
            content_sidecar_candidates_us: AtomicU64::new(0),
            content_sidecar_apply_us: AtomicU64::new(0),
            detached_builds: AtomicU64::new(0),
            detached_entries: AtomicU64::new(0),
            detached_walk_us: AtomicU64::new(0),
            detached_finish_us: AtomicU64::new(0),
            effect_paths: AtomicU64::new(0),
            effect_path_bytes: AtomicU64::new(0),
            impact_candidates: AtomicU64::new(0),
            impact_ancestor_visits: AtomicU64::new(0),
            impact_retained_dirty_paths: AtomicU64::new(0),
            impact_all_dirty: AtomicU64::new(0),
            journal_retained_commits: AtomicU64::new(0),
            journal_cloned_commits: AtomicU64::new(0),
            journal_oversized_commits: AtomicU64::new(0),
            journal_dropped_commits: AtomicU64::new(0),
            adaptive_calibration_chunks: AtomicU64::new(0),
            adaptive_calibration_entries: AtomicU64::new(0),
            adaptive_calibration_work_us: AtomicU64::new(0),
            adaptive_scale_ups: AtomicU64::new(0),
            adaptive_policy_undecided: AtomicU64::new(0),
        }
    }

    fn add(&self, counts: &Counts) {
        atomic_saturating_add(&self.dir_opens, counts.dir_opens);
        atomic_saturating_add(&self.dir_enumeration_calls, counts.dir_enumeration_calls);
        atomic_saturating_add(&self.dir_entries, counts.dir_entries);
        atomic_saturating_add(&self.stats, counts.stats);
        atomic_saturating_add(&self.file_opens, counts.file_opens);
        atomic_saturating_add(&self.file_reads, counts.file_reads);
        atomic_saturating_add(&self.bytes_read, counts.bytes_read);
        atomic_saturating_add(&self.allocs, counts.allocs);
        atomic_saturating_add(&self.reallocs, counts.reallocs);
        atomic_saturating_add(&self.frees, counts.frees);
        atomic_saturating_add(&self.bytes_allocated, counts.bytes_allocated);
        atomic_saturating_add(&self.upserts, counts.upserts);
        atomic_saturating_add(&self.parent_resolutions, counts.parent_resolutions);
        atomic_saturating_add(&self.parent_memo_hits, counts.parent_memo_hits);
        atomic_saturating_add(&self.rollup_merges, counts.rollup_merges);
        atomic_saturating_add(&self.control_reads, counts.control_reads);
        atomic_saturating_add(&self.control_refused, counts.control_refused);
        atomic_saturating_add(&self.control_sources_shared, counts.control_sources_shared);
        atomic_saturating_add(&self.entries_allocated, counts.entries_allocated);
        atomic_saturating_add(&self.baseline_batches, counts.baseline_batches);
        atomic_saturating_add(&self.baseline_accepted_ops, counts.baseline_accepted_ops);
        atomic_saturating_add(&self.opened_batches, counts.opened_batches);
        atomic_saturating_add(&self.opened_accepted_ops, counts.opened_accepted_ops);
        atomic_saturating_add(&self.public_batches, counts.public_batches);
        atomic_saturating_add(&self.public_accepted_ops, counts.public_accepted_ops);
        atomic_saturating_add(&self.ancestry_overlay_inserts, counts.ancestry_overlay_inserts);
        atomic_saturating_add(&self.ancestry_path_comparisons, counts.ancestry_path_comparisons);
        atomic_saturating_add(&self.ancestry_parent_proofs, counts.ancestry_parent_proofs);
        atomic_saturating_add(&self.scanner_prepare_us, counts.scanner_prepare_us);
        atomic_saturating_add(
            &self.scanner_control_projection_us,
            counts.scanner_control_projection_us,
        );
        atomic_saturating_add(&self.scanner_reduce_us, counts.scanner_reduce_us);
        atomic_saturating_add(&self.content_sidecar_read_us, counts.content_sidecar_read_us);
        atomic_saturating_add(&self.content_sidecar_parse_us, counts.content_sidecar_parse_us);
        atomic_saturating_add(
            &self.content_sidecar_candidates_us,
            counts.content_sidecar_candidates_us,
        );
        atomic_saturating_add(&self.content_sidecar_apply_us, counts.content_sidecar_apply_us);
        atomic_saturating_add(&self.detached_builds, counts.detached_builds);
        atomic_saturating_add(&self.detached_entries, counts.detached_entries);
        atomic_saturating_add(&self.detached_walk_us, counts.detached_walk_us);
        atomic_saturating_add(&self.detached_finish_us, counts.detached_finish_us);
        atomic_saturating_add(&self.effect_paths, counts.effect_paths);
        atomic_saturating_add(&self.effect_path_bytes, counts.effect_path_bytes);
        atomic_saturating_add(&self.impact_candidates, counts.impact_candidates);
        atomic_saturating_add(&self.impact_ancestor_visits, counts.impact_ancestor_visits);
        atomic_saturating_add(
            &self.impact_retained_dirty_paths,
            counts.impact_retained_dirty_paths,
        );
        atomic_saturating_add(&self.impact_all_dirty, counts.impact_all_dirty);
        atomic_saturating_add(&self.journal_retained_commits, counts.journal_retained_commits);
        atomic_saturating_add(&self.journal_cloned_commits, counts.journal_cloned_commits);
        atomic_saturating_add(&self.journal_oversized_commits, counts.journal_oversized_commits);
        atomic_saturating_add(&self.journal_dropped_commits, counts.journal_dropped_commits);
        atomic_saturating_add(
            &self.adaptive_calibration_chunks,
            counts.adaptive_calibration_chunks,
        );
        atomic_saturating_add(
            &self.adaptive_calibration_entries,
            counts.adaptive_calibration_entries,
        );
        atomic_saturating_add(
            &self.adaptive_calibration_work_us,
            counts.adaptive_calibration_work_us,
        );
        atomic_saturating_add(&self.adaptive_scale_ups, counts.adaptive_scale_ups);
        atomic_saturating_add(&self.adaptive_policy_undecided, counts.adaptive_policy_undecided);
    }

    fn snapshot(&self) -> Counts {
        Counts {
            dir_opens: self.dir_opens.load(Ordering::Relaxed),
            dir_enumeration_calls: self.dir_enumeration_calls.load(Ordering::Relaxed),
            dir_entries: self.dir_entries.load(Ordering::Relaxed),
            stats: self.stats.load(Ordering::Relaxed),
            file_opens: self.file_opens.load(Ordering::Relaxed),
            file_reads: self.file_reads.load(Ordering::Relaxed),
            bytes_read: self.bytes_read.load(Ordering::Relaxed),
            allocs: self.allocs.load(Ordering::Relaxed),
            reallocs: self.reallocs.load(Ordering::Relaxed),
            frees: self.frees.load(Ordering::Relaxed),
            bytes_allocated: self.bytes_allocated.load(Ordering::Relaxed),
            upserts: self.upserts.load(Ordering::Relaxed),
            parent_resolutions: self.parent_resolutions.load(Ordering::Relaxed),
            parent_memo_hits: self.parent_memo_hits.load(Ordering::Relaxed),
            rollup_merges: self.rollup_merges.load(Ordering::Relaxed),
            control_reads: self.control_reads.load(Ordering::Relaxed),
            control_refused: self.control_refused.load(Ordering::Relaxed),
            control_sources_shared: self.control_sources_shared.load(Ordering::Relaxed),
            entries_allocated: self.entries_allocated.load(Ordering::Relaxed),
            baseline_batches: self.baseline_batches.load(Ordering::Relaxed),
            baseline_accepted_ops: self.baseline_accepted_ops.load(Ordering::Relaxed),
            opened_batches: self.opened_batches.load(Ordering::Relaxed),
            opened_accepted_ops: self.opened_accepted_ops.load(Ordering::Relaxed),
            public_batches: self.public_batches.load(Ordering::Relaxed),
            public_accepted_ops: self.public_accepted_ops.load(Ordering::Relaxed),
            ancestry_overlay_inserts: self.ancestry_overlay_inserts.load(Ordering::Relaxed),
            ancestry_path_comparisons: self.ancestry_path_comparisons.load(Ordering::Relaxed),
            ancestry_parent_proofs: self.ancestry_parent_proofs.load(Ordering::Relaxed),
            scanner_prepare_us: self.scanner_prepare_us.load(Ordering::Relaxed),
            scanner_control_projection_us: self
                .scanner_control_projection_us
                .load(Ordering::Relaxed),
            scanner_reduce_us: self.scanner_reduce_us.load(Ordering::Relaxed),
            content_sidecar_read_us: self.content_sidecar_read_us.load(Ordering::Relaxed),
            content_sidecar_parse_us: self.content_sidecar_parse_us.load(Ordering::Relaxed),
            content_sidecar_candidates_us: self
                .content_sidecar_candidates_us
                .load(Ordering::Relaxed),
            content_sidecar_apply_us: self.content_sidecar_apply_us.load(Ordering::Relaxed),
            detached_builds: self.detached_builds.load(Ordering::Relaxed),
            detached_entries: self.detached_entries.load(Ordering::Relaxed),
            detached_walk_us: self.detached_walk_us.load(Ordering::Relaxed),
            detached_finish_us: self.detached_finish_us.load(Ordering::Relaxed),
            effect_paths: self.effect_paths.load(Ordering::Relaxed),
            effect_path_bytes: self.effect_path_bytes.load(Ordering::Relaxed),
            impact_candidates: self.impact_candidates.load(Ordering::Relaxed),
            impact_ancestor_visits: self.impact_ancestor_visits.load(Ordering::Relaxed),
            impact_retained_dirty_paths: self.impact_retained_dirty_paths.load(Ordering::Relaxed),
            impact_all_dirty: self.impact_all_dirty.load(Ordering::Relaxed),
            journal_retained_commits: self.journal_retained_commits.load(Ordering::Relaxed),
            journal_cloned_commits: self.journal_cloned_commits.load(Ordering::Relaxed),
            journal_oversized_commits: self.journal_oversized_commits.load(Ordering::Relaxed),
            journal_dropped_commits: self.journal_dropped_commits.load(Ordering::Relaxed),
            adaptive_calibration_chunks: self.adaptive_calibration_chunks.load(Ordering::Relaxed),
            adaptive_calibration_entries: self.adaptive_calibration_entries.load(Ordering::Relaxed),
            adaptive_calibration_work_us: self.adaptive_calibration_work_us.load(Ordering::Relaxed),
            adaptive_scale_ups: self.adaptive_scale_ups.load(Ordering::Relaxed),
            adaptive_policy_undecided: self.adaptive_policy_undecided.load(Ordering::Relaxed),
        }
    }

    fn reset(&self) {
        self.dir_opens.store(0, Ordering::Relaxed);
        self.dir_enumeration_calls.store(0, Ordering::Relaxed);
        self.dir_entries.store(0, Ordering::Relaxed);
        self.stats.store(0, Ordering::Relaxed);
        self.file_opens.store(0, Ordering::Relaxed);
        self.file_reads.store(0, Ordering::Relaxed);
        self.bytes_read.store(0, Ordering::Relaxed);
        self.allocs.store(0, Ordering::Relaxed);
        self.reallocs.store(0, Ordering::Relaxed);
        self.frees.store(0, Ordering::Relaxed);
        self.bytes_allocated.store(0, Ordering::Relaxed);
        self.upserts.store(0, Ordering::Relaxed);
        self.parent_resolutions.store(0, Ordering::Relaxed);
        self.parent_memo_hits.store(0, Ordering::Relaxed);
        self.rollup_merges.store(0, Ordering::Relaxed);
        self.control_reads.store(0, Ordering::Relaxed);
        self.control_refused.store(0, Ordering::Relaxed);
        self.control_sources_shared.store(0, Ordering::Relaxed);
        self.entries_allocated.store(0, Ordering::Relaxed);
        self.baseline_batches.store(0, Ordering::Relaxed);
        self.baseline_accepted_ops.store(0, Ordering::Relaxed);
        self.opened_batches.store(0, Ordering::Relaxed);
        self.opened_accepted_ops.store(0, Ordering::Relaxed);
        self.public_batches.store(0, Ordering::Relaxed);
        self.public_accepted_ops.store(0, Ordering::Relaxed);
        self.ancestry_overlay_inserts.store(0, Ordering::Relaxed);
        self.ancestry_path_comparisons.store(0, Ordering::Relaxed);
        self.ancestry_parent_proofs.store(0, Ordering::Relaxed);
        self.scanner_prepare_us.store(0, Ordering::Relaxed);
        self.scanner_control_projection_us.store(0, Ordering::Relaxed);
        self.scanner_reduce_us.store(0, Ordering::Relaxed);
        self.content_sidecar_read_us.store(0, Ordering::Relaxed);
        self.content_sidecar_parse_us.store(0, Ordering::Relaxed);
        self.content_sidecar_candidates_us.store(0, Ordering::Relaxed);
        self.content_sidecar_apply_us.store(0, Ordering::Relaxed);
        self.detached_builds.store(0, Ordering::Relaxed);
        self.detached_entries.store(0, Ordering::Relaxed);
        self.detached_walk_us.store(0, Ordering::Relaxed);
        self.detached_finish_us.store(0, Ordering::Relaxed);
        self.effect_paths.store(0, Ordering::Relaxed);
        self.effect_path_bytes.store(0, Ordering::Relaxed);
        self.impact_candidates.store(0, Ordering::Relaxed);
        self.impact_ancestor_visits.store(0, Ordering::Relaxed);
        self.impact_retained_dirty_paths.store(0, Ordering::Relaxed);
        self.impact_all_dirty.store(0, Ordering::Relaxed);
        self.journal_retained_commits.store(0, Ordering::Relaxed);
        self.journal_cloned_commits.store(0, Ordering::Relaxed);
        self.journal_oversized_commits.store(0, Ordering::Relaxed);
        self.journal_dropped_commits.store(0, Ordering::Relaxed);
        self.adaptive_calibration_chunks.store(0, Ordering::Relaxed);
        self.adaptive_calibration_entries.store(0, Ordering::Relaxed);
        self.adaptive_calibration_work_us.store(0, Ordering::Relaxed);
        self.adaptive_scale_ups.store(0, Ordering::Relaxed);
        self.adaptive_policy_undecided.store(0, Ordering::Relaxed);
    }
}

static GLOBAL: GlobalCounts = GlobalCounts::new();

struct LocalCounts {
    counts: Cell<Counts>,
}

impl LocalCounts {
    const fn new() -> Self {
        Self { counts: Cell::new(Counts::ZERO) }
    }
}

impl Drop for LocalCounts {
    fn drop(&mut self) {
        GLOBAL.add(&self.counts.replace(Counts::default()));
    }
}

std::thread_local! {
    // This non-dropping, const-initialized guard is accessed before `LOCAL`. If first
    // access to a dropping TLS key causes an allocation on a platform, the allocator's
    // nested callback sees `true` and takes the atomic fallback instead of recursing.
    static IN_COUNTER: Cell<bool> = const { Cell::new(false) };
    static LOCAL: LocalCounts = const { LocalCounts::new() };
}

struct ReentryGuard<'a>(&'a Cell<bool>);

impl<'a> ReentryGuard<'a> {
    fn enter(cell: &'a Cell<bool>) -> Option<Self> {
        if cell.replace(true) { None } else { Some(Self(cell)) }
    }
}

impl Drop for ReentryGuard<'_> {
    fn drop(&mut self) {
        self.0.set(false);
    }
}

/// A guard that deterministically folds one worker's counters before it returns.
pub(crate) struct ThreadFlushGuard;

impl Drop for ThreadFlushGuard {
    fn drop(&mut self) {
        flush_thread();
    }
}

/// Begin a worker lifetime whose thread-local counters must be visible after join.
pub(crate) const fn thread_flush_guard() -> ThreadFlushGuard {
    ThreadFlushGuard
}

/// One opt-in instrumentation interval for a binary invocation.
///
/// Construct this before the measured operation and call [`Self::finish`] afterward.
/// When recording is off, construction performs no process sampling and `finish`
/// returns `None`.
#[must_use]
pub struct Measurement {
    process_before: Option<process::Snapshot>,
}

impl Measurement {
    /// Read `FDU_COUNTERS` and begin an enabled interval when it is truthy.
    pub fn from_env() -> Self {
        if !enable_from_env() {
            return Self { process_before: None };
        }

        // The boundary sample can allocate on Linux while parsing `/proc`. Take it first
        // and reset afterward so the application tier describes fdu's work, not its
        // measuring instrument.
        let process_before = process::Snapshot::now();
        reset();
        Self { process_before: Some(process_before) }
    }

    /// Finish the interval and render application and supported process counters.
    #[must_use]
    pub fn finish(mut self) -> Option<String> {
        let before = self.process_before.take()?;
        flush_thread();
        let counts = snapshot();
        let process = process::Snapshot::now().since(&before);
        enable(false);
        Some(render(&counts, &process))
    }
}

impl Drop for Measurement {
    fn drop(&mut self) {
        if self.process_before.is_some() {
            enable(false);
        }
    }
}

/// Turn recording on or off for the whole process.
pub fn enable(on: bool) {
    ENABLED.store(on, Ordering::Relaxed);
}

/// Turn recording on when `FDU_COUNTERS` contains a truthy value.
///
/// Empty strings, `0`, `false`, `no`, and `off` (case-insensitive) disable recording.
/// The parsed state is always applied, including a transition from enabled to disabled.
pub fn enable_from_env() -> bool {
    enable_from_value(std::env::var_os("FDU_COUNTERS").as_deref())
}

fn enable_from_value(value: Option<&OsStr>) -> bool {
    let on = value.is_some_and(|value| {
        value.to_str().is_none_or(|text| {
            let text = text.trim();
            !(text.is_empty()
                || text == "0"
                || text.eq_ignore_ascii_case("false")
                || text.eq_ignore_ascii_case("no")
                || text.eq_ignore_ascii_case("off"))
        })
    });
    enable(on);
    on
}

/// Whether recording is currently on.
#[inline]
#[must_use]
pub fn enabled() -> bool {
    ENABLED.load(Ordering::Relaxed)
}

/// Elapsed microseconds since `started`, saturating at `u64::MAX`.
#[must_use]
pub(crate) fn elapsed_micros(started: std::time::Instant) -> u64 {
    u64::try_from(started.elapsed().as_micros()).unwrap_or(u64::MAX)
}

/// Record a whole-phase duration when recording is on.
pub(crate) fn add_elapsed(started: Option<std::time::Instant>, add: impl FnOnce(&mut Counts, u64)) {
    if let Some(started) = started {
        bump(|counts| add(counts, elapsed_micros(started)));
    }
}

/// Add to one or more counters on the calling thread.
///
/// The fallback path is important for allocation callbacks: it avoids re-entering a TLS
/// initializer and remains usable while a thread's local counter has begun destruction.
#[inline]
pub fn bump(f: impl FnOnce(&mut Counts)) {
    if !enabled() {
        return;
    }

    let mut f = Some(f);
    let stored_locally = IN_COUNTER
        .try_with(|reentry| {
            let Some(_guard) = ReentryGuard::enter(reentry) else { return false };
            LOCAL
                .try_with(|local| {
                    if let Some(update) = f.take() {
                        let mut counts = local.counts.get();
                        update(&mut counts);
                        local.counts.set(counts);
                    }
                })
                .is_ok()
        })
        .unwrap_or(false);

    if !stored_locally {
        let mut delta = Counts::default();
        if let Some(update) = f {
            update(&mut delta);
            GLOBAL.add(&delta);
        }
    }
}

/// Fold this thread's counts into the process total and clear them.
pub fn flush_thread() {
    let _ = LOCAL.try_with(|local| GLOBAL.add(&local.counts.replace(Counts::default())));
}

/// Process-wide totals, including the calling thread's unflushed counts.
#[must_use]
pub fn snapshot() -> Counts {
    let mut total = GLOBAL.snapshot();
    let _ = LOCAL.try_with(|local| total.add(&local.counts.get()));
    total
}

/// The calling thread's unflushed counts, without consuming them.
///
/// Use paired reads to attribute synchronous diagnostic work without also counting
/// workers that finish during it. A [`flush_thread`] between reads starts a new local
/// interval; these are not lifetime totals. No other thread's counts are included.
#[must_use]
pub fn thread_snapshot() -> Counts {
    LOCAL.try_with(|local| local.counts.get()).unwrap_or_default()
}

/// Clear process totals and the calling thread's local counts.
///
/// Call only when other instrumented workers are quiescent: another thread may still
/// hold local counts that a process-global reset cannot reach.
pub fn reset() {
    let _ = LOCAL.try_with(|local| local.counts.set(Counts::default()));
    GLOBAL.reset();
}

/// Create fdu's certified system-allocator wrapper for a binary's global allocator.
#[must_use]
pub const fn system_allocator() -> alloc::CountingAlloc<std::alloc::System> {
    alloc::CountingAlloc::system(alloc::fdu_sinks())
}

/// Render application counters followed by the supported kernel process counters.
#[must_use]
pub fn render(counts: &Counts, process: &process::Snapshot) -> String {
    let mut out = render_rows(&counts.rows());
    out.push('\n');
    out.push_str(&process.render());
    out
}

fn record_alloc(size: u64) {
    bump(|counts| {
        counts.allocs = counts.allocs.saturating_add(1);
        counts.bytes_allocated = counts.bytes_allocated.saturating_add(size);
    });
}

fn record_realloc(growth: u64) {
    bump(|counts| {
        counts.reallocs = counts.reallocs.saturating_add(1);
        counts.bytes_allocated = counts.bytes_allocated.saturating_add(growth);
    });
}

fn record_dealloc() {
    bump(|counts| counts.frees = counts.frees.saturating_add(1));
}

fn atomic_saturating_add(target: &AtomicU64, value: u64) {
    if value == 0 {
        return;
    }
    let _ = target.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
        Some(current.saturating_add(value))
    });
}

#[allow(clippy::cast_precision_loss)]
fn ratio(numerator: u64, denominator: u64) -> f64 {
    let denominator = if denominator == 0 { 1.0 } else { denominator as f64 };
    numerator as f64 / denominator
}

fn render_rows(rows: &[(&str, &str, u64)]) -> String {
    use std::fmt::Write as _;

    let width = rows.iter().map(|(_, label, _)| label.len()).max().unwrap_or(0);
    let mut output = String::new();
    let mut previous = None;
    for (group, label, value) in rows {
        if previous != Some(*group) {
            if previous.is_some() {
                output.push('\n');
            }
            let _ = writeln!(output, "[{group}]");
            previous = Some(*group);
        }
        let _ = writeln!(output, "  {label:<width$}  {value:>13}");
    }
    output
}

/// Serialize tests that touch process-global counter state.
#[cfg(test)]
pub(crate) fn test_serial() -> std::sync::MutexGuard<'static, ()> {
    static SERIAL: std::sync::Mutex<()> = std::sync::Mutex::new(());
    SERIAL.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
}

/// Calling-thread counts for exact unit tests that must ignore unrelated workers.
#[cfg(test)]
pub(crate) fn test_thread_snapshot() -> Counts {
    thread_snapshot()
}

/// Clear only the calling test thread's counts.
#[cfg(test)]
pub(crate) fn test_thread_reset() {
    let _ = LOCAL.try_with(|local| local.counts.set(Counts::default()));
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn an_exiting_thread_folds_without_a_manual_flush() {
        let _serial = test_serial();
        enable(true);
        reset();

        std::thread::spawn(|| bump(|counts| counts.dir_opens = 7)).join().expect("counting worker");

        // Other ordinary tests can perform instrumented work while this process-wide
        // toggle is on, so only the worker's lower bound is deterministic here.
        assert!(snapshot().dir_opens >= 7);
        reset();
        enable(false);
    }

    #[test]
    fn thread_snapshot_excludes_workers_without_clearing_local_counts() {
        let _serial = test_serial();
        enable(true);
        reset();
        bump(|counts| counts.dir_opens = 3);

        std::thread::spawn(|| bump(|counts| counts.dir_opens = 7)).join().expect("worker");

        assert_eq!(thread_snapshot().dir_opens, 3);
        assert_eq!(thread_snapshot().dir_opens, 3, "reading must not consume counts");
        assert!(snapshot().dir_opens >= 10);
        flush_thread();
        assert_eq!(thread_snapshot().dir_opens, 0);
        reset();
        enable(false);
    }

    #[test]
    fn a_falsey_value_disables_a_previously_enabled_process() {
        let _serial = test_serial();
        enable(true);

        assert!(!enable_from_value(Some(OsStr::new("FALSE"))));
        assert!(!enabled());
    }

    #[test]
    fn ratios_treat_a_zero_denominator_as_one() {
        let counts = Counts { allocs: 10, ..Counts::default() };
        let allocations = counts
            .per(0)
            .into_iter()
            .find(|(_, label, _)| *label == "allocations")
            .expect("allocation row");
        assert!((allocations.2 - 10.0).abs() < f64::EPSILON);
    }
}