liminal-server 0.25.4

//! Server-side metric recording over the process-global liminal registry.
//!
//! [`init`] installs a global [`MetricsRegistry`] — which flips the library's
//! metrics gate on for this process — and registers the first-wave server
//! families, caching their handles. Every recording helper no-ops until `init`
//! has run, so a standalone liminal library user (who never calls `init`) pays
//! nothing and the registry stays disabled.

use std::sync::OnceLock;

use liminal::durability::{DedupHeadCounts, dedup_head_counts};
use liminal::metrics::{
    CounterHandle, GaugeHandle, MetricRegistrationError, MetricsRegistry, global_registry,
    install_global_registry,
};

use crate::console::counts::ConsoleFailure;
use crate::server::connection::refusal::AdmissionRefusal;
use crate::server::connection::websocket::UpgradeRefusal;
use crate::server::mount::MountKind;

const CONNECTIONS_ACTIVE: &str = "liminal_connections_active";
const PUBLISHES_TOTAL: &str = "liminal_publishes_total";
const DELIVERIES_TOTAL: &str = "liminal_deliveries_total";
/// Second-wave: deliveries split by the transport that carried them.
///
/// ADDITIVE. [`DELIVERIES_TOTAL`] keeps its meaning and its exposition line
/// unchanged — an operator's existing `/metrics` scrape sees every line it saw
/// before, plus these. The two families count different things and are not
/// expected to agree: the aggregate counts subscriber deliveries accepted at
/// PUBLISH time (channel-actor fan-out), this one counts `Deliver` frames the
/// per-connection pump actually enqueued toward a socket.
const TRANSPORT_DELIVERIES_TOTAL: &str = "liminal_transport_deliveries_total";
/// Second-wave: subscriptions shed by an inbox overflow (P0 #55).
///
/// Deliberately UNLABELLED. A per-channel or per-subscription shed counter is
/// unbounded cardinality on a surface a scraper keeps forever; the channel and
/// subscription of any individual shed are on the `warn` line the shed emits
/// beside this increment, where they cost one log record rather than a permanent
/// time series.
const SHEDS_TOTAL: &str = "liminal_subscription_sheds_total";
const SELF_STREAM_APPEND_FAILURES_TOTAL: &str = "liminal_self_stream_append_failures_total";
const SELF_STREAMS_RESERVED_PUBLICATIONS_TOTAL: &str =
    "liminal_self_streams_reserved_publications_total";
/// Starts that omitted self streams because no participant service was installed.
const SELF_STREAMS_DISABLED_TOTAL: &str = "liminal_self_streams_disabled_total";
/// SRV-012 R4: set to 1 when the console streams are off for want of a
/// participant (no publisher owner runs them).
const CONSOLE_STREAMS_DISABLED_TOTAL: &str = "liminal_console_streams_disabled_total";
/// SRV-012 R8: console bookkeeping failures, by `kind` (bounded by
/// `ConsoleFailure`'s variants).
const CONSOLE_FAILURES_TOTAL: &str = "liminal_console_failures_total";
const KIND_LABEL: &str = "kind";
const SELF_STREAMS_UNMATCHED_PRESSURE_TOTAL: &str = "liminal_self_streams_unmatched_pressure_total";
const SELF_STREAMS_SUPPRESSED_PRESSURE_TOTAL: &str =
    "liminal_self_streams_suppressed_pressure_total";
const SELF_STREAMS_PRESSURE_LEDGER_EVICTIONS_TOTAL: &str =
    "liminal_self_streams_pressure_ledger_evictions_total";
/// Backpressure signals (`Accept`/`Defer`/`Reject`) refused by the access
/// decision on their subscription's subject: the signal has no refusal frame,
/// so the count is the only place the refusal shows besides the log line.
const PRESSURE_REFUSED_TOTAL: &str = "liminal_pressure_refused_total";
/// Label key carried by [`TRANSPORT_DELIVERIES_TOTAL`]. Cardinality is bounded by
/// [`MountKind`]'s variants, which is why THIS split is affordable and a
/// per-channel one is not.
const TRANSPORT_LABEL: &str = "transport";
/// P0 #56: connections turned away at the admission door, by reason class.
///
/// ADDITIVE, and it has to be, because no existing family can answer this.
/// [`CONNECTIONS_ACTIVE`] is incremented in `register_record`
/// (`server/connection/supervisor.rs:2822`), which is reached only AFTER
/// admission has succeeded — a refusal is exactly the case where it never runs.
/// On a server refusing every connection that gauge therefore reads a flat
/// zero, indistinguishable from a healthy idle server, which is what the field
/// estate's dashboards showed while 82,166 consecutive connections were turned
/// away. This counter can only move when a connection was refused.
const ADMISSION_REFUSALS_TOTAL: &str = "liminal_admission_refusals_total";
/// P0 #56: WebSocket upgrades refused before admission was even attempted.
///
/// Separate from [`ADMISSION_REFUSALS_TOTAL`] because it counts a different
/// event at a different door: these connections never reached the shared
/// admission bound, the incarnation authority, or the registry. Folding them
/// together would make an origin-policy rejection look like a capacity problem.
const HANDSHAKE_REFUSALS_TOTAL: &str = "liminal_handshake_refusals_total";
/// Label key carried by both refusal families.
///
/// Cardinality is bounded by the variant count of `AdmissionRefusal` and
/// `UpgradeRefusal` respectively — fixed, enum-derived strings. A peer address,
/// an origin value, or an error message here would be unbounded cardinality on
/// a surface a scraper keeps forever.
const REASON_LABEL: &str = "reason";
/// Lane p0-39: entries displaced out of a per-participant retention window.
///
/// ADDITIVE, and it has to be: the event it counts did not exist before. The
/// stage-8 receipt windows used to REFUSE at their bound, which was loud on the
/// wire (a typed `ReceiptCapacityExceeded` the client could see). They now
/// displace instead, which is silent to the arriving client BY DESIGN — the
/// (N+1)th honest fingerprint lands. A bound that neither refuses nor discloses
/// would hide exactly what the old wall at least made loud, so displacement is
/// silent to experience and loud to record: this counter is the record.
const RECEIPT_DISPLACEMENTS_TOTAL: &str = "liminal_receipt_displacements_total";
/// Lane p0-39: observations that a SHARED receipt pool is carrying a churn
/// storm (`liminal_receipt_pool_runaway_total`).
///
/// The three shared pools stopped being admission gates entirely — no
/// configured number may refuse an honest third party there — so their only
/// bound is the TTL window. This counter is the tripwire that replaces the
/// wall: it is an OCCUPANCY OBSERVATION, never a gate, and nothing is refused
/// because it moved. It increments once per admitted operation that observed a
/// pool at or above its configured reporting threshold, so its rate is the
/// storm's rate.
const RECEIPT_POOL_RUNAWAY_TOTAL: &str = "liminal_receipt_pool_runaway_total";
/// Connection fates currently parked on candidate-lane `Precedence`.
///
/// A parked fate holds a durable Open with no Complete. It waits on the drain of
/// a lane another participant's terminal or a compaction marker occupies, and
/// while it waits nothing on the wire shows it. A gauge that stays above zero is
/// a conversation nobody has published into since, and a non-zero value at a
/// clean stop is a stop that could not settle.
const CONNECTION_FATES_PARKED: &str = "liminal_connection_fates_parked";
/// Label key carried by [`RECEIPT_DISPLACEMENTS_TOTAL`]. Cardinality is bounded
/// by [`ReceiptWindowScope`]'s two variants.
const SCOPE_LABEL: &str = "scope";
/// Label key carried by [`RECEIPT_POOL_RUNAWAY_TOTAL`]. Cardinality is bounded
/// by [`SharedReceiptPool`]'s three variants.
const POOL_LABEL: &str = "pool";

/// The per-participant retention windows that can displace an older entry.
///
/// A fixed, enum-derived label vocabulary: these are the only two scopes whose
/// configured number is a window size rather than a gate.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum ReceiptWindowScope {
    /// Stage-8 `LiveReceiptParticipant`.
    LiveReceiptParticipant,
    /// Stage-8 `ProvenanceParticipant`.
    ProvenanceParticipant,
}

impl ReceiptWindowScope {
    /// Every window scope, in handle-storage order.
    pub(crate) const LABELS: [&'static str; 2] =
        ["live_receipt_participant", "provenance_participant"];

    const fn slot(self) -> usize {
        match self {
            Self::LiveReceiptParticipant => 0,
            Self::ProvenanceParticipant => 1,
        }
    }
}

/// The shared receipt pools whose retention is TTL-bounded and tripwired.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum SharedReceiptPool {
    /// Stage-8 `LiveReceiptServer`.
    LiveReceiptServer,
    /// Stage-8 `ProvenanceServer`.
    ProvenanceServer,
    /// Stage-8 `ProvenanceConversation`.
    ProvenanceConversation,
}

impl SharedReceiptPool {
    /// Every shared pool, in handle-storage order.
    pub(crate) const LABELS: [&'static str; 3] = [
        "live_receipt_server",
        "provenance_server",
        "provenance_conversation",
    ];

    const fn slot(self) -> usize {
        match self {
            Self::LiveReceiptServer => 0,
            Self::ProvenanceServer => 1,
            Self::ProvenanceConversation => 2,
        }
    }

    /// Human-readable pool name for the paired rising-edge warning.
    pub(crate) const fn label(self) -> &'static str {
        Self::LABELS[self.slot()]
    }
}

static SERVER_METRICS: OnceLock<ServerMetrics> = OnceLock::new();

/// Cached handles for the first-wave server metrics.
#[derive(Clone, Debug)]
struct ServerMetrics {
    connections_active: GaugeHandle,
    connection_fates_parked: GaugeHandle,
    publishes_total: CounterHandle,
    deliveries_total: CounterHandle,
    /// One pre-registered handle per [`MountKind`], so the delivery hot path
    /// looks a counter up by enum discriminant and never registers, allocates a
    /// label string, or hashes a name mid-slice.
    transport_deliveries: [CounterHandle; MOUNT_KINDS.len()],
    sheds_total: CounterHandle,
    self_stream_append_failures_total: CounterHandle,
    self_streams_disabled_total: CounterHandle,
    console_streams_disabled_total: CounterHandle,
    console_failures: [CounterHandle; ConsoleFailure::LABELS.len()],
    self_streams_reserved_publications_total: CounterHandle,
    self_streams_unmatched_pressure_total: CounterHandle,
    self_streams_suppressed_pressure_total: CounterHandle,
    self_streams_pressure_ledger_evictions_total: CounterHandle,
    pressure_refused_total: CounterHandle,
    /// One pre-registered handle per admission-refusal class, so the refusal
    /// path looks a counter up by discriminant and never allocates a label
    /// string while it is busy turning a connection away.
    admission_refusals: [CounterHandle; AdmissionRefusal::LABELS.len()],
    /// One pre-registered handle per WebSocket upgrade-refusal class.
    handshake_refusals: [CounterHandle; UpgradeRefusal::LABELS.len()],
    /// One pre-registered handle per per-participant retention window.
    receipt_displacements: [CounterHandle; ReceiptWindowScope::LABELS.len()],
    /// One pre-registered handle per shared receipt pool.
    receipt_pool_runaway: [CounterHandle; SharedReceiptPool::LABELS.len()],
}

/// Every transport a delivery can ride, in the order their handles are stored.
const MOUNT_KINDS: [MountKind; 3] = [MountKind::Tcp, MountKind::WebSocket, MountKind::Loopback];

const fn transport_slot(transport: MountKind) -> usize {
    match transport {
        MountKind::Tcp => 0,
        MountKind::WebSocket => 1,
        MountKind::Loopback => 2,
    }
}

/// Enables metrics for this server process and registers the server families.
///
/// Idempotent: a second call is a no-op. Called once at server startup so the
/// `/metrics` endpoint has data to render; the recording helpers below stay
/// inert until this runs.
pub fn init() {
    if SERVER_METRICS.get().is_some() {
        return;
    }
    let Some(registry) = global_or_install() else {
        tracing::warn!(
            "global metrics registry could not be installed or read back; server metrics stay disabled"
        );
        return;
    };
    if let Some(metrics) = ServerMetrics::register(registry) {
        let _ = SERVER_METRICS.set(metrics);
    }
}

/// Records the spawn of a supervised connection (`liminal_connections_active`
/// gauge increment). Paired with [`connection_closed`] on every teardown route.
pub fn connection_spawned() {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.connections_active.increment();
    }
}

/// Records the teardown of a supervised connection (`liminal_connections_active`
/// gauge decrement). Paired with [`connection_spawned`].
pub fn connection_closed() {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.connections_active.decrement();
    }
}

/// Records one connection fate parked on candidate-lane `Precedence`
/// (`liminal_connection_fates_parked` gauge increment). Paired with
/// [`connection_fate_unparked`] when the fate leaves the index.
pub(crate) fn connection_fate_parked() {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.connection_fates_parked.increment();
    }
}

/// Records one parked connection fate leaving the index for its retry
/// (`liminal_connection_fates_parked` gauge decrement).
pub(crate) fn connection_fate_unparked() {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.connection_fates_parked.decrement();
    }
}

/// Records one accepted publish on the services publish path
/// (`liminal_publishes_total`).
pub fn publish_accepted() {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.publishes_total.increment();
    }
}

/// Records `count` genuine subscriber deliveries from a single publish
/// (`liminal_deliveries_total`). A publish that reached no subscriber records
/// nothing.
pub fn deliveries_recorded(count: u64) {
    if count == 0 {
        return;
    }
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.deliveries_total.increment_by(count);
    }
}

/// Records pump deliveries on `transport` (`liminal_transport_deliveries_total`).
///
/// `count` is the number of `Deliver` frames one connection slice enqueued. A
/// slice that delivered nothing records nothing, so an idle connection touches no
/// atomic at all.
pub fn transport_deliveries_recorded(transport: MountKind, count: u64) {
    if count == 0 {
        return;
    }
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.transport_deliveries[transport_slot(transport)].increment_by(count);
    }
}

/// Records one connection turned away at the admission door
/// (`liminal_admission_refusals_total`, labelled by reason class).
///
/// Called at the three admission doors — TCP accept, WebSocket upgrade, and the
/// in-process loopback — and nowhere else, so a refusal is counted exactly once
/// no matter which mount knocked.
pub(crate) fn admission_refused(refusal: AdmissionRefusal) {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.admission_refusals[refusal.slot()].increment();
    }
}

/// Records one WebSocket upgrade refused before admission was attempted
/// (`liminal_handshake_refusals_total`, labelled by reason class).
pub(crate) fn handshake_refused(refusal: &UpgradeRefusal) {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.handshake_refusals[refusal.slot()].increment();
    }
}

pub(crate) fn self_stream_append_failed() {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.self_stream_append_failures_total.increment();
    }
}

pub(crate) fn self_stream_reserved_publication() {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.self_streams_reserved_publications_total.increment();
    }
}

pub(crate) fn self_streams_disabled() {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.self_streams_disabled_total.increment();
    }
}

/// SRV-012 R8: `amount` console bookkeeping failures of `kind`.
pub(crate) fn console_failure(kind: ConsoleFailure, amount: u64) {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.console_failures[kind.slot()].increment_by(amount);
    }
}

/// SRV-012 R4: the console streams are off for want of a participant.
pub(crate) fn console_streams_disabled() {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.console_streams_disabled_total.increment();
    }
}

pub(crate) fn self_stream_pressure_unmatched() {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.self_streams_unmatched_pressure_total.increment();
    }
}

pub(crate) fn self_stream_pressure_suppressed() {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.self_streams_suppressed_pressure_total.increment();
    }
}

pub(crate) fn pressure_refused() {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.pressure_refused_total.increment();
    }
}

pub(crate) fn pressure_ledger_evicted() {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics
            .self_streams_pressure_ledger_evictions_total
            .increment();
    }
}

/// Records one subscription shed by an inbox overflow
/// (`liminal_subscription_sheds_total`).
///
/// Paired at its only call site with the `warn` that names the channel,
/// subscription and transport: the counter is what an alert fires on, the log
/// line is what the alert is then read against.
pub fn subscription_shed() {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.sheds_total.increment();
    }
}

/// Records `count` entries displaced out of one per-participant retention
/// window (`liminal_receipt_displacements_total`, labelled by scope).
///
/// The arriving client sees nothing — that is the ruled behaviour, "silent to
/// experience" — so this counter and the `debug` line beside it are the only
/// record that a bound did work. A zero count returns early, so the ordinary
/// with-headroom path touches no atomic.
pub(crate) fn receipt_entries_displaced(scope: ReceiptWindowScope, count: u64) {
    if count == 0 {
        return;
    }
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.receipt_displacements[scope.slot()].increment_by(count);
    }
}

/// Records one observation of a shared receipt pool at or above its configured
/// reporting threshold (`liminal_receipt_pool_runaway_total`, labelled by pool).
///
/// An OBSERVATION, not a refusal: the operation that made it was admitted. The
/// rising-edge `warn` beside the first such observation carries the occupancy
/// and threshold; this counter carries the storm's rate.
pub(crate) fn receipt_pool_runaway_observed(pool: SharedReceiptPool) {
    if let Some(metrics) = SERVER_METRICS.get() {
        metrics.receipt_pool_runaway[pool.slot()].increment();
    }
}

/// Appends the dedup head counters (DUR-008 A12) to a `/metrics` body: one
/// unlabelled counter family per name of
/// [`DedupHeadCounts::named`](liminal::durability::DedupHeadCounts::named),
/// all three from one [`dedup_head_counts`] read.
///
/// Rendered at SCRAPE time, never registered. The counters are the library's
/// process-global atomics, moved on the dedup lookup path by one relaxed
/// `fetch_add` with no registry in reach; a registry handle mirroring them
/// would need that path to bump a second atomic, and `liminal::durability` to
/// know the server's registry, or a sampler copying them on a clock. Read here
/// they cost three relaxed loads per scrape, take no lock, and touch nothing a
/// publish or a dedup lookup holds.
///
/// The names are the library's own, iterated rather than retyped, so the
/// exposition cannot drift from them. They carry the server's `liminal_`
/// prefix and the counter `_total` suffix; no registry family is named
/// `liminal_dedup_head_*`, so no family is declared twice. The header lines
/// are the ones `liminal::metrics::render` writes.
pub(crate) fn render_dedup_head_counters(output: &mut String) {
    render_dedup_head_block(output, dedup_head_counts());
}

/// The exposition block for one reading of the dedup head counters: for each
/// counter, in [`DedupHeadCounts::named`]'s order, its `# HELP` line, its
/// `# TYPE` line and its value line.
fn render_dedup_head_block(output: &mut String, counts: DedupHeadCounts) {
    for (name, value) in counts.named() {
        let value = value.to_string();
        for piece in [
            "# HELP ",
            name,
            " ",
            name,
            "\n# TYPE ",
            name,
            " counter\n",
            name,
            " ",
            &value,
            "\n",
        ] {
            output.push_str(piece);
        }
    }
}

/// The current value of the accepted-publish counter, for a test that needs an
/// UNRELATED counter to prove its harness measured anything at all.
///
/// `None` means the family is not readable — either [`init`] has not run in this
/// process or the registry holds no such counter — which a caller must treat as
/// "no measurement", never as zero. The name is read from the same constant the
/// registration uses, so the two cannot drift.
#[cfg(test)]
pub(crate) fn publishes_total_value() -> Option<u64> {
    use liminal::metrics::MetricValue;

    let registry = global_registry()?;
    registry
        .snapshot()
        .metrics()
        .iter()
        .find(|metric| metric.name == PUBLISHES_TOTAL)
        .and_then(|metric| match metric.value {
            MetricValue::Counter(value) => Some(value),
            MetricValue::Gauge(_) | MetricValue::Histogram(_) => None,
        })
}

/// The current value of one lane p0-39 displacement counter, by scope.
///
/// `None` means the family is not readable — either [`init`] has not run in
/// this process or the registry holds no such counter — which a caller must
/// treat as "no measurement", never as zero. Name and label are read from the
/// same constants the registration uses, so a rename shows up as a failure
/// rather than as a vacuous pass.
#[cfg(test)]
pub(crate) fn receipt_displacements_value(scope: ReceiptWindowScope) -> Option<u64> {
    labelled_counter_value(
        RECEIPT_DISPLACEMENTS_TOTAL,
        SCOPE_LABEL,
        ReceiptWindowScope::LABELS[scope.slot()],
    )
}

/// The current value of one lane p0-39 shared-pool tripwire counter, by pool.
///
/// `None` means "not measured", never zero — see
/// [`receipt_displacements_value`].
#[cfg(test)]
pub(crate) fn receipt_pool_runaway_value(pool: SharedReceiptPool) -> Option<u64> {
    labelled_counter_value(RECEIPT_POOL_RUNAWAY_TOTAL, POOL_LABEL, pool.label())
}

/// The current value of the parked connection-fate gauge.
///
/// `None` means "not measured", never zero — see
/// [`receipt_displacements_value`].
#[cfg(test)]
pub(crate) fn connection_fates_parked_value() -> Option<i64> {
    use liminal::metrics::MetricValue;

    let registry = global_registry()?;
    registry
        .snapshot()
        .metrics()
        .iter()
        .find(|metric| metric.name == CONNECTION_FATES_PARKED)
        .and_then(|metric| match metric.value {
            MetricValue::Gauge(value) => Some(value),
            MetricValue::Counter(_) | MetricValue::Histogram(_) => None,
        })
}

#[cfg(test)]
fn labelled_counter_value(name: &str, key: &str, label: &str) -> Option<u64> {
    use liminal::metrics::MetricValue;

    let registry = global_registry()?;
    registry
        .snapshot()
        .metrics()
        .iter()
        .find(|metric| {
            metric.name == name
                && metric
                    .labels
                    .iter()
                    .any(|(metric_key, value)| metric_key == key && value == label)
        })
        .and_then(|metric| match metric.value {
            MetricValue::Counter(value) => Some(value),
            MetricValue::Gauge(_) | MetricValue::Histogram(_) => None,
        })
}

impl ServerMetrics {
    fn register(registry: &MetricsRegistry) -> Option<Self> {
        Some(Self {
            connections_active: register_gauge(registry, CONNECTIONS_ACTIVE)?,
            connection_fates_parked: register_gauge(registry, CONNECTION_FATES_PARKED)?,
            publishes_total: register_counter(registry, PUBLISHES_TOTAL)?,
            deliveries_total: register_counter(registry, DELIVERIES_TOTAL)?,
            transport_deliveries: register_transport_deliveries(registry)?,
            sheds_total: register_counter(registry, SHEDS_TOTAL)?,
            self_stream_append_failures_total: register_counter(
                registry,
                SELF_STREAM_APPEND_FAILURES_TOTAL,
            )?,
            self_streams_disabled_total: register_counter(registry, SELF_STREAMS_DISABLED_TOTAL)?,
            console_streams_disabled_total: register_counter(
                registry,
                CONSOLE_STREAMS_DISABLED_TOTAL,
            )?,
            console_failures: register_labelled(
                registry,
                CONSOLE_FAILURES_TOTAL,
                KIND_LABEL,
                &ConsoleFailure::LABELS,
            )?,
            self_streams_reserved_publications_total: register_counter(
                registry,
                SELF_STREAMS_RESERVED_PUBLICATIONS_TOTAL,
            )?,
            self_streams_unmatched_pressure_total: register_counter(
                registry,
                SELF_STREAMS_UNMATCHED_PRESSURE_TOTAL,
            )?,
            self_streams_suppressed_pressure_total: register_counter(
                registry,
                SELF_STREAMS_SUPPRESSED_PRESSURE_TOTAL,
            )?,
            self_streams_pressure_ledger_evictions_total: register_counter(
                registry,
                SELF_STREAMS_PRESSURE_LEDGER_EVICTIONS_TOTAL,
            )?,
            pressure_refused_total: register_counter(registry, PRESSURE_REFUSED_TOTAL)?,
            admission_refusals: register_labelled(
                registry,
                ADMISSION_REFUSALS_TOTAL,
                REASON_LABEL,
                &AdmissionRefusal::LABELS,
            )?,
            handshake_refusals: register_labelled(
                registry,
                HANDSHAKE_REFUSALS_TOTAL,
                REASON_LABEL,
                &UpgradeRefusal::LABELS,
            )?,
            receipt_displacements: register_labelled(
                registry,
                RECEIPT_DISPLACEMENTS_TOTAL,
                SCOPE_LABEL,
                &ReceiptWindowScope::LABELS,
            )?,
            receipt_pool_runaway: register_labelled(
                registry,
                RECEIPT_POOL_RUNAWAY_TOTAL,
                POOL_LABEL,
                &SharedReceiptPool::LABELS,
            )?,
        })
    }
}

/// One unlabelled counter, registered; `None` after the warning when the
/// registry refuses it.
fn register_counter(registry: &MetricsRegistry, name: &'static str) -> Option<CounterHandle> {
    registry
        .register_counter(name, no_labels())
        .inspect_err(warn_registration_failure)
        .ok()
}

/// One unlabelled gauge, registered; `None` after the warning when the
/// registry refuses it.
fn register_gauge(registry: &MetricsRegistry, name: &'static str) -> Option<GaugeHandle> {
    registry
        .register_gauge(name, no_labels())
        .inspect_err(warn_registration_failure)
        .ok()
}

fn register_transport_deliveries(
    registry: &MetricsRegistry,
) -> Option<[CounterHandle; MOUNT_KINDS.len()]> {
    let mut transport_deliveries = Vec::with_capacity(MOUNT_KINDS.len());
    for transport in MOUNT_KINDS {
        transport_deliveries.push(
            registry
                .register_counter(
                    TRANSPORT_DELIVERIES_TOTAL,
                    [(TRANSPORT_LABEL, transport.as_str())],
                )
                .inspect_err(warn_registration_failure)
                .ok()?,
        );
    }
    transport_deliveries.try_into().ok()
}

/// Pre-registers one counter per label value, in label order.
///
/// Every class is registered at `init`, not lazily on first refusal, so the
/// exposition carries an explicit zero for classes that have not fired. A
/// missing line and a zero line mean very different things to an operator: the
/// first is "this server does not know about that failure mode", the second is
/// "it has not happened".
fn register_labelled<const N: usize>(
    registry: &MetricsRegistry,
    name: &'static str,
    key: &'static str,
    labels: &[&'static str; N],
) -> Option<[CounterHandle; N]> {
    let mut handles = Vec::with_capacity(N);
    for label in labels {
        handles.push(
            registry
                .register_counter(name, [(key, *label)])
                .inspect_err(warn_registration_failure)
                .ok()?,
        );
    }
    handles.try_into().ok()
}

/// Names a failed family registration instead of letting `.ok()?` discard it.
///
/// `register` still answers `None` and metrics recording stays disabled; what
/// changes is that the operator can see why from the log.
fn warn_registration_failure(error: &MetricRegistrationError) {
    tracing::warn!(
        %error,
        "server metrics family registration failed; metrics recording stays disabled"
    );
}

const fn no_labels() -> std::iter::Empty<(&'static str, &'static str)> {
    std::iter::empty()
}

/// Returns the process-global registry, installing a fresh one when none exists.
///
/// Enabling the gate here (rather than in the library) keeps standalone liminal
/// users on the disabled fast path; the server is the sole installer.
fn global_or_install() -> Option<&'static MetricsRegistry> {
    if let Some(registry) = global_registry() {
        return Some(registry);
    }
    // Best-effort install; if a concurrent caller won the race we still read the
    // now-installed registry back below.
    let _ = install_global_registry(MetricsRegistry::new());
    global_registry()
}

#[cfg(test)]
mod tests {
    use super::{
        CONNECTIONS_ACTIVE, DELIVERIES_TOTAL, PUBLISHES_TOTAL, ServerMetrics, connection_spawned,
        deliveries_recorded, init, no_labels, publish_accepted, render_dedup_head_block,
    };
    use liminal::durability::DedupHeadCounts;
    use liminal::metrics::{MetricsRegistry, global_registry, render};

    /// The dedup head block, byte for byte, for one reading with three
    /// distinct values: each counter's `# HELP`, `# TYPE` and value line, in
    /// `named()` order, each line ended by one newline.
    #[test]
    fn the_dedup_head_block_renders_byte_for_byte() {
        let mut output = String::new();
        render_dedup_head_block(
            &mut output,
            DedupHeadCounts {
                refusals: 7,
                record_lost_races: 0,
                repairs: 12,
            },
        );
        assert_eq!(
            output,
            "# HELP liminal_dedup_head_refusals_total liminal_dedup_head_refusals_total\n\
             # TYPE liminal_dedup_head_refusals_total counter\n\
             liminal_dedup_head_refusals_total 7\n\
             # HELP liminal_dedup_head_record_lost_races_total \
             liminal_dedup_head_record_lost_races_total\n\
             # TYPE liminal_dedup_head_record_lost_races_total counter\n\
             liminal_dedup_head_record_lost_races_total 0\n\
             # HELP liminal_dedup_head_repairs_total liminal_dedup_head_repairs_total\n\
             # TYPE liminal_dedup_head_repairs_total counter\n\
             liminal_dedup_head_repairs_total 12\n"
        );
    }

    #[test]
    fn init_registers_the_three_server_families_on_the_global_registry()
    -> Result<(), Box<dyn std::error::Error>> {
        init();
        connection_spawned();
        publish_accepted();
        deliveries_recorded(2);

        let registry =
            global_registry().ok_or("init must install and enable the global registry")?;
        let exposition = render(&registry.snapshot());

        assert!(exposition.contains(CONNECTIONS_ACTIVE));
        assert!(exposition.contains(PUBLISHES_TOTAL));
        assert!(exposition.contains(DELIVERIES_TOTAL));

        Ok(())
    }

    /// A failed family registration must not vanish: `register` returns None
    /// (unchanged) AND warns with the underlying registration error.
    #[test]
    fn register_warns_when_a_family_registration_fails() -> Result<(), Box<dyn std::error::Error>> {
        let registry = MetricsRegistry::new();
        // Pre-register the gauge's name as a COUNTER so the gauge registration
        // fails with an incompatible-kind error.
        registry.register_counter(CONNECTIONS_ACTIVE, no_labels())?;

        let log = crate::test_log::CapturedLog::default();
        let metrics = log.capture(|| ServerMetrics::register(&registry));

        assert!(
            metrics.is_none(),
            "a failed registration must still yield None"
        );
        let output = log.contents();
        assert!(
            output.contains("WARN") && output.contains(CONNECTIONS_ACTIVE),
            "expected a warn carrying the registration error, got: {output}"
        );

        Ok(())
    }
}