polyc-query 2026.9.0

//! Prometheus metrics for this crate's own query-sizing observability (QRY-3
//! hardening review, items D and B).
//!
//! Registered into the process default registry — same pattern as this
//! workspace's other component-level `metrics.rs` modules (`polyc-eventlog`'s
//! is the reference): no separate scrape endpoint, no separate registry
//! plumbing. The `polyc-runtime` side-server's `/metrics` handler gathers the
//! default registry, so these series show up there for free in any binary
//! that links this crate and calls [`crate::init_metrics`] at startup.
//!
//! # Why these series exist
//!
//! `crate::engine::QueryLimits::max_source_events` is an event-COUNT
//! backstop, not a byte-based memory bound (see that field's own doc) — it
//! cannot answer "how large is a real query's replayed data, in bytes" on
//! its own. [`record_query_observed`] is called on every query
//! `crate::authority::ScopedQuery::execute` runs to completion — success or
//! an event-count budget rejection alike — right after replay; a query that
//! trips the BYTE budget aborts mid-replay and records
//! [`record_query_replayed_bytes`] instead (its event count stopped early and
//! is unknowable). Either way the
//! `polychrome_query_source_events`/`polychrome_query_replayed_bytes`
//! histograms give an operator the real distribution needed to tune
//! `max_source_events`/`max_source_bytes` for their own deployment (see
//! `crate::engine::QueryLimits::default`'s doc), rather than trusting a
//! fixture-derived guess. [`record_source_budget_exceeded`] is the
//! alertable counterpart to a budget rejection, so a runaway/misconfigured
//! caller is observable without grepping logs.

use std::sync::OnceLock;

use prometheus::{
    HistogramVec, IntCounter, IntCounterVec, IntGauge, histogram_opts, register_histogram_vec,
    register_int_counter, register_int_counter_vec, register_int_gauge,
};

/// Every `scope` label value this module's series are labeled with —
/// `"fleet"`/`"grant"`/`"persona"` are each produced by exactly one
/// `crate::authority::Principal` kind (`ScopedQuery::scope_label`'s doc names
/// which); `"conversations"` is the generic fallback for
/// `crate::session::QueryScope::Conversations`, kept so the label match stays
/// exhaustive over that enum's own shape even though no current `Principal`
/// constructor reaches it. Kept in one place so [`force`] can pre-create
/// every child without drifting from `scope_label`'s own match arms. Scope
/// TYPE only, deliberately — never a per-partition or per-conversation label,
/// which would blow up cardinality on a Fleet-wide deployment.
const SCOPE_LABELS: [&str; 4] = ["fleet", "grant", "persona", "conversations"];

/// Total events replayed for one query, across every scoped partition — see
/// this module's doc for why this is observed on every query, not just a
/// rejected one.
fn source_events() -> &'static HistogramVec {
    static V: OnceLock<HistogramVec> = OnceLock::new();
    V.get_or_init(|| {
        register_histogram_vec!(
            histogram_opts!(
                "polychrome_query_source_events",
                "Total events replayed for one query, across every scoped partition, by scope.",
                vec![
                    100.0,
                    1_000.0,
                    10_000.0,
                    50_000.0,
                    100_000.0,
                    250_000.0,
                    500_000.0,
                    1_000_000.0,
                    2_000_000.0,
                    5_000_000.0,
                ]
            ),
            &["scope"]
        )
        .expect("register polychrome_query_source_events")
    })
}

/// Total replayed event PAYLOAD bytes for one query — the sum of every
/// replayed event's own `payload.len()`, across every scoped partition. The
/// byte-based signal `max_source_events` itself cannot give (an event-COUNT
/// budget says nothing about how large any one event's payload is) — see
/// this module's doc and issue #1541, which tracks the real byte-based
/// meter this histogram is the observability half of, not the enforcement
/// half.
fn replayed_bytes() -> &'static HistogramVec {
    static V: OnceLock<HistogramVec> = OnceLock::new();
    V.get_or_init(|| {
        register_histogram_vec!(
            histogram_opts!(
                "polychrome_query_replayed_bytes",
                "Total replayed event payload bytes for one query, across every scoped \
                 partition, by scope.",
                vec![
                    1_024.0,
                    16_384.0,
                    131_072.0,
                    1_048_576.0,
                    16_777_216.0,
                    134_217_728.0,
                    536_870_912.0,
                    1_073_741_824.0,
                ]
            ),
            &["scope"]
        )
        .expect("register polychrome_query_replayed_bytes")
    })
}

/// Total EFFECTIVE event volume `crate::engine::QueryEngine::build_from_tables`
/// actually scans for one query — cached rows plus any freshly-replayed
/// tail, summed across every scoped partition
/// (`crate::authority::ScopedQuery::resolve_partitions_cached`'s own
/// `cached_volume_events`). Unlike [`source_events`], this stays accurate on
/// a cache HIT, where `source_events` reports near-zero replay even though
/// `DataFusion` still scans the full cached table — see this module's doc
/// for why folding this into `source_events`/`replayed_bytes` instead of
/// giving it its own labeled dimension would hide exactly the queries an
/// operator most needs to see (the cheap, cache-served ones).
fn cached_scan_events() -> &'static HistogramVec {
    static V: OnceLock<HistogramVec> = OnceLock::new();
    V.get_or_init(|| {
        register_histogram_vec!(
            histogram_opts!(
                "polychrome_query_cached_scan_events",
                "Total effective event volume one query hands to DataFusion — cached rows plus \
                 any freshly-replayed tail, summed across every scoped partition, by scope.",
                vec![
                    100.0,
                    1_000.0,
                    10_000.0,
                    50_000.0,
                    100_000.0,
                    250_000.0,
                    500_000.0,
                    1_000_000.0,
                    2_000_000.0,
                    5_000_000.0,
                ]
            ),
            &["scope"]
        )
        .expect("register polychrome_query_cached_scan_events")
    })
}

/// The byte-sized counterpart to [`cached_scan_events`] — total Arrow array
/// memory (`crate::engine::PartitionTables::memory_bytes`, summed across
/// every scoped partition's resolved tables, hit/tail/miss alike) one query
/// hands to `DataFusion`. Deliberately a different unit than
/// [`replayed_bytes`] (journal payload bytes): this is the decoded,
/// in-memory Arrow footprint `DataFusion` actually plans and scans, which a
/// cache hit's near-zero `replayed_bytes` cannot represent.
fn cached_scan_bytes() -> &'static HistogramVec {
    static V: OnceLock<HistogramVec> = OnceLock::new();
    V.get_or_init(|| {
        register_histogram_vec!(
            histogram_opts!(
                "polychrome_query_cached_scan_bytes",
                "Total Arrow array memory one query hands to DataFusion — cached rows plus any \
                 freshly-replayed tail, summed across every scoped partition, by scope.",
                vec![
                    1_024.0,
                    16_384.0,
                    131_072.0,
                    1_048_576.0,
                    16_777_216.0,
                    134_217_728.0,
                    536_870_912.0,
                    1_073_741_824.0,
                ]
            ),
            &["scope"]
        )
        .expect("register polychrome_query_cached_scan_bytes")
    })
}

/// Count of queries refused for exceeding
/// `crate::engine::QueryLimits::max_source_events` (QRY-3,
/// `crate::authority::ScopedQuery::enforce_source_budget`), labeled `scope`.
fn source_budget_exceeded_total() -> &'static IntCounterVec {
    static V: OnceLock<IntCounterVec> = OnceLock::new();
    V.get_or_init(|| {
        register_int_counter_vec!(
            "polychrome_query_source_budget_exceeded_total",
            "Count of queries refused for exceeding the pre-execution source-event budget, by \
             scope.",
            &["scope"]
        )
        .expect("register polychrome_query_source_budget_exceeded_total")
    })
}

/// Record one query's observed replay size: `events` total events and
/// `bytes` total payload bytes, labeled `scope`. Call on every query whose
/// replay COMPLETED — success or an event-count budget rejection alike —
/// right after replay, before the count budget check. A query that trips the
/// BYTE budget aborts mid-replay before its full event count is knowable, so
/// it records only [`record_query_replayed_bytes`] instead of this.
pub(crate) fn record_query_observed(scope: &str, events: usize, bytes: u64) {
    #[allow(
        clippy::cast_precision_loss,
        reason = "a metrics observation, not an exact accounting value"
    )]
    source_events()
        .with_label_values(&[scope])
        .observe(events as f64);
    record_query_replayed_bytes(scope, bytes);
}

/// Record just the replayed-byte size of one query, labeled `scope`, without
/// the event-count observation. Used when the byte budget aborts replay
/// mid-stream: the accumulated `bytes` (at or just past the budget) is the
/// exact sizing signal an operator needs to decide whether to raise
/// `crate::engine::QueryLimits::max_source_bytes`, but the total event count
/// is unknowable because replay stopped early.
pub(crate) fn record_query_replayed_bytes(scope: &str, bytes: u64) {
    #[allow(
        clippy::cast_precision_loss,
        reason = "a metrics observation, not an exact accounting value"
    )]
    replayed_bytes()
        .with_label_values(&[scope])
        .observe(bytes as f64);
}

/// Record one query refused for exceeding the source-event budget, labeled
/// `scope`.
pub(crate) fn record_source_budget_exceeded(scope: &str) {
    source_budget_exceeded_total()
        .with_label_values(&[scope])
        .inc();
}

/// Record one query's effective cached-scan volume: `events` and `bytes`
/// [`crate::engine::QueryEngine::build_from_tables`] actually hands
/// `DataFusion`, labeled `scope` — its own labeled dimension, deliberately
/// separate from [`record_query_observed`]'s REPLAYED numbers (see this
/// module's doc and [`cached_scan_events`]'s own doc for why folding a cache
/// hit's near-zero replay into this quantity would make the hit invisible).
/// Called on EVERY query alongside [`record_query_observed`] — the same
/// on-every-query observability contract, extended to cover what
/// `crate::cache`'s decode cache made observable-but-unobserved.
pub(crate) fn record_cached_scan_volume(scope: &str, events: usize, bytes: u64) {
    #[allow(
        clippy::cast_precision_loss,
        reason = "a metrics observation, not an exact accounting value"
    )]
    cached_scan_events()
        .with_label_values(&[scope])
        .observe(events as f64);
    #[allow(
        clippy::cast_precision_loss,
        reason = "a metrics observation, not an exact accounting value"
    )]
    cached_scan_bytes()
        .with_label_values(&[scope])
        .observe(bytes as f64);
}

/// Count of `crate::cache::DecodeCache::lookup` calls served entirely from
/// the cache — zero replay, zero decode.
fn cache_hit_total() -> &'static IntCounter {
    static V: OnceLock<IntCounter> = OnceLock::new();
    V.get_or_init(|| {
        register_int_counter!(
            "polychrome_query_cache_hit_total",
            "Count of decode-cache lookups served entirely from the cache, with no replay or \
             decode."
        )
        .expect("register polychrome_query_cache_hit_total")
    })
}

/// Count of `crate::cache::DecodeCache::lookup` calls that replayed and
/// decoded only a tail onto an already-cached partition.
fn cache_tail_total() -> &'static IntCounter {
    static V: OnceLock<IntCounter> = OnceLock::new();
    V.get_or_init(|| {
        register_int_counter!(
            "polychrome_query_cache_tail_total",
            "Count of decode-cache lookups that replayed and decoded only a tail onto an \
             already-cached partition."
        )
        .expect("register polychrome_query_cache_tail_total")
    })
}

/// Count of `crate::cache::DecodeCache::lookup` calls that forced a full
/// replay and decode — no prior entry, or a mutation-epoch mismatch (the
/// erasure-safety case; see `crate::cache`'s module doc).
fn cache_full_rebuild_total() -> &'static IntCounter {
    static V: OnceLock<IntCounter> = OnceLock::new();
    V.get_or_init(|| {
        register_int_counter!(
            "polychrome_query_cache_full_rebuild_total",
            "Count of decode-cache lookups that forced a full replay and decode: no prior entry, \
             or a mutation-epoch mismatch."
        )
        .expect("register polychrome_query_cache_full_rebuild_total")
    })
}

/// Count of partitions the decode cache evicted — LRU-over-budget or
/// proactive hygiene on a destroy/rewrite/migrate notification (see
/// `crate::cache`'s module doc).
fn cache_eviction_total() -> &'static IntCounter {
    static V: OnceLock<IntCounter> = OnceLock::new();
    V.get_or_init(|| {
        register_int_counter!(
            "polychrome_query_cache_eviction_total",
            "Count of partitions the decode cache evicted, by LRU-over-budget or proactive \
             hygiene on a partition mutation."
        )
        .expect("register polychrome_query_cache_eviction_total")
    })
}

/// Count of queries refused for exceeding
/// `crate::cache::CacheConfig::max_cached_source_events` (the cached-scan
/// volume bound, `crate::authority::ScopedQuery::enforce_cached_volume_budget`),
/// labeled `scope` — the cache-aware counterpart to
/// [`source_budget_exceeded_total`].
fn cached_volume_budget_exceeded_total() -> &'static IntCounterVec {
    static V: OnceLock<IntCounterVec> = OnceLock::new();
    V.get_or_init(|| {
        register_int_counter_vec!(
            "polychrome_query_cached_volume_budget_exceeded_total",
            "Count of queries refused for exceeding the cached-scan volume budget, by scope.",
            &["scope"]
        )
        .expect("register polychrome_query_cached_volume_budget_exceeded_total")
    })
}

/// Every `reason` label value the search index's refusal counter carries —
/// `crate::search_index::worker::UnavailableReason::label`'s own match arms,
/// restated here so [`force`] can pre-create every child. Reason only: never a
/// conversation id or partition name, which on a fleet-wide deployment is
/// unbounded cardinality on the one series an alert reads.
const SEARCH_UNAVAILABLE_REASONS: [&str; 6] = [
    "replay_budget_exceeded",
    "record_too_large",
    "replay_failed",
    "source_unreadable",
    "source_empty",
    "store_failed",
];

/// Every `cause` label value the search index's rebuild counter carries —
/// `crate::search_index::worker::RebuildCause::label`'s own match arms.
const SEARCH_REBUILD_CAUSES: [&str; 2] = ["excision", "unavailable_recovery"];

/// Count of conversations the search index left unsearchable, by reason.
///
/// The alertable series for a class of outage that was otherwise invisible:
/// under all-or-nothing coverage, one refused conversation refuses search for
/// everyone who took part in it, and a non-transient reason is never retried on
/// its own. Before this the only trace was a `tracing::warn!`.
fn search_index_unavailable_total() -> &'static IntCounterVec {
    static V: OnceLock<IntCounterVec> = OnceLock::new();
    V.get_or_init(|| {
        register_int_counter_vec!(
            "polychrome_search_index_unavailable_total",
            "Count of conversations the search index left unsearchable, by reason.",
            &["reason"]
        )
        .expect("register polychrome_search_index_unavailable_total")
    })
}

/// Count of conversations the search index rebuilt because a forward pass
/// could not repair them, by cause.
fn search_index_rebuild_total() -> &'static IntCounterVec {
    static V: OnceLock<IntCounterVec> = OnceLock::new();
    V.get_or_init(|| {
        register_int_counter_vec!(
            "polychrome_search_index_rebuild_total",
            "Count of search-index rebuilds a forward pass escalated to, by cause.",
            &["cause"]
        )
        .expect("register polychrome_search_index_rebuild_total")
    })
}

/// Count of indexing passes that could not advance a conversation's watermark
/// because a turn opened and never completed.
///
/// A steady nonzero rate on this is the signal for an orphaned dispatch or an
/// approval nobody answered: every committed turn behind that barrier is
/// unindexable, and the conversation's coverage keeps reading searchable while
/// it happens.
fn search_index_barrier_held_total() -> &'static IntCounter {
    static V: OnceLock<IntCounter> = OnceLock::new();
    V.get_or_init(|| {
        register_int_counter!(
            "polychrome_search_index_barrier_held_total",
            "Count of search-index passes pinned behind a turn that opened and never completed."
        )
        .expect("register polychrome_search_index_barrier_held_total")
    })
}

/// Count of search-index reconcile sweeps, by whether the sweep was allowed to
/// clear the degraded flag.
///
/// A rising `cleared="false"` is a fleet that is refusing every search and
/// cannot get itself back — the failure mode a truncated sweep, an unrecorded
/// refusal, or an outstanding removal each produce.
fn search_index_reconcile_total() -> &'static IntCounterVec {
    static V: OnceLock<IntCounterVec> = OnceLock::new();
    V.get_or_init(|| {
        register_int_counter_vec!(
            "polychrome_search_index_reconcile_total",
            "Count of search-index reconcile sweeps, by whether the sweep cleared the degraded \
             flag.",
            &["cleared"]
        )
        .expect("register polychrome_search_index_reconcile_total")
    })
}

/// Conversations the search index's reconcile sweep left unsearchable, as of
/// the last sweep.
fn search_index_reconcile_refused() -> &'static IntGauge {
    static V: OnceLock<IntGauge> = OnceLock::new();
    V.get_or_init(|| {
        register_int_gauge!(
            "polychrome_search_index_reconcile_refused",
            "Conversations the last search-index reconcile sweep left unsearchable."
        )
        .expect("register polychrome_search_index_reconcile_refused")
    })
}

/// Partitions the search index still owes work on — its drain lag.
fn search_index_pending_partitions() -> &'static IntGauge {
    static V: OnceLock<IntGauge> = OnceLock::new();
    V.get_or_init(|| {
        register_int_gauge!(
            "polychrome_search_index_pending_partitions",
            "Partitions the search index's dirty set still owes work on."
        )
        .expect("register polychrome_search_index_pending_partitions")
    })
}

/// Whether the search index has lost track of what changed and refuses every
/// search: 1 while degraded, 0 otherwise.
fn search_index_degraded() -> &'static IntGauge {
    static V: OnceLock<IntGauge> = OnceLock::new();
    V.get_or_init(|| {
        register_int_gauge!(
            "polychrome_search_index_degraded",
            "1 while the search index is degraded and refuses every search, 0 otherwise."
        )
        .expect("register polychrome_search_index_degraded")
    })
}

/// Record one conversation left unsearchable, by
/// `crate::search_index::worker::UnavailableReason::label`.
pub(crate) fn record_search_index_unavailable(reason: &str) {
    search_index_unavailable_total()
        .with_label_values(&[reason])
        .inc();
}

/// Record one escalated rebuild, by
/// `crate::search_index::worker::RebuildCause::label`.
pub(crate) fn record_search_index_rebuild(cause: &str) {
    search_index_rebuild_total()
        .with_label_values(&[cause])
        .inc();
}

/// Record one pass pinned behind an open turn (see
/// [`search_index_barrier_held_total`]).
pub(crate) fn record_search_index_barrier_held() {
    search_index_barrier_held_total().inc();
}

/// Record one completed reconcile sweep and what it left behind.
pub(crate) fn record_search_index_reconcile(cleared: bool, refused: usize) {
    search_index_reconcile_total()
        .with_label_values(&[if cleared { "true" } else { "false" }])
        .inc();
    search_index_reconcile_refused().set(i64::try_from(refused).unwrap_or(i64::MAX));
}

/// Record the search index's queue depth and degraded state after a drain or a
/// sweep.
pub(crate) fn record_search_index_queue(pending: usize, degraded: bool) {
    search_index_pending_partitions().set(i64::try_from(pending).unwrap_or(i64::MAX));
    search_index_degraded().set(i64::from(degraded));
}

/// Record a decode-cache hit (see [`cache_hit_total`]).
pub(crate) fn record_cache_hit() {
    cache_hit_total().inc();
}

/// Record a decode-cache tail (see [`cache_tail_total`]).
pub(crate) fn record_cache_tail() {
    cache_tail_total().inc();
}

/// Record a decode-cache full rebuild (see [`cache_full_rebuild_total`]).
pub(crate) fn record_cache_full_rebuild() {
    cache_full_rebuild_total().inc();
}

/// Record a decode-cache eviction (see [`cache_eviction_total`]).
pub(crate) fn record_cache_eviction() {
    cache_eviction_total().inc();
}

/// Record one query refused for exceeding the cached-scan volume budget,
/// labeled `scope`.
pub(crate) fn record_cached_volume_budget_exceeded(scope: &str) {
    cached_volume_budget_exceeded_total()
        .with_label_values(&[scope])
        .inc();
}

/// Force-register every series in this module with every known `scope`
/// label pre-created (zero-valued), so `/metrics` answers a query for query
/// sizing from the first scrape — not only after the first query happens to
/// touch one. See [`crate::init_metrics`].
///
/// A `HistogramVec`/`IntCounterVec` produces NO scrape output for a label
/// combination that has never been touched — registering the vec alone is
/// not enough. `with_label_values` creates the zero-valued child without
/// recording an observation, which is what makes it appear. A plain
/// `IntCounter` (the cache hit/tail/full-rebuild/eviction series, which
/// carry no `scope` label) needs no such pre-creation — registering it is
/// enough for it to scrape as zero — so [`cache_hit_total`]/[`cache_tail_total`]/
/// [`cache_full_rebuild_total`]/[`cache_eviction_total`] are force-initialized
/// by simply calling them, not by iterating a label set.
pub(crate) fn force() {
    for scope in SCOPE_LABELS {
        source_events().with_label_values(&[scope]);
        replayed_bytes().with_label_values(&[scope]);
        cached_scan_events().with_label_values(&[scope]);
        cached_scan_bytes().with_label_values(&[scope]);
        source_budget_exceeded_total().with_label_values(&[scope]);
        cached_volume_budget_exceeded_total().with_label_values(&[scope]);
    }
    for reason in SEARCH_UNAVAILABLE_REASONS {
        search_index_unavailable_total().with_label_values(&[reason]);
    }
    for cause in SEARCH_REBUILD_CAUSES {
        search_index_rebuild_total().with_label_values(&[cause]);
    }
    for cleared in ["true", "false"] {
        search_index_reconcile_total().with_label_values(&[cleared]);
    }
    cache_hit_total();
    cache_tail_total();
    cache_full_rebuild_total();
    cache_eviction_total();
    search_index_barrier_held_total();
    search_index_reconcile_refused();
    search_index_pending_partitions();
    search_index_degraded();
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]

    use prometheus::{Encoder as _, TextEncoder};

    use super::{
        record_cache_eviction, record_cache_full_rebuild, record_cache_hit, record_cache_tail,
        record_cached_scan_volume, record_cached_volume_budget_exceeded, record_query_observed,
        record_query_replayed_bytes, record_search_index_barrier_held, record_search_index_queue,
        record_search_index_rebuild, record_search_index_reconcile,
        record_search_index_unavailable, record_source_budget_exceeded,
    };

    fn scrape() -> String {
        let mut buf = Vec::new();
        TextEncoder::new()
            .encode(&prometheus::default_registry().gather(), &mut buf)
            .expect("encode");
        String::from_utf8(buf).expect("utf8")
    }

    /// Recording an observed query makes both the source-event count and the
    /// replayed-bytes histograms visible in a default-registry scrape, under
    /// the same `scope` label — the same registry `polyc-runtime`'s
    /// `/metrics` handler gathers.
    #[test]
    fn record_query_observed_is_visible_in_a_registry_scrape() {
        record_query_observed("fleet", 42, 4_096);
        let text = scrape();
        assert!(
            text.contains("polychrome_query_source_events_bucket"),
            "missing source-events histogram buckets in scrape:\n{text}"
        );
        assert!(
            text.contains("polychrome_query_replayed_bytes_bucket"),
            "missing replayed-bytes histogram buckets in scrape:\n{text}"
        );
        assert!(text.contains("scope=\"fleet\""));
    }

    /// The search index had no metrics at all: a permanently unavailable
    /// conversation, a fleet stuck degraded, and a reconcile that refuses to
    /// clear were `tracing` lines only, so none of them was alertable. Every
    /// label here is a reason or an outcome, never a conversation.
    #[test]
    fn the_search_index_series_are_visible_in_a_registry_scrape() {
        record_search_index_unavailable("replay_budget_exceeded");
        record_search_index_rebuild("excision");
        record_search_index_barrier_held();
        record_search_index_reconcile(false, 3);
        record_search_index_queue(7, true);

        let text = scrape();
        for series in [
            "polychrome_search_index_unavailable_total",
            "polychrome_search_index_rebuild_total",
            "polychrome_search_index_barrier_held_total",
            "polychrome_search_index_reconcile_total",
            "polychrome_search_index_reconcile_refused",
            "polychrome_search_index_pending_partitions",
            "polychrome_search_index_degraded",
        ] {
            assert!(text.contains(series), "missing {series} in scrape:\n{text}");
        }
        assert!(text.contains("reason=\"replay_budget_exceeded\""));
        assert!(text.contains("cleared=\"false\""));
        assert!(
            text.contains("polychrome_search_index_degraded 1"),
            "the degraded gauge must carry the state, not merely exist:\n{text}"
        );
    }

    /// A byte-budget abort records the replayed-bytes histogram alone (the
    /// over-budget size an operator sizes `max_source_bytes` against), under
    /// its `scope` label — so an over-budget query is visible in the same
    /// distribution as the ones that fit, not silently missing.
    #[test]
    fn record_query_replayed_bytes_is_visible_in_a_registry_scrape() {
        record_query_replayed_bytes("conversations", 20_000_000);
        let text = scrape();
        assert!(
            text.contains("polychrome_query_replayed_bytes_bucket"),
            "missing replayed-bytes histogram buckets in scrape:\n{text}"
        );
        assert!(text.contains("scope=\"conversations\""));
    }

    /// `polychrome_query_source_budget_exceeded_total` increments by exactly
    /// one per recorded rejection, labeled `scope` — so a runaway or
    /// misconfigured caller's refusals are alertable, not just logged.
    #[test]
    fn record_source_budget_exceeded_increments_by_exactly_one() {
        let metric = "polychrome_query_source_budget_exceeded_total";
        let before = labeled_counter_value(&scrape(), metric, "grant");
        record_source_budget_exceeded("grant");
        let after = labeled_counter_value(&scrape(), metric, "grant");
        assert_eq!(
            after - before,
            1.0,
            "{metric}{{scope=\"grant\"}} must increment by exactly 1"
        );
    }

    /// Recording a query's cached-scan volume makes both the effective-event
    /// and effective-byte histograms visible in a default-registry scrape,
    /// under the same `scope` label — the decode cache's counterpart to
    /// [`record_query_observed_is_visible_in_a_registry_scrape`], added for
    /// the same decode-cache metrics (`crate::cache`) this module's other
    /// new series cover.
    #[test]
    fn record_cached_scan_volume_is_visible_in_a_registry_scrape() {
        record_cached_scan_volume("persona", 7, 2_048);
        let text = scrape();
        assert!(
            text.contains("polychrome_query_cached_scan_events_bucket"),
            "missing cached-scan-events histogram buckets in scrape:\n{text}"
        );
        assert!(
            text.contains("polychrome_query_cached_scan_bytes_bucket"),
            "missing cached-scan-bytes histogram buckets in scrape:\n{text}"
        );
        assert!(text.contains("scope=\"persona\""));
    }

    /// `polychrome_query_cached_volume_budget_exceeded_total` increments by
    /// exactly one per recorded rejection, labeled `scope` — the cache-aware
    /// counterpart to
    /// [`record_source_budget_exceeded_increments_by_exactly_one`].
    #[test]
    fn record_cached_volume_budget_exceeded_increments_by_exactly_one() {
        let metric = "polychrome_query_cached_volume_budget_exceeded_total";
        let before = labeled_counter_value(&scrape(), metric, "fleet");
        record_cached_volume_budget_exceeded("fleet");
        let after = labeled_counter_value(&scrape(), metric, "fleet");
        assert_eq!(
            after - before,
            1.0,
            "{metric}{{scope=\"fleet\"}} must increment by exactly 1"
        );
    }

    /// `polychrome_query_cache_hit_total` increments by exactly one per
    /// recorded decode-cache hit. Unlike every other series in this module,
    /// the cache hit/tail/full-rebuild/eviction counters carry no `scope`
    /// label — see [`bare_counter_value`].
    #[test]
    fn record_cache_hit_increments_by_exactly_one() {
        let metric = "polychrome_query_cache_hit_total";
        let before = bare_counter_value(&scrape(), metric);
        record_cache_hit();
        let after = bare_counter_value(&scrape(), metric);
        assert_eq!(after - before, 1.0, "{metric} must increment by exactly 1");
    }

    /// `polychrome_query_cache_tail_total` increments by exactly one per
    /// recorded decode-cache tail replay.
    #[test]
    fn record_cache_tail_increments_by_exactly_one() {
        let metric = "polychrome_query_cache_tail_total";
        let before = bare_counter_value(&scrape(), metric);
        record_cache_tail();
        let after = bare_counter_value(&scrape(), metric);
        assert_eq!(after - before, 1.0, "{metric} must increment by exactly 1");
    }

    /// `polychrome_query_cache_full_rebuild_total` increments by exactly one
    /// per recorded decode-cache full rebuild.
    #[test]
    fn record_cache_full_rebuild_increments_by_exactly_one() {
        let metric = "polychrome_query_cache_full_rebuild_total";
        let before = bare_counter_value(&scrape(), metric);
        record_cache_full_rebuild();
        let after = bare_counter_value(&scrape(), metric);
        assert_eq!(after - before, 1.0, "{metric} must increment by exactly 1");
    }

    /// `polychrome_query_cache_eviction_total` increments by exactly one per
    /// recorded decode-cache eviction.
    #[test]
    fn record_cache_eviction_increments_by_exactly_one() {
        let metric = "polychrome_query_cache_eviction_total";
        let before = bare_counter_value(&scrape(), metric);
        record_cache_eviction();
        let after = bare_counter_value(&scrape(), metric);
        assert_eq!(after - before, 1.0, "{metric} must increment by exactly 1");
    }

    /// Parse a labeled counter's current value for one `metric`/`scope` pair
    /// out of a scrape. The prometheus default registry is process-global
    /// and shared across every test in this binary, so callers diff two
    /// calls to this rather than asserting an absolute count.
    fn labeled_counter_value(text: &str, metric: &str, scope: &str) -> f64 {
        let needle = format!("{metric}{{scope=\"{scope}\"}} ");
        text.lines()
            .find(|line| line.starts_with(&needle))
            .and_then(|line| line.rsplit(' ').next())
            .and_then(|v| v.parse::<f64>().ok())
            .unwrap_or(0.0)
    }

    /// Parse a plain (unlabeled) counter's current value for one `metric` out
    /// of a scrape — the cache hit/tail/full-rebuild/eviction series carry no
    /// `scope` label, unlike every other series in this module. Same
    /// process-global-registry diffing caveat as [`labeled_counter_value`].
    fn bare_counter_value(text: &str, metric: &str) -> f64 {
        let needle = format!("{metric} ");
        text.lines()
            .find(|line| line.starts_with(&needle))
            .and_then(|line| line.rsplit(' ').next())
            .and_then(|v| v.parse::<f64>().ok())
            .unwrap_or(0.0)
    }
}