sozu-lib 2.3.0

//! Metrics façade.
//!
//! Defines the per-thread `Aggregator`, the `incr!`/`count!`/`gauge!`/
//! `gauge_add!`/`time!` macros consumed across the lib + bin crates, the
//! local-vs-network drain dispatch, and the label allow/deny filtering
//! used to keep cardinality bounded. Gauge underflow is treated as a
//! correctness bug (saturating clamp + warn log), not a rounding artefact.

mod local_drain;
pub mod names;
mod network_drain;
mod writer;

use std::{
    cell::RefCell,
    collections::{BTreeMap, HashMap},
    io::{self, Write},
    net::SocketAddr,
    num::NonZeroU64,
    str,
    time::{Duration, Instant},
};

use mio::net::UdpSocket;
use sozu_command::config::MetricDetailLevel;
use sozu_command::proto::command::{
    FilteredMetrics, MetricsConfiguration, QueryMetricsOptions, ResponseContent,
};

use crate::metrics::{local_drain::LocalDrain, network_drain::NetworkDrain};

/// Filter `(cluster_id, backend_id)` labels at emission time according to the
/// configured [`MetricDetailLevel`]. Each level is a SUPERSET of the previous
/// in cardinality terms:
///
/// - `Process`: drop both labels — proxy-only counters (smallest keyspace).
/// - `Frontend`: same as `Process` today; reserved for the per-listener
///   counters tracked as a follow-up. Listed in the proto + config so
///   operators can opt in once per-listener labels land.
/// - `Cluster`: keep `cluster_id`, drop `backend_id` — current default.
/// - `Backend`: keep both — current historical behaviour.
///
/// Pure function so the filter can be unit-tested exhaustively without the
/// drain machinery in scope.
fn filter_labels_for_detail<'a>(
    detail: MetricDetailLevel,
    cluster_id: Option<&'a str>,
    backend_id: Option<&'a str>,
) -> (Option<&'a str>, Option<&'a str>) {
    let (out_cluster, out_backend) = match detail {
        MetricDetailLevel::Process | MetricDetailLevel::Frontend => (None, None),
        MetricDetailLevel::Cluster => (cluster_id, None),
        MetricDetailLevel::Backend => (cluster_id, backend_id),
    };
    // The filter only ever DROPS labels: an absent input can never become a
    // present output, and a kept output must be the exact same borrow we
    // were handed (no synthesised label). This is the cardinality-safety
    // post-condition that keeps the local CLI view and the wire consistent.
    debug_assert!(
        cluster_id.is_some() || out_cluster.is_none(),
        "filter must never invent a cluster_id"
    );
    debug_assert!(
        backend_id.is_some() || out_backend.is_none(),
        "filter must never invent a backend_id"
    );
    debug_assert!(
        out_backend.is_none() || out_cluster.is_some(),
        "a kept backend label implies a kept cluster label (no orphan backend)"
    );
    (out_cluster, out_backend)
}

/// Map a numeric HTTP status code to its dedicated counter name, if any.
///
/// Returns `Some("http.status.<code>")` for the eighteen status codes Sōzu
/// either generates as a default answer or that operators routinely chart
/// (`200/201/204`, `301/302/304`, `400/401/403/404/408/413/429`, plus the
/// `5xx` family Sōzu can synthesise — `500/502/503/504/507`). All other
/// codes return `None` so the bucket counter (`http.status.{1xx,…,other}`)
/// remains the sole emission and metric cardinality stays bounded.
///
/// Hoisted out of the protocol modules so every emission site
/// (`mux::stream::generate_access_log`, shared by the H1 and H2 mux paths)
/// cannot drift on which codes get a per-code counter. It was shared with
/// `kawa_h1::save_http_status_metric` until that unreachable H1 session was
/// removed on 2026-09-20 (sozu#1346).
pub(crate) fn http_status_code_metric_name(status: u16) -> Option<&'static str> {
    match status {
        200 => Some("http.status.200"),
        201 => Some("http.status.201"),
        204 => Some("http.status.204"),
        301 => Some("http.status.301"),
        302 => Some("http.status.302"),
        304 => Some("http.status.304"),
        400 => Some("http.status.400"),
        401 => Some("http.status.401"),
        403 => Some("http.status.403"),
        404 => Some("http.status.404"),
        408 => Some("http.status.408"),
        413 => Some("http.status.413"),
        429 => Some("http.status.429"),
        500 => Some("http.status.500"),
        502 => Some("http.status.502"),
        503 => Some("http.status.503"),
        504 => Some("http.status.504"),
        507 => Some("http.status.507"),
        _ => None,
    }
}

thread_local! {
  pub static METRICS: RefCell<Aggregator> = RefCell::new(Aggregator::new(String::from("sozu")));
}

/// Opaque identity of one active lifetime of a cluster metrics row.
///
/// A lifetime starts either when a legal route first references a cluster or
/// when `AddCluster` arrives first. `AddCluster` without an intervening
/// `RemoveCluster` keeps the identity so gauges opened by the implicit route
/// remain balanced. A remove followed by another route or add allocates a new
/// identity, fencing late metrics from the previous lifetime.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) struct ClusterMetricsIncarnation(NonZeroU64);

/// Identity state captured by one delayed metric owner.
///
/// `Missing` is the default before routing. `Invalid` records the terminal
/// allocation-exhaustion path. Neither may be mistaken for a legal implicit
/// cluster lifetime when labels are retained.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(crate) enum ClusterMetricsCapture {
    #[default]
    Missing,
    Active(ClusterMetricsIncarnation),
    Invalid,
}

/// Capture (and, for a legal first implicit reference, allocate) the active
/// metrics identity for `cluster_id`.
pub(crate) fn capture_cluster_metrics(cluster_id: &str) -> ClusterMetricsCapture {
    METRICS.with(|metrics| metrics.borrow_mut().capture_cluster_metrics(cluster_id))
}

#[derive(thiserror::Error, Debug)]
pub enum MetricError {
    #[error("Could not parse udp address {address}: {error}")]
    WrongUdpAddress { address: String, error: String },
    #[error("Could not bind to udp address {address}: {error}")]
    UdpBind { address: String, error: String },
    #[error("No metrics found for object with id {0}")]
    NoMetrics(String),
    #[error("Could not create histogram for time metric {time_metric:?}: {error}")]
    HistogramCreation {
        time_metric: MetricValue,
        error: String,
    },
    #[error("could not record time metric {time_metric:?}: {error}")]
    TimeMetricRecordingError {
        time_metric: MetricValue,
        error: String,
    },
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MetricValue {
    Gauge(usize),
    GaugeAdd(i64),
    Count(i64),
    Time(usize),
}

impl MetricValue {
    fn is_time(&self) -> bool {
        matches!(self, &MetricValue::Time(_))
    }

    fn update(&mut self, key: &'static str, m: MetricValue) -> bool {
        match (self, m) {
            (&mut MetricValue::Gauge(ref mut v1), MetricValue::Gauge(v2)) => {
                // `changed` reflects the absolute set: true iff the stored
                // value actually moves. Pair-assert the post-state matches
                // the requested value and that `changed` is the equality
                // relationship between before and after.
                let before = *v1;
                let changed = *v1 != v2;
                *v1 = v2;
                debug_assert_eq!(*v1, v2, "gauge set must store exactly the requested value");
                debug_assert_eq!(
                    changed,
                    before != v2,
                    "`changed` must report whether the gauge actually moved"
                );
                changed
            }
            (&mut MetricValue::Gauge(ref mut v1), MetricValue::GaugeAdd(v2)) => {
                // Saturating clamp + `error!` log on underflow, symmetric
                // with `AggregatedMetric::update` in `local_drain.rs`. The
                // `debug_assert!` that lived here previously made debug
                // builds panic while release silently clamped; H2
                // `ConnectionH2::Drop` rebalance + the saturating clamps
                // make underflow structurally impossible from live code,
                // and a panic on a metric mismatch is too aggressive — the
                // log line names the offending key for observability.
                //
                // INVARIANT (the canonical gauge-underflow guard): the post
                // value is non-negative, equals `before + v2` on the
                // non-underflow path, and is exactly the `0` floor whenever
                // `before + v2` would have gone negative. We assert the
                // *post-condition* (never panic the recovery path) rather
                // than a `before >= -v2` pre-condition, because a real
                // underflow is recovered (clamp + log) by design.
                let before = *v1;
                let changed = v2 != 0;
                let res = before as i64 + v2;
                *v1 = if res >= 0 {
                    res as usize
                } else {
                    error!(
                        "metric {} underflow: previous value: {}, adding: {}",
                        key, before, v2
                    );
                    0
                };
                debug_assert!(
                    res >= 0 || *v1 == 0,
                    "gauge underflow must clamp to exactly 0 (never wrap)"
                );
                debug_assert!(
                    res < 0 || *v1 as i64 == before as i64 + v2,
                    "non-underflow gauge add must equal before + v2 exactly"
                );

                changed
            }
            (&mut MetricValue::Count(ref mut v1), MetricValue::Count(v2)) => {
                // Counter delta: `v2` may be negative (e.g. `decr!`), so the
                // monotonic-between-resets invariant is encoded as the exact
                // delta — the post value equals the pre value plus `v2`.
                let before = *v1;
                let changed = v2 != 0;
                *v1 += v2;
                debug_assert_eq!(*v1, before + v2, "count update must advance by exactly v2");
                changed
            }
            (s, m) => {
                // Symmetric with `AggregatedMetric::update` in `local_drain.rs`:
                // a type-mismatch is a coding bug (a `metric_name` was
                // re-emitted with a different `MetricValue` variant), not a
                // network-driven event, but log-and-continue is still the
                // right policy — taking the worker process down on a metric
                // bookkeeping mismatch is too aggressive for a single-bug
                // recovery path. The log line names the offending key.
                error!(
                    "tried to update metric {} of value {:?} with an incompatible metric: {:?}",
                    key, s, m
                );
                false
            }
        }
    }
}

#[derive(Debug, Clone)]
pub struct StoredMetricValue {
    last_sent: Instant,
    updated: bool,
    data: MetricValue,
}

impl StoredMetricValue {
    pub fn new(last_sent: Instant, data: MetricValue) -> StoredMetricValue {
        // Symmetric with `AggregatedMetric::new`: a first emission with a
        // negative `GaugeAdd` is a coding bug (a `-1` ran with no paired
        // `+1` earlier in the worker's lifetime); clamp to 0 and log so the
        // offending pattern shows up in operator logs alongside the
        // local-drain side.
        let data = if let MetricValue::GaugeAdd(v) = data {
            if v >= 0 {
                MetricValue::Gauge(v as usize)
            } else {
                error!(
                    "stored metric created with negative GaugeAdd({}), clamping to 0",
                    v
                );
                MetricValue::Gauge(0)
            }
        } else {
            data
        };
        // Post-conditions: a `GaugeAdd` seed never survives as `GaugeAdd`
        // (it is normalised to a non-negative `Gauge`), and a fresh stored
        // value is always flagged `updated` so the next drain emits it.
        debug_assert!(
            !matches!(data, MetricValue::GaugeAdd(_)),
            "GaugeAdd must be normalised to a non-negative Gauge at construction"
        );
        StoredMetricValue {
            last_sent,
            updated: true,
            data,
        }
    }

    pub fn update(&mut self, key: &'static str, m: MetricValue) {
        // The `updated` flag is a monotone latch within one drain cycle: it
        // may only go from false→true here (the drain resets it to false on
        // emit). So once set, it stays set; and if this update reports a
        // change, the flag must be set afterwards.
        let was_updated = self.updated;
        let updated = self.data.update(key, m);
        if !self.updated {
            self.updated = updated;
        }
        debug_assert!(
            self.updated || (!was_updated && !updated),
            "`updated` must stay set once latched; a reported change must latch it"
        );
        debug_assert!(
            !was_updated || self.updated,
            "`updated` is a monotone latch within a drain cycle (never cleared here)"
        );
    }
}

pub fn setup<O: Into<String>>(
    metrics_host: &SocketAddr,
    origin: O,
    use_tagged_metrics: bool,
    prefix: Option<String>,
    detail: MetricDetailLevel,
) -> Result<(), MetricError> {
    let metrics_socket = udp_bind()?;

    debug!(
        "setting up metrics: local address = {:#?}",
        metrics_socket.local_addr()
    );

    METRICS.with(|metrics| {
        if let Some(p) = prefix {
            (*metrics.borrow_mut()).set_up_prefix(p);
        }
        (*metrics.borrow_mut()).set_up_remote(metrics_socket, *metrics_host);
        (*metrics.borrow_mut()).set_up_origin(origin.into());
        (*metrics.borrow_mut()).set_up_tagged_metrics(use_tagged_metrics);
        (*metrics.borrow_mut()).set_up_detail(detail);
    });
    Ok(())
}

pub trait Subscriber {
    fn receive_metric(
        &mut self,
        label: &'static str,
        cluster_id: Option<&str>,
        backend_id: Option<&str>,
        metric: MetricValue,
    );
}

/// How often `lease_tick` actually does work; cheaper than recomputing the
/// effective level on every metric emission. Polled once per worker event-loop
/// iteration (which wakes at least once per one-second poll timeout) and on
/// every worker command, so an expired lease is retired within this interval
/// plus one poll timeout, with or without traffic or commands.
const LEASE_TICK_INTERVAL: Duration = Duration::from_secs(5);

/// Hard cap on lease TTL, mirroring the proto comment on `SetMetricDetail`.
/// Bounds the worst-case effect of a stuck renewer.
pub const LEASE_TTL_MAX: Duration = Duration::from_secs(300);

/// Default lease TTL applied when the proto request omits `ttl_seconds` (or
/// passes `0`). Matches the proto comment.
pub const LEASE_TTL_DEFAULT: Duration = Duration::from_secs(60);

/// Hard cap on the number of simultaneous leases held by the aggregator.
/// `lease_apply` rejects new entries (with [`LeaseApplyOutcome::TableFull`])
/// once the table reaches this size. Bounds the lease table's memory and
/// neutralises the CWE-770 vector where a same-UID attacker rolls
/// `client_id` faster than expiry to grow the map unbounded. 64 is well
/// above any plausible TUI fleet (one TUI per operator + a handful of
/// metric scrapers); legitimate callers renewing the same `client_id`
/// REPLACE rather than insert and therefore don't bump the count.
pub const LEASE_TABLE_CAP: usize = 64;

/// Hard cap on `client_id` length accepted by `lease_apply`. The TUI
/// uses `top:<pid>:<8 hex chars>` ≤ 24 bytes; 64 leaves headroom for
/// other operator-side scrapers while keeping the lease table's per-
/// entry memory bounded and the audit-log lease_id column small.
pub const LEASE_CLIENT_ID_MAX_BYTES: usize = 64;

/// Outcome of [`Aggregator::lease_apply`]. The success arm returns the
/// `(previous_effective, new_effective)` pair the caller can use to
/// decide whether to emit a `MetricDetailChanged` audit event; the
/// failure arms surface bounded-input rejections.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum LeaseApplyOutcome {
    /// Lease inserted / renewed.
    Applied {
        previous_effective: MetricDetailLevel,
        new_effective: MetricDetailLevel,
    },
    /// `client_id` length exceeds [`LEASE_CLIENT_ID_MAX_BYTES`].
    ClientIdTooLong,
    /// Lease table is at [`LEASE_TABLE_CAP`] and the request is a new
    /// insert (not a renewal). Callers MUST surface this as a `FAILURE`
    /// to the wire so the lessor can back off.
    TableFull,
    /// The requested TTL exceeds [`LEASE_TTL_MAX`]. This arm is
    /// theoretically unreachable in the normal flow because the
    /// dispatch site rejects out-of-range values before reaching the
    /// aggregator; surfacing it explicitly catches callers that bypass
    /// dispatch (proto fuzzing, future internal use) instead of
    /// silently clamping their intent.
    TtlOutOfRange,
    /// A renewal request was presented with a [`PeerBinding`] that
    /// disagrees with the existing lease's apply-time binding. Returned
    /// only when the existing binding is fully known (per
    /// [`PeerBinding::is_known`]); unknown-binding leases (no
    /// `SO_PEERCRED` available, pre-binding callers) continue to
    /// accept any renewer. Closes the same-UID `client_id`-collision
    /// takeover where an attacker re-applies a lease against a
    /// victim's id and replaces the binding to lock the victim out of
    /// their own clear.
    Unauthorized,
}

/// Master-populated peer binding stored alongside each lease so subsequent
/// `clear` requests can be authorised against the apply-time owner. The
/// binding pairs an OS pid (from `SO_PEERCRED` on Linux, captured by the
/// master at command-socket accept time) with the master-side connection
/// session ULID. A clear request must present BOTH values matching the
/// apply-time binding. A binding of `(None, None)` ("unknown") at apply
/// time means the connection had no peer credentials available — the
/// worker then accepts any clear for that `client_id` to preserve compat
/// with non-Linux callers and intermediate proxies. See the proto comment
/// on `SetMetricDetail.peer_pid` for the trust model.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct PeerBinding {
    pub pid: Option<i32>,
    /// Session ULID rendered as a `u128` (the master's session anchor).
    /// Stored as the raw u128 rather than the original `Ulid` to avoid
    /// dragging that crate into the metrics-aggregator dependencies.
    pub session_ulid: Option<u128>,
}

impl PeerBinding {
    /// `true` when both halves are known. A fully-known binding is the
    /// only one against which `clear` may be authorised; partial bindings
    /// (one half `None`) degrade to "accept any clear" per the proto
    /// contract on `SetMetricDetail.peer_pid` / `peer_session_ulid`.
    pub fn is_known(&self) -> bool {
        self.pid.is_some() && self.session_ulid.is_some()
    }

    /// True when `self` and `other` are compatible (same `pid` + same
    /// `session_ulid`) AND `self.is_known()`. Used by `lease_clear` to
    /// reject mismatched clears.
    pub fn matches(&self, other: &PeerBinding) -> bool {
        self.is_known() && self.pid == other.pid && self.session_ulid == other.session_ulid
    }
}

/// One lease entry kept inside [`Aggregator::leases`]. Carries the
/// requested cardinality level, the absolute expiry instant, and the
/// master-supplied [`PeerBinding`] captured at apply time.
#[derive(Clone, Copy, Debug)]
struct LeaseEntry {
    level: MetricDetailLevel,
    expires_at: Instant,
    binding: PeerBinding,
}

/// Outcome of [`Aggregator::lease_clear`].
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum LeaseClearOutcome {
    /// The lease was found, the binding matched (or was unknown at apply
    /// time), and the entry has been removed. Carries the worker's
    /// previous effective level so the caller can decide whether to
    /// emit a `MetricDetailChanged` audit event.
    Cleared {
        previous_effective: MetricDetailLevel,
    },
    /// No lease was held by the requested `client_id`. Silent no-op.
    NotFound,
    /// A lease existed but the peer binding presented with the clear did
    /// not match the apply-time binding. The lease is left intact. The
    /// worker MUST surface this as a `FAILURE` response to discourage
    /// guessing attacks against another operator's lease.
    Unauthorized,
}

pub struct Aggregator {
    /// appended to metric keys, usually "sozu-"
    prefix: String,
    /// gathers metrics and sends them on a UDP socket
    network: Option<NetworkDrain>,
    /// gather metrics locally, queried by the CLI
    local: LocalDrain,
    /// Static cardinality knob — set once at boot from `MetricsConfig.detail`.
    /// Filters `(cluster_id, backend_id)` labels at emission time. Each level
    /// is a SUPERSET of the previous one (`Process ⊆ Frontend ⊆ Cluster ⊆ Backend`).
    configured: MetricDetailLevel,
    /// Effective cardinality knob actually applied at emission. Equal to
    /// `max(configured, max(active leases))`. Recomputed only when leases
    /// change, so the hot path (`receive_metric`) reads a single field.
    effective: MetricDetailLevel,
    /// Active TTL leases keyed by `client_id` from `SetMetricDetail`. A live
    /// lease holds the worker's effective level at-or-above its requested
    /// detail until it expires or is explicitly cleared. Multiple clients
    /// (e.g. several `sozu top` sessions) lease independently.
    leases: HashMap<String, LeaseEntry>,
    /// Wall-clock anchor for the polled lease janitor. Updated on every
    /// `lease_tick` call regardless of whether expiry happened, so the
    /// caller's "is it time to tick?" check stays cheap.
    last_lease_tick: Instant,
    /// Active lifetime for each cluster id. A legal implicit route allocates
    /// an entry before `AddCluster`; the add reuses it because only
    /// `RemoveCluster` is a lifetime boundary. This identity deliberately
    /// survives metrics collection being disabled or local storage being
    /// cleared.
    cluster_incarnations: HashMap<String, ClusterMetricsIncarnation>,
    /// Last allocated incarnation. Zero is reserved for `Option`'s niche and
    /// checked arithmetic makes exhaustion fail closed instead of reusing an
    /// old identity.
    last_cluster_incarnation: u64,
}

impl Aggregator {
    pub fn new(prefix: String) -> Aggregator {
        let default_detail = MetricDetailLevel::default();
        Aggregator {
            prefix: prefix.clone(),
            network: None,
            local: LocalDrain::new(prefix),
            configured: default_detail,
            effective: default_detail,
            leases: HashMap::new(),
            last_lease_tick: Instant::now(),
            cluster_incarnations: HashMap::new(),
            last_cluster_incarnation: 0,
        }
    }

    pub fn set_up_prefix(&mut self, prefix: String) {
        self.prefix = prefix;
    }

    pub fn set_up_remote(&mut self, socket: UdpSocket, addr: SocketAddr) {
        self.network = Some(NetworkDrain::new(self.prefix.clone(), socket, addr));
    }

    pub fn set_up_origin(&mut self, origin: String) {
        if let Some(n) = self.network.as_mut() {
            n.origin = origin;
        }
    }

    pub fn set_up_tagged_metrics(&mut self, tagged: bool) {
        if let Some(n) = self.network.as_mut() {
            n.use_tagged_metrics = tagged;
        }
    }

    /// Set the static cardinality floor (`MetricsConfig.detail` from the TOML
    /// configuration). Live leases applied via [`Self::lease_apply`] can
    /// elevate the effective level at runtime; the configured floor is the
    /// lower bound the worker falls back to when no lease is active.
    ///
    /// See [`MetricDetailLevel`] and `filter_labels_for_detail` for the
    /// per-level filtering rules.
    pub fn set_up_detail(&mut self, detail: MetricDetailLevel) {
        self.configured = detail;
        self.recompute_effective();
        // Raising the configured floor can only raise/hold the effective
        // level; the full invariant set must hold afterwards.
        debug_assert!(
            self.effective >= self.configured,
            "effective must dominate the freshly-set configured floor"
        );
        #[cfg(debug_assertions)]
        self.check_lease_invariants();
    }

    /// Returns the static (configured) cardinality floor. Independent of
    /// runtime leases.
    pub fn detail_configured(&self) -> MetricDetailLevel {
        self.configured
    }

    /// Returns the cardinality level actually applied to emissions. Equal to
    /// `max(configured, max(active leases))`.
    pub fn detail_effective(&self) -> MetricDetailLevel {
        self.effective
    }

    /// Apply or renew a runtime cardinality lease for `client_id`. If a lease
    /// for the same client already exists it is REPLACED (used for renewals).
    /// `ttl` values above [`LEASE_TTL_MAX`] are **rejected** with
    /// [`LeaseApplyOutcome::TtlOutOfRange`] (NOT clamped); callers must
    /// handle that arm or pre-validate the TTL. On success the call returns
    /// `(previous_effective, new_effective)` so callers can decide whether
    /// to emit a `MetricDetailChanged` audit event.
    ///
    /// The proto contract on `SetMetricDetail.ttl_seconds` is that the worker
    /// **rejects** out-of-range values with a `FAILURE` response — that
    /// enforcement lives at the dispatch site in `lib/src/server.rs::notify`.
    /// `lease_apply` itself returns [`LeaseApplyOutcome::TtlOutOfRange`]
    /// when called with `ttl > LEASE_TTL_MAX` so callers that bypass the
    /// dispatch site (proto fuzzing, future internal use) see a loud
    /// rejection instead of silently capped semantics. Same shape for
    /// over-long `client_id` and a full lease table — see
    /// [`LeaseApplyOutcome`] for the failure arms.
    ///
    /// `now` is parameterised exactly like [`Self::lease_tick`] so the whole
    /// lease lifecycle (apply, tick) is driven off one caller-supplied clock
    /// reading instead of `Instant::now()` on the datapath — the same
    /// sans-io shape as `UdpManager::handle_input` — letting a deterministic
    /// simulation drive `lease_apply` under a virtual clock. `SessionMetrics`
    /// and the timer wheel elsewhere in this module intentionally keep the
    /// real clock and are out of scope for this parameterisation.
    pub fn lease_apply(
        &mut self,
        client_id: String,
        level: MetricDetailLevel,
        ttl: Duration,
        binding: PeerBinding,
        now: Instant,
    ) -> LeaseApplyOutcome {
        if client_id.len() > LEASE_CLIENT_ID_MAX_BYTES {
            return LeaseApplyOutcome::ClientIdTooLong;
        }
        if ttl > LEASE_TTL_MAX {
            return LeaseApplyOutcome::TtlOutOfRange;
        }
        // Cap the table size BEFORE the insert, but only when the caller
        // is inserting a fresh entry. Renewals (same `client_id` already
        // present) REPLACE the existing entry and therefore keep the
        // count stable — they must always succeed so an active operator
        // never loses their lease just because the table is full.
        let is_renewal = self.leases.contains_key(&client_id);
        if !is_renewal && self.leases.len() >= LEASE_TABLE_CAP {
            return LeaseApplyOutcome::TableFull;
        }
        // Renewal-binding gate: when a lease already exists for this
        // `client_id` and its apply-time binding is fully known, the
        // renewer's presented binding MUST match. Without this check
        // any same-UID caller that learns the `client_id` (PID from
        // /proc, suffix from audit log) could re-apply against it,
        // overwriting the binding to point at the attacker's session
        // and then driving the victim's Drop-time `clear` into
        // `Unauthorized`. Unknown apply-time bindings continue to
        // accept any renewer per the proto contract on
        // `SetMetricDetail.peer_pid` / `peer_session_ulid`.
        if is_renewal
            && let Some(entry) = self.leases.get(&client_id)
            && entry.binding.is_known()
            && !entry.binding.matches(&binding)
        {
            return LeaseApplyOutcome::Unauthorized;
        }
        let expires_at = now + ttl;
        let before_len = self.leases.len();
        self.leases.insert(
            client_id,
            LeaseEntry {
                level,
                expires_at,
                binding,
            },
        );
        // Map-accounting invariant: a renewal REPLACES in place (count
        // stable) while a fresh insert grows the table by exactly one. The
        // table must never exceed the cap on the insert path because the
        // fresh-insert branch was gated by the `>= LEASE_TABLE_CAP` check
        // above.
        debug_assert_eq!(
            self.leases.len(),
            before_len + (!is_renewal) as usize,
            "lease_apply grows the table by exactly one iff this is a fresh insert"
        );
        debug_assert!(
            self.leases.len() <= LEASE_TABLE_CAP,
            "lease table must never exceed LEASE_TABLE_CAP after an accepted apply"
        );
        let previous_effective = self.effective;
        self.recompute_effective();
        // A lease is a mechanism for a client to temporarily ELEVATE
        // cardinality on its own behalf, not a one-way ratchet on the
        // shared `effective` level: `effective` is, and must stay,
        // `max(configured, max over live leases)`. A renewal at a LOWER
        // level than the client's own previous entry is that client
        // withdrawing part of its own request, so `effective` dropping in
        // response is the correct outcome -- not an invariant violation.
        // The only invariant that must hold here is the configured floor,
        // checked below.
        debug_assert!(
            self.effective >= self.configured,
            "effective must never drop below the configured floor"
        );
        #[cfg(debug_assertions)]
        self.check_lease_invariants();
        LeaseApplyOutcome::Applied {
            previous_effective,
            new_effective: self.effective,
        }
    }

    /// Explicitly release a lease keyed by `client_id`. The clear is
    /// authorised against the apply-time [`PeerBinding`] when one was
    /// recorded — see [`LeaseClearOutcome`] for the three result states.
    /// A clear request with `presented = PeerBinding::default()` matches
    /// only leases whose apply-time binding was also unknown, preserving
    /// compat with pre-binding callers and platforms without
    /// `SO_PEERCRED`.
    pub fn lease_clear(&mut self, client_id: &str, presented: PeerBinding) -> LeaseClearOutcome {
        let Some(entry) = self.leases.get(client_id) else {
            return LeaseClearOutcome::NotFound;
        };
        // If the apply-time binding is fully known we MUST authorise the
        // clear against it; presenting `default()` (an empty binding) is
        // a mismatch. If the apply-time binding is unknown (a pre-binding
        // caller, or a non-Linux peer with no credentials), we permit
        // any clear — there is nothing to authorise against.
        if entry.binding.is_known() && !entry.binding.matches(&presented) {
            return LeaseClearOutcome::Unauthorized;
        }
        // We only reach here after the `get` above proved the key is
        // present, so the removal evicts exactly one entry.
        let before_len = self.leases.len();
        self.leases.remove(client_id);
        debug_assert!(
            !self.leases.contains_key(client_id),
            "lease_clear must evict the cleared client_id"
        );
        debug_assert_eq!(
            self.leases.len(),
            before_len - 1,
            "lease_clear removes exactly one entry on the authorised path"
        );
        let previous = self.effective;
        self.recompute_effective();
        // Clearing a lease can only LOWER or hold the effective level (one
        // contributor to the max is gone), never raise it.
        debug_assert!(
            self.effective <= previous,
            "lease_clear must not raise the effective detail level"
        );
        debug_assert!(
            self.effective >= self.configured,
            "effective must never drop below the configured floor"
        );
        #[cfg(debug_assertions)]
        self.check_lease_invariants();
        LeaseClearOutcome::Cleared {
            previous_effective: previous,
        }
    }

    /// Returns the number of active (non-expired-as-of-last-tick) leases.
    /// Surfaced in `WorkerMetricDetailStatus` so the TUI can show
    /// "another client is still leasing this level".
    pub fn lease_count(&self) -> u32 {
        self.leases.len() as u32
    }

    /// Polled lease-expiry janitor. Called from the worker's event loop on
    /// every iteration and from `notify`, both gated by
    /// [`Self::lease_tick_due`]; cheap when nothing has expired. Returns
    /// `Some(previous_effective)` when at least one lease expired AND that
    /// expiry actually moved the effective level (so the caller can emit a
    /// `MetricDetailChanged` audit event), or `None` for the no-change path.
    ///
    /// `now` is parameterised so unit tests can advance the clock
    /// deterministically without sleeping.
    pub fn lease_tick(&mut self, now: Instant) -> Option<MetricDetailLevel> {
        self.last_lease_tick = now;
        let before = self.leases.len();
        self.leases.retain(|_, entry| entry.expires_at > now);
        // Expiry is a pure shrink: `retain` can only drop entries, never
        // add them, and every surviving lease must be strictly in the
        // future relative to `now`.
        debug_assert!(
            self.leases.len() <= before,
            "lease_tick (expiry) can only shrink the table, never grow it"
        );
        debug_assert!(
            self.leases.values().all(|entry| entry.expires_at > now),
            "every surviving lease must expire strictly after `now`"
        );
        if self.leases.len() == before {
            return None;
        }
        let previous = self.effective;
        self.recompute_effective();
        // Losing leases can only lower or hold the effective level.
        debug_assert!(
            self.effective <= previous,
            "lease expiry must not raise the effective detail level"
        );
        debug_assert!(
            self.effective >= self.configured,
            "effective must never drop below the configured floor"
        );
        #[cfg(debug_assertions)]
        self.check_lease_invariants();
        if previous != self.effective {
            Some(previous)
        } else {
            None
        }
    }

    /// True when at least `LEASE_TICK_INTERVAL` has passed since the last
    /// `lease_tick`. Gates the polled janitor so the worker's event loop
    /// does not pay a HashMap walk on every iteration.
    pub fn lease_tick_due(&self, now: Instant) -> bool {
        now.duration_since(self.last_lease_tick) >= LEASE_TICK_INTERVAL
    }

    /// Recompute `effective = max(configured, max(active leases))`. Cheap (one
    /// linear pass over the lease table) and only called on apply/clear/tick,
    /// never on the metric-emission hot path.
    fn recompute_effective(&mut self) {
        let mut max_lease = self.configured;
        for entry in self.leases.values() {
            if entry.level > max_lease {
                max_lease = entry.level;
            }
        }
        self.effective = max_lease;
        // Post-condition: effective is the floor AND every active lease's
        // ceiling — i.e. it is at least `configured` and at least each
        // lease level, and it equals one of those contributors (max).
        debug_assert!(
            self.effective >= self.configured,
            "effective is bounded below by the configured floor"
        );
        debug_assert!(
            self.leases.values().all(|e| self.effective >= e.level),
            "effective dominates every active lease level"
        );
        debug_assert!(
            self.effective == self.configured
                || self.leases.values().any(|e| e.level == self.effective),
            "effective must equal the configured floor or one active lease level"
        );
    }

    /// Debug-only full invariant sweep for the lease table, called as a
    /// run-to-completion post-condition from every lease-mutating method.
    /// Gathers the cross-field relationships that the individual delta
    /// assertions cannot express on their own.
    #[cfg(debug_assertions)]
    fn check_lease_invariants(&self) {
        debug_assert!(
            self.leases.len() <= LEASE_TABLE_CAP,
            "lease table size must stay within LEASE_TABLE_CAP"
        );
        debug_assert!(
            self.leases
                .keys()
                .all(|id| id.len() <= LEASE_CLIENT_ID_MAX_BYTES),
            "every stored client_id must respect LEASE_CLIENT_ID_MAX_BYTES"
        );
        debug_assert!(
            self.effective >= self.configured,
            "effective is at least the configured floor"
        );
        debug_assert!(
            self.leases.values().all(|e| self.effective >= e.level),
            "effective dominates every active lease level"
        );
        debug_assert!(
            self.effective == self.configured
                || self.leases.values().any(|e| e.level == self.effective),
            "effective equals the configured floor or some active lease level"
        );
    }

    pub fn socket(&self) -> Option<&UdpSocket> {
        self.network.as_ref().map(|n| &n.remote.get_ref().socket)
    }

    pub fn socket_mut(&mut self) -> Option<&mut UdpSocket> {
        self.network
            .as_mut()
            .map(|n| &mut n.remote.get_mut().socket)
    }

    pub fn count_add(&mut self, key: &'static str, count_value: i64) {
        self.receive_metric(key, None, None, MetricValue::Count(count_value));
    }

    pub fn set_gauge(&mut self, key: &'static str, gauge_value: usize) {
        self.receive_metric(key, None, None, MetricValue::Gauge(gauge_value));
    }

    pub fn gauge_add(&mut self, key: &'static str, gauge_value: i64) {
        self.receive_metric(key, None, None, MetricValue::GaugeAdd(gauge_value));
    }

    pub fn writable(&mut self) {
        if let Some(ref mut net) = self.network.as_mut() {
            net.writable();
        }
    }

    pub fn send_data(&mut self) {
        if let Some(ref mut net) = self.network.as_mut() {
            net.send_metrics();
        }
    }

    pub fn dump_local_proxy_metrics(&mut self) -> BTreeMap<String, FilteredMetrics> {
        self.local.dump_proxy_metrics(&Vec::new())
    }

    pub fn query(&mut self, q: &QueryMetricsOptions) -> Result<ResponseContent, MetricError> {
        self.local.query(q)
    }

    pub fn clear_local(&mut self) {
        if let Some(ref mut net) = self.network.as_mut() {
            net.clear();
        }
        self.local.clear();
    }

    pub fn configure(&mut self, config: &MetricsConfiguration) {
        self.local.configure(config);
    }

    /// Drop all metric storage for a cluster across BOTH drains and arm
    /// the per-drain tombstone so subsequent emissions for the cluster are
    /// dropped on the floor instead of resurrecting the row via
    /// `entry().or_default()`. Called from the worker IPC dispatch on
    /// [`RequestType::RemoveCluster`](sozu_command::proto::command::request::RequestType::RemoveCluster). Network-side draining the queued
    /// `MetricLine`s produces immediate silence on the wire (any unsent
    /// statsd interval for the cluster is discarded).
    ///
    /// Worker-event-loop only — `METRICS` is a `thread_local!`, and the
    /// mutable borrow taken here must not be re-entered from a session-
    /// bound code path. Calling this from a metrics emission site would
    /// deadlock the thread-local on `borrow_mut`.
    pub fn remove_cluster(&mut self, cluster_id: &str) {
        self.cluster_incarnations.remove(cluster_id);
        if let Some(ref mut net) = self.network.as_mut() {
            net.remove_cluster(cluster_id);
        }
        self.local.remove_cluster(cluster_id);
    }

    /// Re-arm a previously-removed cluster id across BOTH drains. Called
    /// from the worker IPC dispatch on [`RequestType::AddCluster`](sozu_command::proto::command::request::RequestType::AddCluster).
    /// Idempotent on ids that were never removed. Without this hook a
    /// cluster removed then re-added would stay tombstoned forever and
    /// every fresh metric emission would be dropped.
    pub fn add_cluster(&mut self, cluster_id: &str) {
        if !self.cluster_incarnations.contains_key(cluster_id) {
            let ClusterMetricsCapture::Active(incarnation) =
                self.allocate_cluster_metrics_incarnation(cluster_id)
            else {
                return;
            };
            self.cluster_incarnations
                .insert(cluster_id.to_owned(), incarnation);
        }
        if let Some(ref mut net) = self.network.as_mut() {
            net.add_cluster(cluster_id);
        }
        self.local.add_cluster(cluster_id);
    }

    fn allocate_cluster_metrics_incarnation(&mut self, cluster_id: &str) -> ClusterMetricsCapture {
        let Some(next) = self.last_cluster_incarnation.checked_add(1) else {
            error!(
                "cluster metrics incarnation space exhausted; refusing metrics identity for cluster {:?}",
                cluster_id
            );
            return ClusterMetricsCapture::Invalid;
        };
        let Some(incarnation) = NonZeroU64::new(next).map(ClusterMetricsIncarnation) else {
            unreachable!("checked increment from a live u64 incarnation cannot yield zero");
        };
        self.last_cluster_incarnation = next;
        ClusterMetricsCapture::Active(incarnation)
    }

    /// Capture the active lifetime of `cluster_id`, allocating it when a
    /// legal frontend/backend route references the id before `AddCluster`.
    fn capture_cluster_metrics(&mut self, cluster_id: &str) -> ClusterMetricsCapture {
        if let Some(incarnation) = self.cluster_incarnations.get(cluster_id).copied() {
            return ClusterMetricsCapture::Active(incarnation);
        }
        let capture = self.allocate_cluster_metrics_incarnation(cluster_id);
        if let ClusterMetricsCapture::Active(incarnation) = capture {
            self.cluster_incarnations
                .insert(cluster_id.to_owned(), incarnation);
            // This implicit route starts the replacement lifetime just as an
            // AddCluster would. Re-arm both drains so the new identity can
            // create its row while the identity gate still rejects owners
            // captured before RemoveCluster.
            if let Some(ref mut net) = self.network.as_mut() {
                net.add_cluster(cluster_id);
            }
            self.local.add_cluster(cluster_id);
        }
        capture
    }

    /// Return the active lifetime of `cluster_id`.
    pub(crate) fn cluster_incarnation(
        &self,
        cluster_id: &str,
    ) -> Option<ClusterMetricsIncarnation> {
        self.cluster_incarnations.get(cluster_id).copied()
    }

    /// Whether an emission for `cluster_id` captured with `capture`
    /// may reach the drains under the current effective detail.
    ///
    /// Process and Frontend detail discard cluster/backend labels, so every
    /// capture state contributes to that aggregate: a late decrement must
    /// balance the earlier increment. Cluster and Backend detail retain the
    /// label and therefore require a real captured identity equal to the
    /// current active identity. Missing and exhausted captures fail closed.
    pub(crate) fn accepts_cluster_capture(
        &self,
        cluster_id: &str,
        capture: ClusterMetricsCapture,
    ) -> bool {
        if !matches!(
            self.effective,
            MetricDetailLevel::Cluster | MetricDetailLevel::Backend
        ) {
            return true;
        }
        matches!(
            capture,
            ClusterMetricsCapture::Active(incarnation)
                if self.cluster_incarnation(cluster_id) == Some(incarnation)
        )
    }

    /// Receive a metric from an owner that captured a cluster incarnation.
    ///
    /// The identity check intentionally precedes cardinality filtering. A
    /// stale event is rejected when the effective detail retains its cluster
    /// label, protecting a same-id replacement row. At `Process` and
    /// `Frontend` detail the label is deliberately removed to form a worker
    /// aggregate, so the event remains part of that aggregate: while detail
    /// stays at that level, an old connection's late `-1` still balances its
    /// earlier `+1`. Existing detail transitions do not migrate stored gauge
    /// contributions between shapes. See
    /// [`Self::accepts_cluster_capture`] for the identity comparison.
    pub(crate) fn receive_metric_for_incarnation(
        &mut self,
        label: &'static str,
        cluster_id: Option<&str>,
        backend_id: Option<&str>,
        capture: ClusterMetricsCapture,
        metric: MetricValue,
    ) {
        if cluster_id.is_none_or(|cluster_id| self.accepts_cluster_capture(cluster_id, capture)) {
            self.receive_metric(label, cluster_id, backend_id, metric);
        }
    }

    /// Drop all metric storage for one backend across BOTH drains. Called
    /// from the worker IPC dispatch on [`RequestType::RemoveBackend`](sozu_command::proto::command::request::RequestType::RemoveBackend).
    /// Does NOT tombstone the cluster (only `remove_cluster` does).
    pub fn remove_backend(&mut self, cluster_id: &str, backend_id: &str) {
        if let Some(ref mut net) = self.network.as_mut() {
            net.remove_backend(cluster_id, backend_id);
        }
        self.local.remove_backend(cluster_id, backend_id);
    }
}

impl Subscriber for Aggregator {
    fn receive_metric(
        &mut self,
        label: &'static str,
        cluster_id: Option<&str>,
        backend_id: Option<&str>,
        metric: MetricValue,
    ) {
        // Apply the cardinality knob BEFORE handing the metric to either
        // drain. Both drains see the same filtered labels, keeping the local
        // CLI view consistent with what statsd receives on the wire. Reads
        // `effective` (max of the configured floor and any active leases),
        // which is recomputed off the hot path.
        let (cluster_id, backend_id) =
            filter_labels_for_detail(self.effective, cluster_id, backend_id);
        if let Some(ref mut net) = self.network.as_mut() {
            net.receive_metric(label, cluster_id, backend_id, metric.to_owned());
        }
        self.local
            .receive_metric(label, cluster_id, backend_id, metric);
    }
}

pub struct MetricSocket {
    pub addr: SocketAddr,
    pub socket: UdpSocket,
}

impl Write for MetricSocket {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.socket.send_to(buf, self.addr)
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

pub fn udp_bind() -> Result<UdpSocket, MetricError> {
    let address = "0.0.0.0:0";

    let udp_address =
        address
            .parse::<SocketAddr>()
            .map_err(|parse_error| MetricError::WrongUdpAddress {
                address: address.to_owned(),
                error: parse_error.to_string(),
            })?;

    UdpSocket::bind(udp_address).map_err(|parse_error| MetricError::UdpBind {
        address: udp_address.to_string(),
        error: parse_error.to_string(),
    })
}

/// adds a value to a counter
#[macro_export]
macro_rules! count (
  ($key:expr, $value: expr) => ({
    let v = $value;
    $crate::metrics::METRICS.with(|metrics| {
      (*metrics.borrow_mut()).count_add($key, v);
    });
  })
);

/// adds 1 to a counter
#[macro_export]
macro_rules! incr (
  ($key:expr) => (count!($key, 1));
  ($key:expr, $cluster_id:expr, $backend_id:expr) => {
    {
        use $crate::metrics::Subscriber;

        $crate::metrics::METRICS.with(|metrics| {
          (*metrics.borrow_mut()).receive_metric($key, $cluster_id, $backend_id, $crate::metrics::MetricValue::Count(1));
        });
    }
  };
  ($key:expr, $cluster_id:expr, $backend_id:expr, $incarnation:expr) => {
    {
        $crate::metrics::METRICS.with(|metrics| {
          (*metrics.borrow_mut()).receive_metric_for_incarnation(
            $key,
            $cluster_id,
            $backend_id,
            $incarnation,
            $crate::metrics::MetricValue::Count(1),
          );
        });
    }
  }
);

#[macro_export]
macro_rules! decr (
  ($key:expr) => (count!($key, -1))
);

#[macro_export]
macro_rules! gauge (
  ($key:expr, $value: expr) => ({
    let v = $value;
    $crate::metrics::METRICS.with(|metrics| {
      (*metrics.borrow_mut()).set_gauge($key, v);
    });
  });
  ($key:expr, $value:expr, $cluster_id:expr, $backend_id:expr) => {
    {
        use $crate::metrics::Subscriber;
        let v = $value;

        $crate::metrics::METRICS.with(|metrics| {
          (*metrics.borrow_mut()).receive_metric($key, $cluster_id, $backend_id, $crate::metrics::MetricValue::Gauge(v as usize));
        });
    }
  };
  ($key:expr, $value:expr, $cluster_id:expr, $backend_id:expr, $incarnation:expr) => {
    {
        let v = $value;
        $crate::metrics::METRICS.with(|metrics| {
          (*metrics.borrow_mut()).receive_metric_for_incarnation(
            $key,
            $cluster_id,
            $backend_id,
            $incarnation,
            $crate::metrics::MetricValue::Gauge(v as usize),
          );
        });
    }
  }
);

#[macro_export]
macro_rules! gauge_add (
  ($key:expr, $value: expr) => ({
    let v = $value;
    $crate::metrics::METRICS.with(|metrics| {
      (*metrics.borrow_mut()).gauge_add($key, v);
    });
  });
  ($key:expr, $value:expr, $cluster_id:expr, $backend_id:expr) => {
    {
        use $crate::metrics::Subscriber;
        let v = $value;

        $crate::metrics::METRICS.with(|metrics| {
          (*metrics.borrow_mut()).receive_metric($key, $cluster_id, $backend_id, $crate::metrics::MetricValue::GaugeAdd(v));
        });
    }
  };
  ($key:expr, $value:expr, $cluster_id:expr, $backend_id:expr, $incarnation:expr) => {
    {
        let v = $value;
        $crate::metrics::METRICS.with(|metrics| {
          (*metrics.borrow_mut()).receive_metric_for_incarnation(
            $key,
            $cluster_id,
            $backend_id,
            $incarnation,
            $crate::metrics::MetricValue::GaugeAdd(v),
          );
        });
    }
  }
);

#[macro_export]
macro_rules! time (
  ($key:expr, $value: expr) => ({
    use $crate::metrics::{MetricValue,Subscriber};
    let v = $value;
    $crate::metrics::METRICS.with(|metrics| {
      let m = &mut *metrics.borrow_mut();

      m.receive_metric($key, None, None, MetricValue::Time(v as usize));
    });
  });
  ($key:expr, $cluster_id:expr, $value: expr) => ({
    use $crate::metrics::{MetricValue,Subscriber};
    let v = $value;
    $crate::metrics::METRICS.with(|metrics| {
      let m = &mut *metrics.borrow_mut();
      let cluster: &str = $cluster_id;

      m.receive_metric($key, Some(cluster), None, MetricValue::Time(v as usize));
    });
  });
  ($key:expr, $cluster_id:expr, $value:expr, $incarnation:expr) => ({
    use $crate::metrics::MetricValue;
    let v = $value;
    $crate::metrics::METRICS.with(|metrics| {
      let m = &mut *metrics.borrow_mut();
      let cluster: &str = $cluster_id;

      m.receive_metric_for_incarnation(
        $key,
        Some(cluster),
        None,
        $incarnation,
        MetricValue::Time(v as usize),
      );
    });
  })
);

#[macro_export]
macro_rules! record_backend_metrics (
  ($cluster_id:expr, $backend_id:expr, $response_time: expr, $backend_connection_time: expr, $backend_header_time: expr, $bin: expr, $bout: expr) => {
    use $crate::metrics::{MetricValue,Subscriber};
    $crate::metrics::METRICS.with(|metrics| {
      let m = &mut *metrics.borrow_mut();
      let cluster_id: &str = $cluster_id;
      let backend_id: &str = $backend_id;

      m.receive_metric($crate::metrics::names::backend::BYTES_IN, Some(cluster_id), Some(backend_id), MetricValue::Count($bin as i64));
      m.receive_metric($crate::metrics::names::backend::BYTES_OUT, Some(cluster_id), Some(backend_id), MetricValue::Count($bout as i64));
      m.receive_metric($crate::metrics::names::backend::RESPONSE_TIME, Some(cluster_id), Some(backend_id), MetricValue::Time($response_time as usize));
      if let Some(t) = $backend_connection_time {
        m.receive_metric($crate::metrics::names::backend::CONNECTION_TIME, Some(cluster_id), Some(backend_id), MetricValue::Time(t.as_millis() as usize));
      }
      // `Option`, like the connection time and unlike the response time: a
      // protocol with no response headers at all (raw TCP) must record
      // nothing here rather than a zero.
      if let Some(t) = $backend_header_time {
        m.receive_metric($crate::metrics::names::backend::HEADER_TIME, Some(cluster_id), Some(backend_id), MetricValue::Time(t.as_millis() as usize));
      }

      m.receive_metric($crate::metrics::names::backend::REQUESTS, Some(cluster_id), Some(backend_id), MetricValue::Count(1));
    });
  };
  ($cluster_id:expr, $backend_id:expr, $response_time:expr, $backend_connection_time:expr, $backend_header_time:expr, $bin:expr, $bout:expr, $incarnation:expr) => {
    use $crate::metrics::{MetricValue,Subscriber};
    $crate::metrics::METRICS.with(|metrics| {
      let m = &mut *metrics.borrow_mut();
      let cluster_id: &str = $cluster_id;
      let backend_id: &str = $backend_id;

      // One identity lookup gates the whole burst: every emission below
      // shares the same cluster and captured incarnation.
      if m.accepts_cluster_capture(cluster_id, $incarnation) {
        m.receive_metric($crate::metrics::names::backend::BYTES_IN, Some(cluster_id), Some(backend_id), MetricValue::Count($bin as i64));
        m.receive_metric($crate::metrics::names::backend::BYTES_OUT, Some(cluster_id), Some(backend_id), MetricValue::Count($bout as i64));
        m.receive_metric($crate::metrics::names::backend::RESPONSE_TIME, Some(cluster_id), Some(backend_id), MetricValue::Time($response_time as usize));
        if let Some(t) = $backend_connection_time {
          m.receive_metric($crate::metrics::names::backend::CONNECTION_TIME, Some(cluster_id), Some(backend_id), MetricValue::Time(t.as_millis() as usize));
        }
        if let Some(t) = $backend_header_time {
          m.receive_metric($crate::metrics::names::backend::HEADER_TIME, Some(cluster_id), Some(backend_id), MetricValue::Time(t.as_millis() as usize));
        }

        m.receive_metric($crate::metrics::names::backend::REQUESTS, Some(cluster_id), Some(backend_id), MetricValue::Count(1));
      }
    });
  }
);

#[cfg(test)]
mod tests {
    use super::*;
    use sozu_command::proto::command::filtered_metrics;

    #[test]
    fn filter_labels_process_drops_both() {
        assert_eq!(
            filter_labels_for_detail(MetricDetailLevel::Process, Some("c"), Some("b")),
            (None, None),
        );
    }

    #[test]
    fn filter_labels_frontend_drops_both_today() {
        // Reserved for per-listener counters; same as Process for now.
        assert_eq!(
            filter_labels_for_detail(MetricDetailLevel::Frontend, Some("c"), Some("b")),
            (None, None),
        );
    }

    #[test]
    fn filter_labels_cluster_keeps_cluster_drops_backend() {
        assert_eq!(
            filter_labels_for_detail(MetricDetailLevel::Cluster, Some("c"), Some("b")),
            (Some("c"), None),
        );
    }

    #[test]
    fn cluster_incarnation_fences_labelled_rows_and_preserves_process_aggregates() {
        let mut aggregator = Aggregator::new(String::new());
        aggregator.set_up_detail(MetricDetailLevel::Backend);
        aggregator.add_cluster("reused");
        let old_token = aggregator
            .cluster_incarnation("reused")
            .expect("configured cluster has an identity");
        let old = ClusterMetricsCapture::Active(old_token);

        // Neither a live AddCluster update nor local collection controls end
        // the configured cluster lifetime.
        aggregator.add_cluster("reused");
        assert_eq!(aggregator.cluster_incarnation("reused"), Some(old_token));
        aggregator.configure(&MetricsConfiguration::Disabled);
        aggregator.configure(&MetricsConfiguration::Enabled);
        aggregator.configure(&MetricsConfiguration::Clear);
        assert_eq!(aggregator.cluster_incarnation("reused"), Some(old_token));

        aggregator.remove_cluster("reused");
        aggregator.add_cluster("reused");
        let replacement_token = aggregator
            .cluster_incarnation("reused")
            .expect("replacement cluster has an identity");
        let replacement = ClusterMetricsCapture::Active(replacement_token);
        assert_ne!(replacement_token, old_token);

        // At Backend detail an obsolete event is rejected instead of being
        // applied to the replacement row.
        aggregator.receive_metric_for_incarnation(
            "incarnation_test",
            Some("reused"),
            Some("same-backend"),
            old,
            MetricValue::Count(7),
        );
        assert!(
            !aggregator
                .local
                .dump_cluster_metrics(&[])
                .expect("dump cluster metrics after stale emission")
                .contains_key("reused"),
            "a labelled stale event must not recreate the replacement row",
        );

        // The same gate must still accept an event from the active lifetime.
        aggregator.receive_metric_for_incarnation(
            "incarnation_test",
            Some("reused"),
            Some("same-backend"),
            replacement,
            MetricValue::Count(3),
        );
        let labelled = aggregator
            .local
            .dump_cluster_metrics(&[])
            .expect("dump active cluster metrics");
        let active_backend = labelled["reused"]
            .backends
            .iter()
            .find(|backend| backend.backend_id == "same-backend")
            .expect("active backend metric row");
        assert_eq!(
            active_backend
                .metrics
                .get("incarnation_test")
                .and_then(|metric| metric.inner.as_ref()),
            Some(&filtered_metrics::Inner::Count(3)),
        );

        // Process detail deliberately aggregates cluster/backend events by
        // removing their labels. Both old and current work contribute while
        // the effective detail remains at that level.
        aggregator.set_up_detail(MetricDetailLevel::Process);
        aggregator.receive_metric_for_incarnation(
            "incarnation_test",
            Some("reused"),
            Some("same-backend"),
            old,
            MetricValue::Count(7),
        );
        aggregator.receive_metric_for_incarnation(
            "incarnation_test",
            Some("reused"),
            Some("same-backend"),
            replacement,
            MetricValue::Count(3),
        );
        assert_eq!(
            aggregator
                .dump_local_proxy_metrics()
                .get("incarnation_test")
                .and_then(|metric| metric.inner.as_ref()),
            Some(&filtered_metrics::Inner::Count(10)),
        );

        // A true proxy metric has no cluster identity and is independent of
        // the incarnation gate at every detail level.
        aggregator.receive_metric("direct_proxy_test", None, None, MetricValue::Count(5));
        assert_eq!(
            aggregator
                .dump_local_proxy_metrics()
                .get("direct_proxy_test")
                .and_then(|metric| metric.inner.as_ref()),
            Some(&filtered_metrics::Inner::Count(5)),
        );

        // Exhaustion never wraps to a previously issued identity and never
        // re-arms the drain tombstone.
        let mut exhausted = Aggregator::new(String::new());
        exhausted.last_cluster_incarnation = u64::MAX;
        exhausted.add_cluster("exhausted");
        assert_eq!(exhausted.cluster_incarnation("exhausted"), None);
    }

    /// Value of `key` in the local drain for `cluster_id`: the backend row
    /// when `backend_id` is given, the cluster row otherwise.
    fn local_labelled_value(
        aggregator: &mut Aggregator,
        cluster_id: &str,
        backend_id: Option<&str>,
        key: &str,
    ) -> Option<filtered_metrics::Inner> {
        let dump = aggregator
            .local
            .dump_cluster_metrics(&[])
            .expect("dump cluster metrics");
        let row = dump.get(cluster_id)?;
        let metrics = match backend_id {
            Some(backend_id) => {
                &row.backends
                    .iter()
                    .find(|backend| backend.backend_id == backend_id)?
                    .metrics
            }
            None => &row.cluster,
        };
        metrics.get(key)?.inner.clone()
    }

    #[test]
    fn add_without_remove_keeps_implicit_identity_and_balances_labelled_gauge() {
        let mut aggregator = Aggregator::new(String::new());
        aggregator.set_up_detail(MetricDetailLevel::Backend);
        let capture = aggregator.capture_cluster_metrics("implicit-then-added");
        let ClusterMetricsCapture::Active(token) = capture else {
            panic!("implicit route must allocate an active identity");
        };
        aggregator.receive_metric_for_incarnation(
            "implicit_balance",
            Some("implicit-then-added"),
            Some("backend"),
            capture,
            MetricValue::GaugeAdd(1),
        );

        aggregator.add_cluster("implicit-then-added");
        assert_eq!(
            aggregator.cluster_incarnation("implicit-then-added"),
            Some(token),
        );
        aggregator.receive_metric_for_incarnation(
            "implicit_balance",
            Some("implicit-then-added"),
            Some("backend"),
            capture,
            MetricValue::GaugeAdd(-1),
        );

        assert_eq!(
            local_labelled_value(
                &mut aggregator,
                "implicit-then-added",
                Some("backend"),
                "implicit_balance",
            ),
            Some(filtered_metrics::Inner::Gauge(0)),
        );
    }

    #[test]
    fn implicit_cluster_capture_is_recorded_at_process_detail() {
        // A frontend may route to a cluster that never received AddCluster;
        // its owners capture a valid implicit identity. The label-stripped
        // worker aggregate must still count those events.
        let mut aggregator = Aggregator::new(String::new());
        aggregator.set_up_detail(MetricDetailLevel::Process);
        let capture = aggregator.capture_cluster_metrics("undeclared");
        aggregator.receive_metric_for_incarnation(
            "undeclared_test",
            Some("undeclared"),
            Some("backend"),
            capture,
            MetricValue::Count(4),
        );
        assert_eq!(
            aggregator
                .dump_local_proxy_metrics()
                .get("undeclared_test")
                .and_then(|metric| metric.inner.as_ref()),
            Some(&filtered_metrics::Inner::Count(4)),
        );
    }

    #[test]
    fn implicit_cluster_capture_is_recorded_at_labelled_detail() {
        for detail in [MetricDetailLevel::Cluster, MetricDetailLevel::Backend] {
            let mut aggregator = Aggregator::new(String::new());
            aggregator.set_up_detail(detail);
            let capture = aggregator.capture_cluster_metrics("undeclared");
            assert!(matches!(capture, ClusterMetricsCapture::Active(_)));
            aggregator.receive_metric_for_incarnation(
                "undeclared_test",
                Some("undeclared"),
                Some("backend"),
                capture,
                MetricValue::Count(4),
            );
            let backend_id = match detail {
                MetricDetailLevel::Backend => Some("backend"),
                _ => None,
            };
            let recorded =
                local_labelled_value(&mut aggregator, "undeclared", backend_id, "undeclared_test");
            assert_eq!(
                recorded,
                Some(filtered_metrics::Inner::Count(4)),
                "the active implicit identity must pass the gate at {detail:?}",
            );
        }
    }

    #[test]
    fn mismatched_incarnation_is_dropped_at_labelled_detail() {
        for detail in [MetricDetailLevel::Cluster, MetricDetailLevel::Backend] {
            let mut aggregator = Aggregator::new(String::new());
            aggregator.set_up_detail(detail);

            // Captured Some(I1), current Some(I2).
            aggregator.add_cluster("c");
            let first = aggregator
                .cluster_incarnation("c")
                .expect("first configured identity");
            aggregator.remove_cluster("c");
            aggregator.add_cluster("c");
            let second = aggregator
                .cluster_incarnation("c")
                .expect("replacement configured identity");
            assert_ne!(first, second);
            aggregator.receive_metric_for_incarnation(
                "mismatch_test",
                Some("c"),
                Some("b"),
                ClusterMetricsCapture::Active(first),
                MetricValue::Count(1),
            );

            // A missing capture never adopts the current lifetime.
            aggregator.receive_metric_for_incarnation(
                "mismatch_test",
                Some("c"),
                Some("b"),
                ClusterMetricsCapture::Missing,
                MetricValue::Count(1),
            );
            assert!(
                !aggregator
                    .local
                    .dump_cluster_metrics(&[])
                    .expect("dump cluster metrics")
                    .contains_key("c"),
                "stale or missing capture must not reach the Some(I2) row at {detail:?}",
            );

            // Captured I1, current identity absent under this id.
            aggregator.receive_metric_for_incarnation(
                "mismatch_test",
                Some("other"),
                Some("b"),
                ClusterMetricsCapture::Active(first),
                MetricValue::Count(1),
            );
            assert!(
                !aggregator
                    .local
                    .dump_cluster_metrics(&[])
                    .expect("dump cluster metrics")
                    .contains_key("other"),
                "Some(I1) must not reach an undeclared (None) cluster at {detail:?}",
            );
        }
    }

    #[test]
    fn removed_implicit_capture_stays_rejected_after_clear() {
        let mut aggregator = Aggregator::new(String::new());
        aggregator.set_up_detail(MetricDetailLevel::Backend);

        // A legal route allocates an implicit identity before this cluster
        // has ever received AddCluster.
        let captured_before_add = aggregator.capture_cluster_metrics("reviewer-reused");
        let ClusterMetricsCapture::Active(implicit_token) = captured_before_add else {
            panic!("a legal implicit route must capture an active identity");
        };

        aggregator.add_cluster("reviewer-reused");
        assert_eq!(
            aggregator.cluster_incarnation("reviewer-reused"),
            Some(implicit_token),
            "AddCluster without RemoveCluster keeps the implicit lifetime",
        );
        aggregator.remove_cluster("reviewer-reused");

        // Clear is a supported metrics operation. It clears drain storage,
        // including tombstones, but it must not make a removed lifetime's
        // earlier identity become current again.
        aggregator.configure(&MetricsConfiguration::Clear);
        aggregator.receive_metric_for_incarnation(
            "reviewer_late_close",
            Some("reviewer-reused"),
            Some("backend"),
            captured_before_add,
            MetricValue::Count(1),
        );

        assert!(
            !aggregator
                .local
                .dump_cluster_metrics(&[])
                .expect("dump metrics after late removed emission")
                .contains_key("reviewer-reused"),
            "a removed lifetime must stay fenced after Clear removes the drain tombstone",
        );

        let replacement = aggregator.capture_cluster_metrics("reviewer-reused");
        let ClusterMetricsCapture::Active(replacement_token) = replacement else {
            panic!("a route after removal must allocate a replacement identity");
        };
        assert_ne!(replacement_token, implicit_token);
        aggregator.add_cluster("reviewer-reused");
        assert_eq!(
            aggregator.cluster_incarnation("reviewer-reused"),
            Some(replacement_token),
        );
        aggregator.receive_metric_for_incarnation(
            "reviewer_replacement",
            Some("reviewer-reused"),
            Some("backend"),
            captured_before_add,
            MetricValue::Count(1),
        );
        aggregator.receive_metric_for_incarnation(
            "reviewer_replacement",
            Some("reviewer-reused"),
            Some("backend"),
            replacement,
            MetricValue::Count(2),
        );
        assert_eq!(
            local_labelled_value(
                &mut aggregator,
                "reviewer-reused",
                Some("backend"),
                "reviewer_replacement",
            ),
            Some(filtered_metrics::Inner::Count(2)),
            "only the post-remove identity may reach the replacement row",
        );
    }

    #[test]
    fn implicit_route_after_remove_rearms_drains_without_add_cluster() {
        let mut aggregator = Aggregator::new(String::new());
        aggregator.set_up_detail(MetricDetailLevel::Backend);

        let first = aggregator.capture_cluster_metrics("implicit-replacement");
        aggregator.receive_metric_for_incarnation(
            "implicit_replacement",
            Some("implicit-replacement"),
            Some("backend"),
            first,
            MetricValue::Count(1),
        );
        aggregator.remove_cluster("implicit-replacement");

        // Frontends and backends may legally recreate this id without an
        // AddCluster. The first new route is therefore both the identity and
        // drain-lifetime boundary for the replacement.
        let replacement = aggregator.capture_cluster_metrics("implicit-replacement");
        assert_ne!(replacement, first);
        aggregator.receive_metric_for_incarnation(
            "implicit_replacement",
            Some("implicit-replacement"),
            Some("backend"),
            replacement,
            MetricValue::Count(2),
        );
        aggregator.receive_metric_for_incarnation(
            "implicit_replacement",
            Some("implicit-replacement"),
            Some("backend"),
            first,
            MetricValue::Count(4),
        );

        assert_eq!(
            local_labelled_value(
                &mut aggregator,
                "implicit-replacement",
                Some("backend"),
                "implicit_replacement",
            ),
            Some(filtered_metrics::Inner::Count(2)),
            "the replacement implicit lifetime must be visible while the removed lifetime stays fenced",
        );
    }

    #[test]
    fn exhausted_add_is_not_an_implicit_identity() {
        let mut aggregator = Aggregator::new(String::new());
        aggregator.set_up_detail(MetricDetailLevel::Backend);
        aggregator.last_cluster_incarnation = u64::MAX;
        aggregator.add_cluster("reviewer-exhausted");
        assert_eq!(aggregator.cluster_incarnation("reviewer-exhausted"), None);

        aggregator.receive_metric_for_incarnation(
            "reviewer_exhausted",
            Some("reviewer-exhausted"),
            Some("backend"),
            ClusterMetricsCapture::Invalid,
            MetricValue::Count(1),
        );

        assert!(
            !aggregator
                .local
                .dump_cluster_metrics(&[])
                .expect("dump metrics after exhausted emission")
                .contains_key("reviewer-exhausted"),
            "allocation exhaustion must fail closed instead of sharing the undeclared identity",
        );
    }

    #[test]
    fn exhausted_implicit_capture_does_not_rearm_removed_drains() {
        let mut aggregator = Aggregator::new(String::new());
        aggregator.set_up_detail(MetricDetailLevel::Backend);
        assert!(matches!(
            aggregator.capture_cluster_metrics("exhausted-after-remove"),
            ClusterMetricsCapture::Active(_),
        ));
        aggregator.remove_cluster("exhausted-after-remove");
        aggregator.last_cluster_incarnation = u64::MAX;

        let exhausted = aggregator.capture_cluster_metrics("exhausted-after-remove");
        assert_eq!(exhausted, ClusterMetricsCapture::Invalid);
        aggregator.receive_metric_for_incarnation(
            "exhausted_after_remove",
            Some("exhausted-after-remove"),
            Some("backend"),
            exhausted,
            MetricValue::Count(1),
        );
        // Probe the drain directly as well: the identity-aware call above is
        // rejected before dispatch and therefore cannot prove that allocation
        // exhaustion left the RemoveCluster tombstone armed.
        aggregator.receive_metric(
            "exhausted_after_remove_probe",
            Some("exhausted-after-remove"),
            Some("backend"),
            MetricValue::Count(1),
        );

        assert!(
            !aggregator
                .local
                .dump_cluster_metrics(&[])
                .expect("dump metrics after exhausted implicit capture")
                .contains_key("exhausted-after-remove"),
            "allocation exhaustion must leave the removed drain tombstone armed",
        );
    }

    #[test]
    fn every_capture_state_keeps_process_and_frontend_aggregates() {
        for detail in [MetricDetailLevel::Process, MetricDetailLevel::Frontend] {
            let mut aggregator = Aggregator::new(String::new());
            aggregator.set_up_detail(detail);
            let active = aggregator.capture_cluster_metrics("reviewer-undeclared");
            for capture in [
                active,
                ClusterMetricsCapture::Missing,
                ClusterMetricsCapture::Invalid,
            ] {
                aggregator.receive_metric_for_incarnation(
                    "reviewer_unconfigured_aggregate",
                    Some("reviewer-undeclared"),
                    Some("backend"),
                    capture,
                    MetricValue::Count(1),
                );
            }
            assert_eq!(
                aggregator
                    .dump_local_proxy_metrics()
                    .get("reviewer_unconfigured_aggregate")
                    .and_then(|metric| metric.inner.as_ref()),
                Some(&filtered_metrics::Inner::Count(3)),
                "identity state cannot suppress a label-free {detail:?} aggregate",
            );
        }
    }

    #[test]
    fn filter_labels_backend_keeps_both() {
        assert_eq!(
            filter_labels_for_detail(MetricDetailLevel::Backend, Some("c"), Some("b")),
            (Some("c"), Some("b")),
        );
    }

    #[test]
    fn filter_labels_none_in_none_out() {
        // Absent labels stay absent regardless of detail level — the filter
        // never invents a label, only drops.
        for detail in [
            MetricDetailLevel::Process,
            MetricDetailLevel::Frontend,
            MetricDetailLevel::Cluster,
            MetricDetailLevel::Backend,
        ] {
            assert_eq!(filter_labels_for_detail(detail, None, None), (None, None));
        }
    }

    #[test]
    fn aggregator_default_detail_is_cluster() {
        // Pre-knob behaviour preserved: if a worker / process never calls
        // `set_up_detail`, cluster-scoped metrics still reach the drains.
        let agg = Aggregator::new("sozu".to_owned());
        assert_eq!(agg.detail_configured(), MetricDetailLevel::Cluster);
        assert_eq!(agg.detail_effective(), MetricDetailLevel::Cluster);
        assert_eq!(agg.lease_count(), 0);
    }

    /// Fully-known binding used by tests that don't otherwise care about the
    /// peer-binding mechanic. Two distinct values (`OWNER_*` / `OTHER_*`)
    /// let `lease_clear` tests assert authorised vs unauthorised paths.
    fn owner_binding() -> PeerBinding {
        PeerBinding {
            pid: Some(1234),
            session_ulid: Some(0x0123_4567_89ab_cdef_0123_4567_89ab_cdefu128),
        }
    }

    fn other_binding() -> PeerBinding {
        PeerBinding {
            pid: Some(5678),
            session_ulid: Some(0xfedc_ba98_7654_3210_fedc_ba98_7654_3210u128),
        }
    }

    /// Extract `(previous_effective, new_effective)` from a successful
    /// `lease_apply`; panics on any failure arm so tests that don't care
    /// about the failure paths stay compact.
    fn unwrap_applied(outcome: LeaseApplyOutcome) -> (MetricDetailLevel, MetricDetailLevel) {
        match outcome {
            LeaseApplyOutcome::Applied {
                previous_effective,
                new_effective,
            } => (previous_effective, new_effective),
            other => panic!("expected LeaseApplyOutcome::Applied, got {other:?}"),
        }
    }

    #[test]
    fn lease_apply_elevates_effective_above_configured() {
        // Configured floor stays at Cluster; a lease for Backend lifts the
        // effective level until the lease expires or is cleared.
        let mut agg = Aggregator::new("sozu".to_owned());
        agg.set_up_detail(MetricDetailLevel::Cluster);
        let (prev, new) = unwrap_applied(agg.lease_apply(
            "test:1".to_owned(),
            MetricDetailLevel::Backend,
            Duration::from_secs(60),
            PeerBinding::default(),
            Instant::now(),
        ));
        assert_eq!(prev, MetricDetailLevel::Cluster);
        assert_eq!(new, MetricDetailLevel::Backend);
        assert_eq!(agg.detail_configured(), MetricDetailLevel::Cluster);
        assert_eq!(agg.detail_effective(), MetricDetailLevel::Backend);
        assert_eq!(agg.lease_count(), 1);
    }

    #[test]
    fn lease_apply_below_configured_does_not_lower_effective() {
        // A lease can only ELEVATE the floor, never push below `configured`.
        let mut agg = Aggregator::new("sozu".to_owned());
        agg.set_up_detail(MetricDetailLevel::Backend);
        let (prev, new) = unwrap_applied(agg.lease_apply(
            "test:1".to_owned(),
            MetricDetailLevel::Cluster,
            Duration::from_secs(60),
            PeerBinding::default(),
            Instant::now(),
        ));
        assert_eq!(prev, MetricDetailLevel::Backend);
        assert_eq!(new, MetricDetailLevel::Backend);
    }

    #[test]
    fn lease_apply_rejects_client_id_over_cap() {
        // Defence-in-depth: even if dispatch lets a too-long id through,
        // the aggregator refuses to store it.
        let mut agg = Aggregator::new("sozu".to_owned());
        let too_long = "x".repeat(LEASE_CLIENT_ID_MAX_BYTES + 1);
        assert_eq!(
            agg.lease_apply(
                too_long,
                MetricDetailLevel::Backend,
                Duration::from_secs(60),
                PeerBinding::default(),
                Instant::now(),
            ),
            LeaseApplyOutcome::ClientIdTooLong
        );
        assert_eq!(agg.lease_count(), 0);
    }

    #[test]
    fn lease_apply_rejects_when_table_is_full() {
        // Fill the table to capacity with distinct client_ids; one more
        // insert is refused. A RENEWAL of an existing entry must still
        // succeed (replaces in place, count unchanged).
        let mut agg = Aggregator::new("sozu".to_owned());
        let now = Instant::now();
        for i in 0..LEASE_TABLE_CAP {
            assert!(matches!(
                agg.lease_apply(
                    format!("client:{i:02}"),
                    MetricDetailLevel::Backend,
                    Duration::from_secs(60),
                    PeerBinding::default(),
                    now,
                ),
                LeaseApplyOutcome::Applied { .. }
            ));
        }
        assert_eq!(agg.lease_count() as usize, LEASE_TABLE_CAP);
        // New distinct client → rejected.
        assert_eq!(
            agg.lease_apply(
                "newcomer".to_owned(),
                MetricDetailLevel::Backend,
                Duration::from_secs(60),
                PeerBinding::default(),
                now,
            ),
            LeaseApplyOutcome::TableFull,
        );
        assert_eq!(agg.lease_count() as usize, LEASE_TABLE_CAP);
        // Renewal of an existing entry → still accepted.
        assert!(matches!(
            agg.lease_apply(
                "client:00".to_owned(),
                MetricDetailLevel::Backend,
                Duration::from_secs(120),
                PeerBinding::default(),
                now,
            ),
            LeaseApplyOutcome::Applied { .. }
        ));
        assert_eq!(agg.lease_count() as usize, LEASE_TABLE_CAP);
    }

    #[test]
    fn lease_apply_rejects_ttl_over_max() {
        // The aggregator no longer silently clamps oversize TTLs.
        let mut agg = Aggregator::new("sozu".to_owned());
        assert_eq!(
            agg.lease_apply(
                "client:0".to_owned(),
                MetricDetailLevel::Backend,
                LEASE_TTL_MAX + Duration::from_secs(1),
                PeerBinding::default(),
                Instant::now(),
            ),
            LeaseApplyOutcome::TtlOutOfRange,
        );
        assert_eq!(agg.lease_count(), 0);
    }

    #[test]
    fn lease_apply_renewal_replaces_previous_for_same_client() {
        // The renewer client re-sends with the same `client_id`; the entry
        // is REPLACED (not duplicated). Lease count stays at 1.
        // Unknown bindings on both sides skip the renewal-binding gate.
        let mut agg = Aggregator::new("sozu".to_owned());
        let now = Instant::now();
        let _ = agg.lease_apply(
            "renewer".to_owned(),
            MetricDetailLevel::Backend,
            Duration::from_secs(30),
            PeerBinding::default(),
            now,
        );
        let _ = agg.lease_apply(
            "renewer".to_owned(),
            MetricDetailLevel::Backend,
            Duration::from_secs(60),
            PeerBinding::default(),
            now,
        );
        assert_eq!(agg.lease_count(), 1);
    }

    #[test]
    fn lease_apply_renewal_at_lower_level_lowers_effective() {
        // Issue #1408: a client holding the current maximum renews its own
        // lease at a LOWER level. A lease only ever elevates cardinality on
        // the client's own behalf, so a client giving part of that back is
        // withdrawing its own request, not violating a shared invariant.
        // `effective` must recompute to max(configured, max over live
        // leases) exactly like any other apply -- dropping here is the
        // correct outcome, not a bug.
        let mut agg = Aggregator::new("sozu".to_owned());
        agg.set_up_detail(MetricDetailLevel::Process);
        let now = Instant::now();
        let (_, new) = unwrap_applied(agg.lease_apply(
            "renewer".to_owned(),
            MetricDetailLevel::Backend,
            Duration::from_secs(60),
            PeerBinding::default(),
            now,
        ));
        assert_eq!(new, MetricDetailLevel::Backend);
        assert_eq!(agg.detail_effective(), MetricDetailLevel::Backend);
        // Same client_id renews at a strictly lower level.
        let (prev, new) = unwrap_applied(agg.lease_apply(
            "renewer".to_owned(),
            MetricDetailLevel::Frontend,
            Duration::from_secs(60),
            PeerBinding::default(),
            now,
        ));
        assert_eq!(prev, MetricDetailLevel::Backend);
        assert_eq!(new, MetricDetailLevel::Frontend);
        assert_eq!(agg.detail_effective(), MetricDetailLevel::Frontend);
        assert_eq!(agg.lease_count(), 1);
    }

    #[test]
    fn lease_apply_renewal_rejects_foreign_binding() {
        // Same-UID `client_id` collision attack: the victim applies with
        // a known binding; an attacker that learns the `client_id`
        // attempts to renew under a different (pid, session_ulid). The
        // renewal must be refused so the victim's apply-time binding
        // remains the sole authoritative owner — both for subsequent
        // renewals AND for the victim's own Drop-time `clear`.
        let mut agg = Aggregator::new("sozu".to_owned());
        let now = Instant::now();
        let victim = PeerBinding {
            pid: Some(4242),
            session_ulid: Some(0x0123_4567_89AB_CDEF_FEDC_BA98_7654_3210),
        };
        let outcome = agg.lease_apply(
            "topcli".to_owned(),
            MetricDetailLevel::Backend,
            Duration::from_secs(60),
            victim,
            now,
        );
        assert!(
            matches!(outcome, LeaseApplyOutcome::Applied { .. }),
            "victim's initial apply must succeed"
        );
        let attacker = PeerBinding {
            pid: Some(9999),
            session_ulid: Some(0xDEAD_BEEF_DEAD_BEEF_DEAD_BEEF_DEAD_BEEF),
        };
        let outcome = agg.lease_apply(
            "topcli".to_owned(),
            MetricDetailLevel::Backend,
            Duration::from_secs(60),
            attacker,
            now,
        );
        assert_eq!(
            outcome,
            LeaseApplyOutcome::Unauthorized,
            "renewal with a mismatched known binding must be refused"
        );
        // The victim can still clear their own lease — proof the
        // refused attempt did not corrupt the stored binding.
        let clear = agg.lease_clear("topcli", victim);
        assert!(
            matches!(clear, LeaseClearOutcome::Cleared { .. }),
            "victim's original binding must still clear cleanly after \
             the foreign-binding renewal was refused"
        );
    }

    #[test]
    fn lease_apply_renewal_with_matching_binding_succeeds() {
        // Symmetry case: the legitimate owner re-applies with the same
        // (pid, session_ulid). The renewal must succeed so the TUI's
        // own renewer thread keeps the lease alive across its TTL.
        let mut agg = Aggregator::new("sozu".to_owned());
        let now = Instant::now();
        let owner = PeerBinding {
            pid: Some(1234),
            session_ulid: Some(0xAAAA_BBBB_CCCC_DDDD_EEEE_FFFF_0000_1111),
        };
        let _ = agg.lease_apply(
            "topcli".to_owned(),
            MetricDetailLevel::Backend,
            Duration::from_secs(30),
            owner,
            now,
        );
        let outcome = agg.lease_apply(
            "topcli".to_owned(),
            MetricDetailLevel::Backend,
            Duration::from_secs(60),
            owner,
            now,
        );
        assert!(
            matches!(outcome, LeaseApplyOutcome::Applied { .. }),
            "renewal with matching binding must succeed (otherwise the \
             TUI's own renewer thread would be locked out)"
        );
    }

    #[test]
    fn lease_apply_max_merge_two_clients() {
        // Two clients, two levels: effective = max(both leases, configured).
        // Use `Process` as the floor so the Frontend lease is observable
        // after the Backend lease is cleared (otherwise the configured
        // Cluster floor would mask the Frontend lease).
        let mut agg = Aggregator::new("sozu".to_owned());
        agg.set_up_detail(MetricDetailLevel::Process);
        let now = Instant::now();
        let _ = agg.lease_apply(
            "scraper".to_owned(),
            MetricDetailLevel::Frontend,
            Duration::from_secs(60),
            PeerBinding::default(),
            now,
        );
        let _ = agg.lease_apply(
            "topcli".to_owned(),
            MetricDetailLevel::Backend,
            Duration::from_secs(60),
            PeerBinding::default(),
            now,
        );
        assert_eq!(agg.detail_effective(), MetricDetailLevel::Backend);
        assert_eq!(agg.lease_count(), 2);
        // Clearing the higher lease drops effective back to the lower one.
        let outcome = agg.lease_clear("topcli", PeerBinding::default());
        assert_eq!(
            outcome,
            LeaseClearOutcome::Cleared {
                previous_effective: MetricDetailLevel::Backend,
            }
        );
        assert_eq!(agg.detail_effective(), MetricDetailLevel::Frontend);
        assert_eq!(agg.lease_count(), 1);
    }

    #[test]
    fn lease_clear_unknown_id_is_silent_noop() {
        // Mismatched IDs are silently ignored (other clients' leases unaffected).
        let mut agg = Aggregator::new("sozu".to_owned());
        let _ = agg.lease_apply(
            "real".to_owned(),
            MetricDetailLevel::Backend,
            Duration::from_secs(60),
            PeerBinding::default(),
            Instant::now(),
        );
        assert_eq!(
            agg.lease_clear("ghost", PeerBinding::default()),
            LeaseClearOutcome::NotFound
        );
        assert_eq!(agg.detail_effective(), MetricDetailLevel::Backend);
        assert_eq!(agg.lease_count(), 1);
    }

    #[test]
    fn lease_clear_with_matching_binding_authorised() {
        // Apply with a known binding, clear with the same binding -> Cleared.
        let mut agg = Aggregator::new("sozu".to_owned());
        let _ = agg.lease_apply(
            "owner-lease".to_owned(),
            MetricDetailLevel::Backend,
            Duration::from_secs(60),
            owner_binding(),
            Instant::now(),
        );
        let outcome = agg.lease_clear("owner-lease", owner_binding());
        assert!(matches!(outcome, LeaseClearOutcome::Cleared { .. }));
        assert_eq!(agg.lease_count(), 0);
    }

    #[test]
    fn lease_clear_with_mismatched_binding_is_unauthorized() {
        // Apply with one binding, attempt clear with a different binding ->
        // Unauthorized; lease left intact.
        let mut agg = Aggregator::new("sozu".to_owned());
        let _ = agg.lease_apply(
            "owner-lease".to_owned(),
            MetricDetailLevel::Backend,
            Duration::from_secs(60),
            owner_binding(),
            Instant::now(),
        );
        let outcome = agg.lease_clear("owner-lease", other_binding());
        assert_eq!(outcome, LeaseClearOutcome::Unauthorized);
        assert_eq!(agg.lease_count(), 1);
    }

    #[test]
    fn lease_clear_unknown_apply_binding_accepts_any_clear() {
        // Pre-binding / non-Linux apply -> any clear authorised.
        let mut agg = Aggregator::new("sozu".to_owned());
        let _ = agg.lease_apply(
            "legacy".to_owned(),
            MetricDetailLevel::Backend,
            Duration::from_secs(60),
            PeerBinding::default(),
            Instant::now(),
        );
        let outcome = agg.lease_clear("legacy", owner_binding());
        assert!(matches!(outcome, LeaseClearOutcome::Cleared { .. }));
        assert_eq!(agg.lease_count(), 0);
    }

    #[test]
    fn lease_clear_known_apply_rejects_default_clear() {
        // Known apply binding -> a default ("unknown") clear is rejected.
        let mut agg = Aggregator::new("sozu".to_owned());
        let _ = agg.lease_apply(
            "owner-lease".to_owned(),
            MetricDetailLevel::Backend,
            Duration::from_secs(60),
            owner_binding(),
            Instant::now(),
        );
        let outcome = agg.lease_clear("owner-lease", PeerBinding::default());
        assert_eq!(outcome, LeaseClearOutcome::Unauthorized);
    }

    #[test]
    fn lease_tick_expires_only_past_due_leases() {
        // `lease_tick(now)` parameterises the clock so we can test expiry
        // without sleeping. Setup: one lease past due, one still active.
        // Use `Process` as the floor so the surviving Frontend lease drives
        // the effective level after the Backend lease expires (otherwise
        // the Cluster floor would mask it).
        let mut agg = Aggregator::new("sozu".to_owned());
        agg.set_up_detail(MetricDetailLevel::Process);
        let now = Instant::now();
        // Inject directly into the table to control expires_at deterministically.
        agg.leases.insert(
            "expired".to_owned(),
            LeaseEntry {
                level: MetricDetailLevel::Backend,
                expires_at: now - Duration::from_secs(1),
                binding: PeerBinding::default(),
            },
        );
        agg.leases.insert(
            "live".to_owned(),
            LeaseEntry {
                level: MetricDetailLevel::Frontend,
                expires_at: now + Duration::from_secs(60),
                binding: PeerBinding::default(),
            },
        );
        agg.recompute_effective();
        assert_eq!(agg.detail_effective(), MetricDetailLevel::Backend);
        let prev = agg.lease_tick(now);
        assert_eq!(prev, Some(MetricDetailLevel::Backend));
        assert_eq!(agg.detail_effective(), MetricDetailLevel::Frontend);
        assert_eq!(agg.lease_count(), 1);
    }

    #[test]
    fn lease_tick_no_change_returns_none() {
        // No leases -> no-op tick, no audit signal.
        let mut agg = Aggregator::new("sozu".to_owned());
        assert!(agg.lease_tick(Instant::now()).is_none());
    }

    #[test]
    fn lease_apply_at_max_ttl_succeeds() {
        // Boundary: exactly LEASE_TTL_MAX is allowed; LEASE_TTL_MAX + 1ns is
        // rejected (covered by lease_apply_rejects_ttl_over_max above). With
        // `now` injected (rather than read from the host clock) the expiry
        // is checked for EXACT equality to `now + LEASE_TTL_MAX`, not a
        // fuzzy upper bound — the whole point of parameterising the clock.
        let mut agg = Aggregator::new("sozu".to_owned());
        let now = Instant::now();
        let outcome = agg.lease_apply(
            "max".to_owned(),
            MetricDetailLevel::Backend,
            LEASE_TTL_MAX,
            PeerBinding::default(),
            now,
        );
        assert!(matches!(outcome, LeaseApplyOutcome::Applied { .. }));
        let entry = agg.leases.get("max").unwrap();
        assert_eq!(entry.expires_at, now + LEASE_TTL_MAX);
    }
}