gunnar_core/

observe.rs

//! **What an operator can see**: the `tracing` targets, the request id, and
//! the server's counters.
//!
//! # The measurement this module answers
//!
//! 58 `tracing::` call sites in 27182 lines of code, no `span!`, no
//! `#[instrument]`, no request id, no timing and no counters. Closing the gRPC
//! audit hole moved the worst of that distribution and none of the total.
//! At forge scale an operator has minutes to answer *which connection, which
//! principal, which repository, how long did it take, how many failed, is the
//! queue growing*. None of it was answerable from gunnar's output at all.
//!
//! # One vocabulary, not one per crate
//!
//! [`TARGET_AUDIT`] already existed, in `gunnar_grpc::auth`. Everything added
//! since is a **sibling of it**, defined here once, so `RUST_LOG` selects a
//! *subsystem* rather than a crate name that happens to hold the call site:
//!
//! | target | what is on it |
//! |---|---|
//! | [`TARGET_AUDIT`] | control-plane mutations: who called what, and the decision |
//! | [`TARGET_SSH`] | the connection: accept, publickey, exec dispatch, every refusal |
//! | [`TARGET_GIT`] | transfers: upload-pack, receive-pack, objects, packs, bytes |
//! | [`TARGET_POLICY`] | the push gate: which gate refused which reference |
//! | [`TARGET_STORE`] | ref transactions, compare-and-swap contention, repack |
//! | [`TARGET_METRICS`] | the counters: the heartbeat, and the scrape endpoint |
//! | [`TARGET_LFS`] | large-file transfers: the session, its verbs, its refusals |
//! | [`TARGET_RUNNER`] | the build runner: what it picked up, what it built, its refusals |
//! | [`TARGET_ARCHIVE`] | archiving a repository off the box and back in |
//! | [`TARGET_REPLICATION`] | replication to a standby: shipped, lagging, refused |
//! | [`TARGET_SPACE`] | disk headroom: what a write may take, and the refusals when it is short |
//! | [`TARGET_REPAIR`] | the repair plane: the boot check, the pull channel, the recharges they run |
//! | [`TARGET_LIVE`] | the live update: a version staged, the re-exec, the trial and its roll back |
//!
//! ```text
//!   RUST_LOG=warn,gunnar::audit=info,gunnar::git=info   # the two an operator watches
//!   RUST_LOG=gunnar::ssh=debug                          # one connection, in detail
//! ```
//!
//! `gunnar-grpc` cannot link this crate (it is expressed in `gunnar-proto`
//! terms and deliberately has no `gunnar-core` edge), so it keeps its own
//! `AUDIT_TARGET` **string**. That is the one place a constant is stated twice,
//! and it is not left to a comment: `gunnar-cli` links both crates and
//! `the_audit_target_is_the_same_string_in_both_crates` asserts they are equal.
//!
//! # Levels
//!
//! `info` is the production default and must stay legible: one line per
//! connection authenticated, one per transfer served with its applied outputs,
//! one per reference moved, and every refusal at `warn`. `debug` is the
//! diagnostic tier: per-frame detail, environment requests, CAS retries. A
//! server that has to run at `trace` to be legible has failed.
//!
//! # Cost
//!
//! [`RequestId`] is a `u64` that formats itself; it never allocates and never
//! touches a lock. [`ServerMetrics`] is a flat block of [`AtomicU64`] read and
//! written with [`Ordering::Relaxed`]. **No mutex is taken to record a
//! counter**, because a signal that costs a lock is paid for by the thing it is
//! measuring. `tracing`'s macros evaluate their field expressions only inside
//! the `enabled!` branch, so a line that will not be emitted allocates nothing
//! even where the field is a `String`.

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::OnceLock;

// ───────────────────────────────────────────────────────────────────────────
// Targets
// ───────────────────────────────────────────────────────────────────────────

/// Control-plane mutations: who called what, over which credential, and what
/// was decided. Written by `gunnar-grpc`, which states the same literal
/// because it cannot link this crate. See the module docs.
pub const TARGET_AUDIT: &str = "gunnar::audit";

/// The SSH connection: accepted, authenticated, dispatched, refused.
pub const TARGET_SSH: &str = "gunnar::ssh";

/// Git transfers: `upload-pack`, `receive-pack`, and their applied outputs.
pub const TARGET_GIT: &str = "gunnar::git";

/// The push gate: which gate refused which reference, and why.
pub const TARGET_POLICY: &str = "gunnar::policy";

/// Storage: ref transactions, compare-and-swap contention, repack.
pub const TARGET_STORE: &str = "gunnar::store";

/// The counter heartbeat, and the scrape endpoint.
pub const TARGET_METRICS: &str = "gunnar::metrics";

/// Large-file transfers: the LFS session, its verbs, and every refusal.
///
/// Added 2026-08-05, and it did not replace nothing. `gunnar-lfs` was emitting
/// on **`"gunnar.lfs"`** — a dot, not a `::` — at five call sites, and
/// the old znippy engine crate, `gunnar-cli`'s znippy-engine arm and `gunnar-znippy` were
/// emitting on `"gunnar.store"` at twelve more. `RUST_LOG=gunnar::lfs=debug`
/// selected none of them; neither did `gunnar::store=info`, which this
/// module's own table tells an operator to use. Seventeen lines that could be
/// written and never read.
///
/// That is exactly the failure [`TARGETS`] was written to prevent, and
/// [`TARGETS`] could not see it: it checked the constants in its own list and
/// no call site ever passes through it. The guard that can is
/// `every_tracing_target_in_the_shipped_source_is_one_of_ours` in `gunnar-cli`'s
/// `tests/observability.rs`, which reads the source of every shipped crate.
pub const TARGET_LFS: &str = "gunnar::lfs";

/// The build runner: what it picked up, what it built, and every refusal.
///
/// Added with the runner (roadmap phase 3). A separate target from
/// [`TARGET_GIT`] because the two answer different questions — "did the push
/// land" and "did the build run" — and an operator watching one rarely wants a
/// wall of the other.
pub const TARGET_RUNNER: &str = "gunnar::runner";

/// Archiving a repository out of the box and back in (`Repos.ArchiveOut`).
///
/// `repo.archive_out` when a repository is written to its archive,
/// `repo.archive_out_failed` when that write does not land.
///
/// Emitted by `gunnar-grpc`'s `service::repos` and undeclared until the two
/// halves of this repository were merged (2026-08-31) — so
/// `gunnar_core::observe::log_filter`, the filter a deployment installs,
/// selected neither of its two call sites. The same defect [`TARGET_LFS`]
/// documents, found by the guard that exists for it.
///
/// Undeclared is not harmless here: an archive-out that FAILS is the one event
/// an operator has to see, so the failure of the very thing that protects the
/// objects was the least visible line gunnar emitted.
pub const TARGET_ARCHIVE: &str = "gunnar::archive";

/// Replication to a standby: what was shipped, what lagged, what
/// refused.
///
/// `replication.never_synced`, `replication.stale`, `replication.recovered`.
///
/// Emitted by `gunnar-server`'s `replication_view` and undeclared until
/// 2026-08-31, alongside [`TARGET_ARCHIVE`] and for the same reason. It is the
/// worst of the three to have had dark: a standby that receives nothing looks
/// exactly like a standby with nothing to receive, and the lines that tell them
/// apart were written into a filter that did not select them — so a mirror
/// could quietly stop being a mirror.
pub const TARGET_REPLICATION: &str = "gunnar::replication";
/// The process's own limits: the descriptor budget PID 1 raised, and the WARN
/// row when it could not raise it far enough.
///
/// Added 2026-08-30 for a defect of exactly the shape [`TARGET_LFS`]
/// documents: `crate::limits` shipped emitting on `"gunnar::limits"`
/// at two call sites, and no documented filter selected either of them. One of
/// the two is the `!WARN limits.nofile.low` row the dashboard guide tells an
/// operator to watch for.
pub const TARGET_LIMITS: &str = "gunnar::limits";

/// The object index: a reader's own scan of a pending pack, and the memory
/// watchdog's pause and spill.
///
/// Emitted by the znippy git plugin, which cannot link this crate and states
/// the literal, and rendered by the dashboard as `index paused: …` /
/// `index.spilled`. Undeclared until 2026-08-30, so
/// `gunnar_core::observe::log_filter` — the filter a deployment installs —
/// selected none of it, which is exactly the memory-pressure story an operator
/// needs when a box collapses under sustained create load.
pub const TARGET_INDEX: &str = "gunnar::index";

/// Plan enforcement (the `metered` build): a fact accepted or refused at
/// `Entitlement.Set`, a push refused for a suspended namespace, a cap applied.
/// Emitted by `gunnar-server::entitlement`; absent from a community binary,
/// declared here regardless so the filter and the probes name it.
pub const TARGET_ENTITLEMENT: &str = "gunnar::entitlement";
/// The twin following a primary's transaction log (`gunnar-server::twin`).
pub const TARGET_TWIN: &str = "gunnar::twin";
/// Disk space: the headroom a write is admitted against, the space a push may
/// take, and the refusals when the volume is short (`gunnar-ssh::space`,
/// `gunnar-ssh::headroom`). Emitted since 5.x but undeclared until 2026-10-10
/// (stinky row 22), so the installed filter selected none of it.
pub const TARGET_SPACE: &str = "gunnar::space";

/// The repair plane: the boot check's findings, the pull channel's word, and
/// the recharges they run (`gunnar-server::serve`, 5.4 resilience).
pub const TARGET_REPAIR: &str = "gunnar::repair";


/// The live update: a version staged, the graceful stop and re-exec, the trial
/// and a roll back.
///
/// Emitted by `gunnar-server::live_update` since the live update landed
/// (2026-10-12) and undeclared until 2026-10-14 — so the filter a deployment
/// installs selected none of its call sites. The defect [`TARGET_LFS`]
/// documents, found by the guard that exists for it.
pub const TARGET_LIVE: &str = "gunnar::live";

/// Every target gunnar emits on, so a filter, a document or a test can
/// enumerate them rather than guess.
///
/// A test asserts this list is exactly the constants above and that each is a
/// `gunnar::` sibling. A target invented at one call site and spelled slightly
/// differently is invisible to every filter written for the others, which is
/// the failure this constant exists to prevent.
pub const TARGETS: &[&str] = &[
    TARGET_AUDIT,
    TARGET_SSH,
    TARGET_GIT,
    TARGET_POLICY,
    TARGET_STORE,
    TARGET_METRICS,
    TARGET_LFS,
    TARGET_RUNNER,
    TARGET_ARCHIVE,
    TARGET_REPLICATION,
    TARGET_LIMITS,
    TARGET_INDEX,
    TARGET_ENTITLEMENT,
    TARGET_TWIN,
    TARGET_SPACE,
    TARGET_REPAIR,
    TARGET_LIVE,
];

/// A `RUST_LOG` directive list: `base` for everything gunnar did not write, and
/// `gunnar` for every one of [`TARGETS`].
///
/// **Built FROM [`TARGETS`], never written out beside it.** The table in this
/// module's docs is prose and a reader can copy it wrong; a filter that a
/// deployment installs must not be able to fall behind the list (one implementation). The
/// 2026-08-05 `"gunnar.lfs"` finding is what a hand-written filter costs:
/// seventeen call sites no documented filter selected, and nothing failed.
///
/// The two levels are separate arguments because they answer different
/// questions. `base` is how much `russh`, `h2` and `gix` an operator wants to
/// wade through; `gunnar` is how much of gunnar's own story they want. A single
/// level cannot say "everything gunnar does, and nothing else at that volume".
pub fn log_filter(base: &str, gunnar: &str) -> String {
    let mut spec = String::with_capacity(base.len() + TARGETS.len() * 24);
    spec.push_str(base);
    for target in TARGETS {
        spec.push(',');
        spec.push_str(target);
        spec.push('=');
        spec.push_str(gunnar);
    }
    spec
}

/// Everything outside gunnar, in the filter a service is installed with.
///
/// `info` and not `warn`: the lines that say a listener bound, a host key was
/// loaded and a runtime started come from gunnar at `info` too, and an operator
/// reading a journal from the top wants the boot to be legible.
pub const SERVICE_LOG_BASE: &str = "info";

/// gunnar's own targets, in the filter a service is installed with.
///
/// `debug` rather than `info`, deliberately. A service is watched when
/// something is being commissioned or has gone wrong, and the rungs above
/// `info` in gunnar are per-frame protocol detail on gunnar's OWN targets only
/// — `base` stays at `info`, so this does not drag `russh`, `h2` and `gix`
/// debug into the journal with it. Turn it down with `--log-filter`, or move it
/// live with `SIGUSR1` (`gunnar_cli::logging`).
pub const SERVICE_LOG_GUNNAR: &str = "debug";

/// **The `RUST_LOG` a `gunnar service install` bakes into the unit.**
///
/// One function, so the unit, the documentation and the test that proves the
/// filter selects gunnar's lines all read the same string.
pub fn service_log_filter() -> String {
    log_filter(SERVICE_LOG_BASE, SERVICE_LOG_GUNNAR)
}

// ───────────────────────────────────────────────────────────────────────────
// The request id
// ───────────────────────────────────────────────────────────────────────────

/// A per-connection (or per-request) identifier, carried as a span field so
/// every line emitted underneath is attributable without repeating it.
///
/// # Why not a UUID
///
/// A UUID costs a dependency, 16 bytes and a formatting allocation to say
/// something a 64-bit integer already says. This is a **process seed** in the
/// high 32 bits and a monotonic counter in the low 32, rendered as 16 hex
/// digits. Two ids from the same process never collide; two ids from different
/// runs of the same server collide only if both start within the same
/// microsecond *and* reach the same counter value, which is not a property a
/// log correlation needs.
///
/// It is `Copy`, it is one `fetch_add(Relaxed)` to mint, and its `Display`
/// writes into the formatter without allocating.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct RequestId(u64);

static NEXT_REQUEST_ID: AtomicU64 = AtomicU64::new(1);

/// The high half: distinct per process, so two servers' logs merged into one
/// stream do not appear to share connections.
fn process_seed() -> u64 {
    static SEED: OnceLock<u64> = OnceLock::new();
    *SEED.get_or_init(|| {
        let nanos = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos() as u64)
            .unwrap_or(0);
        // Mix the pid in so two processes started in the same nanosecond bucket
        // still differ. `id()` is a `u32` and free.
        (nanos ^ ((std::process::id() as u64) << 40)) & 0xffff_ffff
    })
}

impl RequestId {
    /// Mint the next id. One relaxed atomic increment; never blocks.
    pub fn next() -> RequestId {
        let n = NEXT_REQUEST_ID.fetch_add(1, Ordering::Relaxed);
        RequestId((process_seed() << 32) | (n & 0xffff_ffff))
    }

    /// The raw value, for a caller that wants to store it as a number.
    pub fn get(&self) -> u64 {
        self.0
    }
}

impl std::fmt::Display for RequestId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{:016x}", self.0)
    }
}

// ───────────────────────────────────────────────────────────────────────────
// The counters
// ───────────────────────────────────────────────────────────────────────────

/// Everything an operator watches, as one flat block of atomics.
///
/// Held behind an `Arc` and shared by the SSH listener and the git transport,
/// which is what makes "connections refused" and "pushes rejected" numbers
/// about the **same server** rather than two crates' private tallies.
///
/// Every method is one `fetch_add(Relaxed)`. There is no lock anywhere in this
/// type, and [`ServerMetrics::snapshot`] reads the fields independently rather
/// than under a barrier: these are a signal, and a signal read atomically
/// across eighteen fields would cost more than it tells anyone.
#[derive(Debug, Default)]
pub struct ServerMetrics {
    connections_opened: AtomicU64,
    connections_accepted: AtomicU64,
    connections_anonymous: AtomicU64,
    connections_refused: AtomicU64,
    auth_faults: AtomicU64,
    exec_refused: AtomicU64,
    transfers_started: AtomicU64,
    transfers_in_flight: AtomicU64,
    transfers_failed: AtomicU64,
    packs_served: AtomicU64,
    objects_sent: AtomicU64,
    bytes_out: AtomicU64,
    bytes_in: AtomicU64,
    pushes_accepted: AtomicU64,
    pushes_rejected: AtomicU64,
    refs_updated: AtomicU64,
    refs_deleted: AtomicU64,
    refs_rejected: AtomicU64,
    gate_refusals: AtomicU64,
}

/// A reading of [`ServerMetrics`], taken at one moment.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct MetricsSnapshot {
    /// TCP connections russh handed to gunnar.
    pub connections_opened: u64,
    /// Connections whose offered key named a principal.
    pub connections_accepted: u64,
    /// Connections admitted with **no key at all**, served as the
    /// anonymous principal, which can only ever read a public repository.
    ///
    /// Counted separately from [`MetricsSnapshot::connections_accepted`] rather
    /// than folded into it: "how much of my traffic is strangers" is the first
    /// question an operator asks after making anything public, and a total that
    /// mixed the two could not answer it.
    pub connections_anonymous: u64,
    /// Connections whose offered key named nobody. **Not** the same as
    /// [`MetricsSnapshot::auth_faults`].
    pub connections_refused: u64,
    /// Connections refused because the identity table could not be *read*. A
    /// storage fault looks exactly like an empty table from the client's side,
    /// and an operator must be able to tell them apart.
    pub auth_faults: u64,
    /// Exec requests refused or denied after authentication: a non-git
    /// command, an unknown repository, a missing grant.
    pub exec_refused: u64,
    /// Transfers admitted and begun.
    pub transfers_started: u64,
    /// Transfers running right now. A **gauge**, not a total.
    pub transfers_in_flight: u64,
    /// Transfers that ended with a protocol or storage error.
    pub transfers_failed: u64,
    /// `upload-pack` sessions that sent a packfile.
    pub packs_served: u64,
    /// Objects written into those packs.
    pub objects_sent: u64,
    /// Bytes gunnar wrote to git clients, counted at the wire.
    pub bytes_out: u64,
    /// Bytes gunnar read from git clients, counted at the wire.
    pub bytes_in: u64,
    /// Pushes in which every reference landed.
    pub pushes_accepted: u64,
    /// Pushes in which at least one reference was refused.
    pub pushes_rejected: u64,
    /// References moved.
    pub refs_updated: u64,
    /// References deleted.
    pub refs_deleted: u64,
    /// Reference updates refused, for any reason.
    pub refs_rejected: u64,
    /// Reference updates refused **by a named policy gate**. A subset of
    /// [`MetricsSnapshot::refs_rejected`], which also counts connectivity
    /// failures and the collateral refusals of an `--atomic` push.
    pub gate_refusals: u64,
}

// **Every field above is written by real code.** A counter nothing increments
// reads as "this never happens" and is worse than no counter at all. `repacks`
// was in this struct for exactly one revision before it came out again, because
// `gunnar gc` runs in a separate process that has no `ServerMetrics` to write
// to. `a_real_clone_and_push_move_the_counters` in `gunnar-cli` pins the ones a
// transfer touches; the rest have a red-first proof in their own crate.

/// What a number **means to a reader**: a total that only climbs, or a level
/// that moves both ways.
///
/// One field in this type is a gauge — [`MetricsSnapshot::transfers_in_flight`]
/// — and it is the whole reason the distinction is carried rather than assumed.
/// A scraper told everything is a counter charts a busy server's concurrency as
/// a *rate of transfers per second*, which is a different number with a
/// plausible shape, and the operator reading it has no way to know.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MetricKind {
    /// Monotonic since start-up. A reader takes its rate.
    Counter,
    /// A level, read as it stands.
    Gauge,
}

impl MetricKind {
    /// The word the exposition format's `# TYPE` line uses.
    pub fn as_str(self) -> &'static str {
        match self {
            MetricKind::Counter => "counter",
            MetricKind::Gauge => "gauge",
        }
    }
}

/// The kind word in the [`counters!`] table, as a value.
///
/// Two arms and no fallback, deliberately: a counter declared `guage` — or
/// declared with no kind at all — **fails to compile**. The alternative is a
/// default, and a default here would silently publish the next gauge as a
/// counter, which is the one error in this file that produces a plausible
/// wrong number instead of an obvious one.
macro_rules! metric_kind {
    (counter) => {
        MetricKind::Counter
    };
    (gauge) => {
        MetricKind::Gauge
    };
}

macro_rules! counters {
    ($($kind:ident $field:ident => $add:ident),* $(,)?) => {
        impl ServerMetrics {
            $(
                #[doc = concat!("Add `n` to `", stringify!($field), "`.")]
                pub fn $add(&self, n: u64) {
                    self.$field.fetch_add(n, Ordering::Relaxed);
                }
            )*

            /// Read every counter.
            pub fn snapshot(&self) -> MetricsSnapshot {
                MetricsSnapshot {
                    $($field: self.$field.load(Ordering::Relaxed),)*
                }
            }
        }

        impl MetricsSnapshot {
            /// Every counter as `(name, kind, value)`, in declaration order.
            ///
            /// Generated by the same macro arm that declares the fields, so a
            /// counter added to [`ServerMetrics`] appears here — and therefore
            /// on the scrape endpoint — **without anyone remembering to add
            /// it**. The alternative was a second hand-written list of nineteen
            /// names, and the counter left out of it would be the one nobody
            /// could see.
            ///
            /// That is not a hypothetical. The heartbeat *was* such a list, and
            /// it carried eighteen of the nineteen: `connections_anonymous` —
            /// written by both transports, and the answer to *how much of my
            /// traffic is strangers* — never appeared in a single line gunnar
            /// ever printed. See [`MetricsSnapshot::emit_heartbeat`].
            pub fn fields(&self) -> Vec<(&'static str, MetricKind, u64)> {
                vec![$((stringify!($field), metric_kind!($kind), self.$field),)*]
            }

            /// The heartbeat line: one `info` on [`TARGET_METRICS`] naming every
            /// counter.
            ///
            /// Here, and generated, for the reason above: this used to be a
            /// hand-written `tracing::info!` in `gunnar-cli` listing eighteen of
            /// the nineteen fields. Expanding from the declaration is the only
            /// version of it that cannot fall behind (one implementation), and it puts the
            /// log line and [`MetricsSnapshot::fields`] — the log and the wire —
            /// on one source of truth rather than two that agree today.
            ///
            /// Structured fields and not a rendered string, unchanged from what
            /// it replaced: `bytes_out=` has to stay greppable in the server's
            /// stderr, and a JSON subscriber has to still see nineteen typed
            /// numbers rather than one blob.
            pub fn emit_heartbeat(&self) {
                tracing::info!(
                    target: TARGET_METRICS,
                    event = "metrics",
                    $($field = self.$field,)*
                    "gunnar"
                );
            }
        }
    };
}

counters! {
    counter connections_opened => add_connections_opened,
    counter connections_accepted => add_connections_accepted,
    counter connections_anonymous => add_connections_anonymous,
    counter connections_refused => add_connections_refused,
    counter auth_faults => add_auth_faults,
    counter exec_refused => add_exec_refused,
    counter transfers_started => add_transfers_started,
    gauge   transfers_in_flight => add_transfers_in_flight,
    counter transfers_failed => add_transfers_failed,
    counter packs_served => add_packs_served,
    counter objects_sent => add_objects_sent,
    counter bytes_out => add_bytes_out,
    counter bytes_in => add_bytes_in,
    counter pushes_accepted => add_pushes_accepted,
    counter pushes_rejected => add_pushes_rejected,
    counter refs_updated => add_refs_updated,
    counter refs_deleted => add_refs_deleted,
    counter refs_rejected => add_refs_rejected,
    counter gate_refusals => add_gate_refusals,
}

impl ServerMetrics {
    /// A fresh set of counters, all zero.
    pub fn new() -> ServerMetrics {
        ServerMetrics::default()
    }

    /// A transfer began. Returns a guard whose [`Drop`] decrements
    /// `transfers_in_flight`, so the gauge cannot be leaked by an error path
    /// that forgot to close it, the failure mode that turns a working gauge
    /// into a server that appears permanently busy.
    pub fn transfer_started(self: &std::sync::Arc<Self>) -> InFlightGuard {
        self.transfers_started.fetch_add(1, Ordering::Relaxed);
        self.transfers_in_flight.fetch_add(1, Ordering::Relaxed);
        InFlightGuard {
            metrics: std::sync::Arc::clone(self),
        }
    }
}

/// A held `transfers_in_flight` slot. Released by [`Drop`], never by a call.
#[derive(Debug)]
pub struct InFlightGuard {
    metrics: std::sync::Arc<ServerMetrics>,
}

impl Drop for InFlightGuard {
    fn drop(&mut self) {
        self.metrics
            .transfers_in_flight
            .fetch_sub(1, Ordering::Relaxed);
    }
}

impl MetricsSnapshot {
    /// Has anything at all been served? Used by the heartbeat to stay quiet on
    /// an idle server without going quiet on a busy one.
    pub fn is_zero(&self) -> bool {
        *self == MetricsSnapshot::default()
    }
}

// ───────────────────────────────────────────────────────────────────────────
// The scrape body
// ───────────────────────────────────────────────────────────────────────────

/// What every exposed name starts with, so one `gunnar_` selector reaches all
/// of them and nothing gunnar publishes can collide with a neighbour's series.
pub const SCRAPE_PREFIX: &str = "gunnar_";

/// The `Content-Type` a Prometheus-compatible scraper expects. Stated once
/// here, beside the renderer that decides the format, rather than in the
/// listener that merely writes the header.
pub const SCRAPE_CONTENT_TYPE: &str = "text/plain; version=0.0.4; charset=utf-8";

impl MetricsSnapshot {
    /// This reading in the Prometheus text exposition format.
    ///
    /// ```text
    /// # TYPE gunnar_bytes_out counter
    /// gunnar_bytes_out 4096
    /// # TYPE gunnar_transfers_in_flight gauge
    /// gunnar_transfers_in_flight 1
    /// ```
    ///
    /// # Why the names are not suffixed `_total`
    ///
    /// Prometheus convention would call the first one `gunnar_bytes_out_total`.
    /// It is a convention and not a requirement — `# TYPE` is what actually
    /// carries the semantics, and this renderer emits it for every series — and
    /// paying it would mean **one counter with two names**: `bytes_out=` in the
    /// log and `gunnar_bytes_out_total` on the wire. The module's opening
    /// section argues for one vocabulary across the whole server, and an
    /// operator who greps the heartbeat and then greps the endpoint for the
    /// same string is exactly who that section is for. The prefix is worth its
    /// cost because it prevents collision; the suffix buys a lint.
    ///
    /// # Why there is no `# HELP`
    ///
    /// Every one of these fields is documented on [`MetricsSnapshot`], and a
    /// `HELP` line would be a second copy of that prose, in a macro table,
    /// diverging from the first (one implementation). `# TYPE` is the part a scraper reads.
    ///
    /// # Why this is here and not in the listener
    ///
    /// It is pure — a `&self` to a `String`, no clock, no I/O, no socket — so
    /// the format can be asserted without binding a port, and the listener that
    /// serves it stays a listener. It also means `gunnar-core` gained no
    /// dependency for the endpoint: the HTTP is somebody else's problem.
    pub fn render_prometheus(&self) -> String {
        let mut out = String::with_capacity(self.fields().len() * 64);
        for (name, kind, value) in self.fields() {
            out.push_str("# TYPE ");
            out.push_str(SCRAPE_PREFIX);
            out.push_str(name);
            out.push(' ');
            out.push_str(kind.as_str());
            out.push('\n');
            out.push_str(SCRAPE_PREFIX);
            out.push_str(name);
            out.push(' ');
            {
                use std::fmt::Write as _;
                // Writing into the buffer rather than `format!`-ing nineteen
                // little `String`s and dropping each one. `write!` to a
                // `String` is infallible; the `Result` is discarded for that
                // reason and not because a failure would be tolerable.
                let _ = write!(out, "{value}");
            }
            out.push('\n');
        }
        out
    }
}

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

    #[test]
    fn every_target_is_a_gunnar_sibling_and_the_list_is_complete() {
        // The scheme, asserted rather than described: one prefix, one `::`
        // separator, no crate names. A target invented at a call site and
        // spelled `gunnar_ssh` would be invisible to every filter written for
        // the others.
        for target in TARGETS {
            assert!(
                target.starts_with("gunnar::"),
                "{target} is not a sibling of {TARGET_AUDIT}"
            );
            assert_eq!(
                target.matches("::").count(),
                1,
                "{target} is not one level deep"
            );
        }
        let mut sorted = TARGETS.to_vec();
        sorted.sort_unstable();
        sorted.dedup();
        assert_eq!(sorted.len(), TARGETS.len(), "TARGETS holds a duplicate");
        assert!(TARGETS.contains(&TARGET_AUDIT));
    }

    /// The filter must name **every** target, and it must be derived rather
    /// than restated — that is the whole reason it is a function.
    ///
    /// The identity-value guard is the second half: a filter of `"trace"` would
    /// select every line in this process and satisfy any test that only asked
    /// "does gunnar's output appear". So this asserts the filter is a LIST that
    /// names each target with the gunnar level, and that the base level is
    /// still in it and is NOT that level. `tests/observability.rs` in
    /// `gunnar-cli` then asserts the string actually decides events that way.
    #[test]
    fn the_service_filter_is_derived_from_the_target_list() {
        let filter = service_log_filter();
        for target in TARGETS {
            let directive = format!("{target}={SERVICE_LOG_GUNNAR}");
            assert!(
                filter.contains(&directive),
                "{directive} is missing from {filter}: a target added to TARGETS did not \
                 reach the filter a service is installed with"
            );
        }
        assert!(
            filter.starts_with(SERVICE_LOG_BASE),
            "the filter must open with the base level: {filter}"
        );
        assert_ne!(
            SERVICE_LOG_BASE, SERVICE_LOG_GUNNAR,
            "with one level for both, this filter says nothing a bare level does not, and \
             the test above cannot tell a derived filter from a blanket one"
        );
        assert_eq!(
            filter.matches(',').count(),
            TARGETS.len(),
            "the filter holds a directive that is not one of ours: {filter}"
        );
    }

    #[test]
    fn request_ids_do_not_repeat() {
        let a = RequestId::next();
        let b = RequestId::next();
        assert_ne!(a, b, "two ids from the same process collided");
        assert_ne!(
            a.to_string(),
            b.to_string(),
            "two ids RENDERED the same, which is what a log correlation sees"
        );
        assert_eq!(a.to_string().len(), 16, "the id must be fixed-width");
        assert!(
            a.to_string().chars().all(|c| c.is_ascii_hexdigit()),
            "the id must be greppable"
        );
        // …and the seed half is stable within a process, so a human can spot
        // two connections belonging to the same server at a glance.
        assert_eq!(a.get() >> 32, b.get() >> 32);
    }

    #[test]
    fn a_counter_moves_and_the_gauge_comes_back_down() {
        let metrics = Arc::new(ServerMetrics::new());
        let before = metrics.snapshot();
        assert!(before.is_zero(), "a fresh set of counters must read zero");

        metrics.add_bytes_out(4096);
        metrics.add_objects_sent(7);
        let guard = metrics.transfer_started();
        let during = metrics.snapshot();
        assert_eq!(during.bytes_out - before.bytes_out, 4096);
        assert_eq!(during.objects_sent - before.objects_sent, 7);
        assert_eq!(during.transfers_started, 1);
        assert_eq!(during.transfers_in_flight, 1, "the gauge must be up");

        drop(guard);
        let after = metrics.snapshot();
        assert_eq!(
            after.transfers_in_flight, 0,
            "the in-flight gauge leaked; a server that never comes back to zero \
             reads as permanently saturated"
        );
        assert_eq!(
            after.transfers_started, 1,
            "the total is a total, not a gauge"
        );
    }

    /// The guard that would have caught `connections_anonymous`.
    ///
    /// It names its own subject: **every field of [`MetricsSnapshot`] must be
    /// reachable through [`MetricsSnapshot::fields`]**, counted against the
    /// struct itself and not against a list this test also keeps. A second list
    /// here would be the same defect one layer up — the one the endpoint would
    /// then be missing a counter from, silently.
    ///
    /// It is counted from `Debug`, which the compiler derives from the struct
    /// definition, so adding a field to `MetricsSnapshot` and forgetting the
    /// `counters!` table makes this fail without anyone touching this file.
    #[test]
    fn every_field_of_the_snapshot_is_published() {
        let snapshot = MetricsSnapshot::default();
        let published: Vec<&str> = snapshot.fields().iter().map(|(name, _, _)| *name).collect();

        // `MetricsSnapshot { connections_opened: 0, connections_accepted: 0, … }`
        let debug = format!("{snapshot:?}");
        let declared: Vec<String> = debug
            .trim_start_matches("MetricsSnapshot {")
            .trim_end_matches('}')
            .split(',')
            .filter_map(|pair| pair.split_once(':'))
            .map(|(name, _)| name.trim().to_owned())
            .collect();
        assert_eq!(
            declared.len(),
            19,
            "the Debug parse found {} fields, not the 19 this struct has, so this guard is \
             reading nothing and would pass against an empty `fields()`: {debug}",
            declared.len()
        );

        for field in &declared {
            assert!(
                published.contains(&field.as_str()),
                "`{field}` is a counter of this server that NOTHING publishes: not the \
                 heartbeat, not the scrape endpoint. That is exactly how \
                 `connections_anonymous` stayed invisible."
            );
        }
        assert_eq!(
            published.len(),
            declared.len(),
            "fields() publishes {} names for {} struct fields",
            published.len(),
            declared.len()
        );
    }

    /// The scrape body, asserted as **text a scraper parses**, not as a call
    /// that returned.
    #[test]
    fn the_scrape_body_names_every_counter_with_its_kind_and_value() {
        let metrics = Arc::new(ServerMetrics::new());
        metrics.add_bytes_out(4096);
        metrics.add_connections_anonymous(3);
        let _guard = metrics.transfer_started();
        let body = metrics.snapshot().render_prometheus();

        // The value, not merely the name: a renderer that printed every series
        // as 0 would satisfy a name-only assertion, and 0 is what a broken
        // counter also looks like.
        assert!(
            body.contains("\ngunnar_bytes_out 4096\n"),
            "the applied value is not in the body:\n{body}"
        );
        assert!(
            body.contains("\ngunnar_connections_anonymous 3\n"),
            "the counter no line ever carried is still not carried:\n{body}"
        );
        // Kind, and the one that matters: a scraper told the gauge is a counter
        // charts concurrency as a rate.
        assert!(
            body.contains("# TYPE gunnar_transfers_in_flight gauge\n"),
            "the in-flight gauge is not typed as a gauge:\n{body}"
        );
        assert!(
            body.contains("# TYPE gunnar_bytes_out counter\n"),
            "a total is not typed as a counter:\n{body}"
        );
        // Every series is prefixed and every series is typed, checked over the
        // whole body rather than over the four names above.
        let mut samples = 0;
        for line in body.lines() {
            if let Some(spec) = line.strip_prefix("# TYPE ") {
                let (name, kind) = spec.split_once(' ').expect("a TYPE line is `name kind`");
                assert!(name.starts_with(SCRAPE_PREFIX), "{name} is unprefixed");
                assert!(kind == "counter" || kind == "gauge", "{kind} is not a kind");
                continue;
            }
            assert!(line.starts_with(SCRAPE_PREFIX), "stray line: {line:?}");
            let (name, value) = line.split_once(' ').expect("a sample is `name value`");
            value
                .parse::<u64>()
                .unwrap_or_else(|_| panic!("{name} rendered {value:?}, which is not a number"));
            assert!(
                body.contains(&format!("# TYPE {name} ")),
                "{name} is a sample with no TYPE line"
            );
            samples += 1;
        }
        assert_eq!(
            samples,
            metrics.snapshot().fields().len(),
            "the body carries {samples} samples for {} counters",
            metrics.snapshot().fields().len()
        );
    }

    #[test]
    fn counters_survive_concurrent_writers() {
        // Relaxed ordering is still atomic. If any of this were a `+= 1` on a
        // plain integer the total would come out short.
        let metrics = Arc::new(ServerMetrics::new());
        let mut threads = Vec::new();
        for _ in 0..8 {
            let metrics = Arc::clone(&metrics);
            threads.push(std::thread::spawn(move || {
                for _ in 0..1000 {
                    metrics.add_bytes_out(1);
                    let _guard = metrics.transfer_started();
                }
            }));
        }
        for t in threads {
            t.join().expect("a counting thread");
        }
        let snapshot = metrics.snapshot();
        assert_eq!(snapshot.bytes_out, 8000);
        assert_eq!(snapshot.transfers_started, 8000);
        assert_eq!(snapshot.transfers_in_flight, 0);
    }
}