beck_rt/

telemetry.rs

//! Telemetry — and the question of what it is *for*, in an architecture that already has a log.
//!
//! # Why this is not the usual answer
//!
//! Distributed tracing exists because in a system of services nobody knows what happened. You
//! reconstruct causality after the fact from correlated, sampled spans, and you accept that the
//! reconstruction is partial.
//!
//! Beck already has something strictly stronger for the part tracing usually covers: a durable
//! total order of every state transition. `state = fold(f, init, log[..seq])` is not a sample and
//! not a reconstruction — it is the actual history, and [`crate::replay_to`] will rebuild any state
//! the system was ever in. Tracing the fold's internal call tree as spans would re-record, lossily
//! and at cost, what the log records exactly and for free.
//!
//! So the division of labour is specific:
//!
//! | question | answered by |
//! |---|---|
//! | what happened, in what order, and what state did it produce | the log |
//! | what state was the system in at 14:02 | the log, by replay |
//! | why did this command produce that event | the log, by replay |
//! | how long did the fold take | **here** |
//! | how long did the append wait on Postgres | **here** |
//! | what was rejected, and never became an event | **here** |
//! | how many sessions are connected | **here** |
//! | what the maintained views cost, shared and per session | **here** |
//! | did the pod get killed mid-batch | **here** |
//!
//! Everything in the right column is either wall-clock, resource use, or a *non-event*: something
//! the log deliberately does not record, because §4.8 requires the fold to be replay-pure and a
//! fold that recorded its own duration would not replay identically. Telemetry is not a weaker
//! substitute for the log here; it is the complement of it, and the boundary between them is exactly
//! the boundary of determinism.
//!
//! # Correlation is `seq`, not a trace id
//!
//! A random trace id identifies a request. `seq` identifies a *state*: given one, `beck replay
//! --to <seq>` reproduces the system exactly as it was. So every record that has a sequence number
//! carries `beck.seq`, and a span in any OTel backend is one command away from a reproducible
//! debugging session. That is a property this architecture has and a microservice fleet does not.
//!
//! # Is OpenTelemetry valid here?
//!
//! Yes, for the right column, and this module speaks it: [`Telemetry::otlp_metrics`] and
//! [`Telemetry::otlp_logs`] produce OTLP/HTTP JSON, which is a first-class encoding in the OTLP
//! specification — same field names and semantics as the protobuf form, with no `tonic`, no
//! `prost`, and no code generation. Export is pull-only: the dashboard serves this data from
//! memory at `/_beck/otlp/metrics` and `/_beck/otlp/logs`, and nothing pushes to a collector
//! (a push exporter is scheduled, not built — `docs/101` §101.8).
//!
//! What Beck should *not* do is adopt OTel's model as its own. Spans belong at the boundaries —
//! ingress, validate, append, fold, view, patch — and not inside the fold, where the log is the
//! better instrument.
//!
//! # Cost
//!
//! Counters and histogram buckets are `AtomicU64`: recording is one relaxed fetch-add and no
//! allocation, so instrumenting the fold does not perturb what it measures. Histograms are fixed
//! power-of-two buckets, so the bucket index is a `leading_zeros`. The log ring is bounded and
//! overwrites oldest-first, so a process that runs for a month does not accumulate.

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

use serde_json::{json, Value as J};

/// Buckets covering 1 µs to ~1 s in powers of two, plus an overflow bucket.
const BUCKETS: usize = 21;

/// A histogram of durations in microseconds.
///
/// Power-of-two buckets so that recording is `leading_zeros` and a fetch-add: no locks, no
/// allocation, and no comparison chain. The bound this trades away is resolution — a value is
/// placed within a factor of two — which is the right trade for "is the fold suddenly slow", and
/// the wrong one for a billing meter. Nothing here is a billing meter.
#[derive(Debug)]
pub struct Histogram {
    buckets: [AtomicU64; BUCKETS],
    count: AtomicU64,
    sum_us: AtomicU64,
}

impl Default for Histogram {
    fn default() -> Self {
        Histogram {
            buckets: std::array::from_fn(|_| AtomicU64::new(0)),
            count: AtomicU64::new(0),
            sum_us: AtomicU64::new(0),
        }
    }
}

impl Histogram {
    pub fn record_us(&self, us: u64) {
        let i = if us == 0 {
            0
        } else {
            (64 - us.leading_zeros() as usize).min(BUCKETS - 1)
        };
        self.buckets[i].fetch_add(1, Ordering::Relaxed);
        self.count.fetch_add(1, Ordering::Relaxed);
        self.sum_us.fetch_add(us, Ordering::Relaxed);
    }

    pub fn record(&self, d: std::time::Duration) {
        self.record_us(d.as_micros() as u64);
    }

    pub fn count(&self) -> u64 {
        self.count.load(Ordering::Relaxed)
    }

    pub fn sum_us(&self) -> u64 {
        self.sum_us.load(Ordering::Relaxed)
    }

    pub fn mean_us(&self) -> f64 {
        let n = self.count();
        if n == 0 {
            0.0
        } else {
            self.sum_us() as f64 / n as f64
        }
    }

    /// The upper bound of each bucket, in microseconds — OTLP's `explicitBounds`.
    pub fn bounds() -> Vec<f64> {
        (0..BUCKETS - 1).map(|i| (1u64 << i) as f64).collect()
    }

    pub fn counts(&self) -> Vec<u64> {
        self.buckets
            .iter()
            .map(|b| b.load(Ordering::Relaxed))
            .collect()
    }

    /// The bucket at or below which `q` of the observations fall, as an upper bound in µs.
    ///
    /// A bucketed estimate, not an exact quantile: with power-of-two buckets the true value is
    /// within a factor of two of what this returns, and saying "p99 is at most 4 ms" is the honest
    /// form of the claim.
    pub fn quantile_us(&self, q: f64) -> u64 {
        let total = self.count();
        if total == 0 {
            return 0;
        }
        let target = (total as f64 * q).ceil() as u64;
        let mut seen = 0;
        for (i, b) in self.buckets.iter().enumerate() {
            seen += b.load(Ordering::Relaxed);
            if seen >= target {
                return 1u64 << i;
            }
        }
        1u64 << (BUCKETS - 1)
    }
}

#[derive(Debug, Default)]
pub struct Counter(AtomicU64);

impl Counter {
    pub fn incr(&self) {
        self.0.fetch_add(1, Ordering::Relaxed);
    }
    pub fn add(&self, n: u64) {
        self.0.fetch_add(n, Ordering::Relaxed);
    }
    /// Adopt a total that is counted somewhere else.
    ///
    /// For a counter whose authoritative value lives outside this struct: the shared dataflow
    /// counts its own releases, and this exports that number rather than counting the same events a
    /// second time. Still monotone, because the source is — but `incr`/`add` and this must not be
    /// mixed on one counter, or neither is right.
    pub fn sync(&self, total: u64) {
        self.0.store(total, Ordering::Relaxed);
    }
    pub fn get(&self) -> u64 {
        self.0.load(Ordering::Relaxed)
    }
}

/// A gauge — a value that goes up and down, like the number of connected sessions.
#[derive(Debug, Default)]
pub struct Gauge(AtomicU64);

impl Gauge {
    pub fn incr(&self) {
        self.0.fetch_add(1, Ordering::Relaxed);
    }
    pub fn decr(&self) {
        self.0.fetch_sub(1, Ordering::Relaxed);
    }
    pub fn set(&self, n: u64) {
        self.0.store(n, Ordering::Relaxed);
    }
    /// Replace one contributor's share: subtract what it held, add what it holds now.
    ///
    /// A gauge that aggregates over many things — the entries every connected subscription is
    /// arranging — cannot be `set` by any one of them, and re-summing them all on every render is
    /// the scan the number exists to avoid. Saturating, because a contributor that reports its
    /// departure twice must not wrap the gauge to `u64::MAX`.
    pub fn adjust(&self, was: u64, now: u64) {
        if now >= was {
            self.0.fetch_add(now - was, Ordering::Relaxed);
        } else {
            let d = was - now;
            let _ = self
                .0
                .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
                    Some(v.saturating_sub(d))
                });
        }
    }
    pub fn get(&self) -> u64 {
        self.0.load(Ordering::Relaxed)
    }
}

/// One log record, kept in memory for the dashboard and exported as an OTLP log record.
#[derive(Clone, Debug, serde::Serialize)]
pub struct Record {
    pub at_unix_nanos: u64,
    pub level: &'static str,
    pub target: String,
    pub message: String,
    /// The sequence number this record is about, when there is one. This is the correlation key:
    /// with it, any record points at a state `beck replay` can reproduce exactly.
    pub seq: Option<u64>,
}

/// The instruments.
///
/// One value, reached through [`telemetry`], because a metric registry that has to be threaded
/// through every call site gets threaded through some of them.
#[derive(Debug, Default)]
pub struct Telemetry {
    // --- what the log cannot tell you, because it is time ---
    /// How long one fold step took.
    pub fold: Histogram,
    /// How long rendering a view took. The dominant cost until Phase 3 (docs/19 §19.4 item 3).
    pub view: Histogram,
    /// How long the diff between two views took.
    pub diff: Histogram,
    /// How long an append to the log store took — the substrate's latency, not the program's.
    pub append: Histogram,
    /// How long a snapshot took.
    pub snapshot: Histogram,
    /// How long a cold replay took, and over how many events.
    pub replay: Histogram,

    // --- what the log cannot tell you, because it never happened ---
    /// Proposals rejected by `validate`. Rejections never become events, so nothing in the log
    /// records that anyone tried.
    pub rejected: Counter,
    /// Proposals dropped as duplicates by the idempotency key.
    pub deduplicated: Counter,
    /// Appends that failed. If this is non-zero, the log is missing something the program believed.
    pub append_failures: Counter,
    /// Snapshots that failed. Recoverable — the log is still the truth — but slower to recover.
    pub snapshot_failures: Counter,
    /// Client messages that did not parse.
    pub bad_messages: Counter,
    /// Routes taken on a live subscription — a `nav` frame that moved the client somewhere new.
    ///
    /// Counted because it is the one interaction whose cost differs between the two modes by
    /// construction: in Mode A it is a render and a patch on the server, and in Mode B it is a
    /// message the server answers with nothing at all.
    pub navigations: Counter,
    /// Connections refused because the identity did not verify.
    ///
    /// Counted separately from `rejected`, which is `validate` refusing a *command*: one is "who
    /// are you" and the other is "you may not do that", and an operator watching for an attack
    /// needs to tell them apart.
    pub unauthenticated: Counter,
    /// Proposals refused because the actor was over F3's write quota.
    ///
    /// A third counter beside `rejected` and `unauthenticated`, for the same reason those two are
    /// apart: "you may not do that", "who are you" and "not that often" are three different things
    /// for an operator watching an attack, and one number covering all three tells them nothing.
    pub throttled: Counter,

    // --- what the log does record, counted here so a rate is available without folding ---
    pub events_appended: Counter,
    pub patch_frames: Counter,
    pub patch_bytes: Counter,

    // --- what is true right now ---
    pub sessions: Gauge,

    /// Recent log records, oldest evicted first.
    ring: Mutex<VecDeque<Record>>,
    /// The `seq` the process has folded to. A gauge in spirit; stored so the dashboard has it
    /// without touching the app.
    pub head: Gauge,

    // --- what a fanout costs, which §5.3 names as a metric and nothing exported until now ---
    /// Arrangement entries held by the **one** shared dataflow: the operators that do not read the
    /// session, maintained once however many subscribers there are (docs/23).
    ///
    /// Entries rather than bytes. Bytes would need `Engine::footprint`, which walks the accumulator
    /// to charge shared structure to the fold — right for a report, far too expensive to sample on
    /// a live process. Entries are `O(operators)` to read and they are the number that scales.
    pub shared_arranged: Gauge,
    /// Arrangement entries held by connected subscriptions, between them.
    ///
    /// This is the one that multiplies by the fanout, and putting the two side by side is the whole
    /// operational question: `shared_arranged` is paid once, `session_arranged` is paid per
    /// connection. A program whose second number dwarfs the first has its cut in the wrong place.
    pub session_arranged: Gauge,
    /// Versions of change history the shared dataflow is keeping for subscribers that are behind.
    ///
    /// Bounded above by the configured retention depth and below by the laggiest connected
    /// subscriber, so on a healthy fanout it sits at 1 and a rising number means renders are not
    /// keeping up with events. That makes it a *lag* signal rather than a memory one, which is the
    /// more useful reading of the same number.
    pub shared_retained: Gauge,
    /// How many times the shared dataflow gave up its arrangements because nothing was subscribed.
    ///
    /// Each one is a cold start charged to whichever subscriber reconnects first. A process whose
    /// releases track its connection count is one whose clients are flapping, and is the case for
    /// turning `AppConfig::retention.release_when_idle` off.
    pub shared_releases: Counter,
}

/// How many log records to keep. Bounded so a long-running process does not accumulate; the log
/// store is the durable record, and this is a window onto the present.
const RING_CAPACITY: usize = 2_000;

impl Telemetry {
    pub fn log(&self, level: &'static str, target: &str, message: String, seq: Option<u64>) {
        let record = Record {
            at_unix_nanos: now_unix_nanos(),
            level,
            target: target.to_string(),
            message,
            seq,
        };
        let mut ring = self.ring.lock().expect("telemetry ring poisoned");
        if ring.len() == RING_CAPACITY {
            ring.pop_front();
        }
        ring.push_back(record);
    }

    /// The most recent records, newest first.
    pub fn records(&self, limit: usize) -> Vec<Record> {
        let ring = self.ring.lock().expect("telemetry ring poisoned");
        ring.iter().rev().take(limit).cloned().collect()
    }

    /// Everything, as the dashboard's JSON. Not OTLP: this is shaped for a table on a screen.
    pub fn snapshot(&self) -> J {
        let hist = |name: &str, h: &Histogram| {
            json!({
                "name": name,
                "count": h.count(),
                "mean_us": h.mean_us(),
                "p50_us": h.quantile_us(0.50),
                "p99_us": h.quantile_us(0.99),
            })
        };
        json!({
            "counters": {
                "events_appended": self.events_appended.get(),
                "rejected": self.rejected.get(),
                "unauthenticated": self.unauthenticated.get(),
                "throttled": self.throttled.get(),
                "deduplicated": self.deduplicated.get(),
                "append_failures": self.append_failures.get(),
                "snapshot_failures": self.snapshot_failures.get(),
                "bad_messages": self.bad_messages.get(),
                "navigations": self.navigations.get(),
                "patch_frames": self.patch_frames.get(),
                "patch_bytes": self.patch_bytes.get(),
                "shared_releases": self.shared_releases.get(),
            },
            "gauges": {
                "sessions": self.sessions.get(),
                "head": self.head.get(),
                "shared_arranged": self.shared_arranged.get(),
                "session_arranged": self.session_arranged.get(),
                "shared_retained": self.shared_retained.get(),
            },
            "histograms": [
                hist("fold", &self.fold),
                hist("view", &self.view),
                hist("diff", &self.diff),
                hist("append", &self.append),
                hist("snapshot", &self.snapshot),
                hist("replay", &self.replay),
            ],
        })
    }

    /// Every scalar this runtime records: the name, what it means, and whether it only rises.
    ///
    /// **One table, read by both exports.** The OTLP JSON and the OpenMetrics text are two
    /// spellings of the same numbers, and a metric added to one and not the other is the obvious
    /// way for them to drift. Adding a counter here adds it to both, which is a structure rather
    /// than a thing to remember.
    ///
    /// The names are OTLP's, dotted. [`Telemetry::openmetrics`] transliterates them, because a
    /// Prometheus name cannot hold a dot.
    fn scalars(&self) -> Vec<Scalar> {
        let counter = |name, help, value| Scalar {
            name,
            help,
            monotonic: true,
            value,
        };
        let gauge = |name, help, value| Scalar {
            name,
            help,
            monotonic: false,
            value,
        };
        vec![
            counter(
                "beck.events.appended",
                "events appended to the log",
                self.events_appended.get(),
            ),
            counter(
                "beck.proposals.rejected",
                "proposals `validate` refused",
                self.rejected.get(),
            ),
            counter(
                "beck.connections.unauthenticated",
                "connections that presented no usable identity",
                self.unauthenticated.get(),
            ),
            counter(
                "beck.proposals.deduplicated",
                "proposals already seen, by idempotency key",
                self.deduplicated.get(),
            ),
            counter(
                "beck.log.append.failures",
                "appends the store refused",
                self.append_failures.get(),
            ),
            counter(
                "beck.snapshot.failures",
                "snapshots the store refused",
                self.snapshot_failures.get(),
            ),
            counter(
                "beck.messages.malformed",
                "client messages that did not parse",
                self.bad_messages.get(),
            ),
            counter(
                "beck.navigations",
                "route changes a client reported",
                self.navigations.get(),
            ),
            counter(
                "beck.patch.frames",
                "patch frames sent to clients",
                self.patch_frames.get(),
            ),
            counter(
                "beck.patch.bytes",
                "bytes of patch payload sent to clients",
                self.patch_bytes.get(),
            ),
            counter(
                "beck.views.shared_releases",
                "shared arrangements released when their last reader left",
                self.shared_releases.get(),
            ),
            gauge(
                "beck.sessions.active",
                "sessions currently connected",
                self.sessions.get(),
            ),
            gauge("beck.log.head", "the log's head `seq`", self.head.get()),
            gauge(
                "beck.views.shared_arranged",
                "arrangements held once for every subscriber",
                self.shared_arranged.get(),
            ),
            gauge(
                "beck.views.session_arranged",
                "arrangements held per subscriber",
                self.session_arranged.get(),
            ),
            gauge(
                "beck.views.shared_retained",
                "readers holding a shared arrangement",
                self.shared_retained.get(),
            ),
        ]
    }

    /// Every duration this runtime records. The same table rule as [`Telemetry::scalars`].
    fn histograms(&self) -> Vec<(&'static str, &Histogram)> {
        vec![
            ("beck.fold.duration", &self.fold),
            ("beck.view.duration", &self.view),
            ("beck.diff.duration", &self.diff),
            ("beck.log.append.duration", &self.append),
            ("beck.snapshot.duration", &self.snapshot),
            ("beck.replay.duration", &self.replay),
        ]
    }

    /// OTLP/HTTP JSON for metrics — the body of a POST to `/v1/metrics`.
    ///
    /// Field names and the numeric enums (`aggregationTemporality: 2` is CUMULATIVE) are the
    /// specification's, so an ordinary collector accepts this without a Beck-specific receiver.
    pub fn otlp_metrics(&self, service: &str) -> J {
        // Start first: it is lazily initialised, so reading `now` first would make the very first
        // export report a start *after* the observation it bounds. A collector is entitled to drop
        // a cumulative point whose window runs backwards, and it would drop it silently.
        let start = start_unix_nanos().to_string();
        let now = now_unix_nanos().to_string();

        let mut metrics: Vec<J> = Vec::new();
        for m in self.scalars() {
            metrics.push(if m.monotonic {
                json!({
                    "name": m.name,
                    "unit": "1",
                    "sum": {
                        "dataPoints": [{
                            "asInt": m.value.to_string(),
                            "startTimeUnixNano": start,
                            "timeUnixNano": now,
                        }],
                        "aggregationTemporality": 2,
                        "isMonotonic": true,
                    }
                })
            } else {
                json!({
                    "name": m.name,
                    "unit": "1",
                    "gauge": { "dataPoints": [{ "asInt": m.value.to_string(), "timeUnixNano": now }] }
                })
            });
        }
        for (name, h) in self.histograms() {
            metrics.push(json!({
                "name": name,
                "unit": "us",
                "histogram": {
                    "dataPoints": [{
                        "count": h.count().to_string(),
                        "sum": h.sum_us() as f64,
                        "bucketCounts": h.counts().iter().map(u64::to_string).collect::<Vec<_>>(),
                        "explicitBounds": Histogram::bounds(),
                        "startTimeUnixNano": start,
                        "timeUnixNano": now,
                    }],
                    "aggregationTemporality": 2,
                }
            }));
        }

        json!({
            "resourceMetrics": [{
                "resource": { "attributes": resource_attributes(service) },
                "scopeMetrics": [{
                    "scope": { "name": "beck" },
                    "metrics": metrics
                }]
            }]
        })
    }

    /// OpenMetrics 1.0.0 text, for a Prometheus scraper — `docs/12` §12.8's chartered row.
    ///
    /// The same numbers [`Telemetry::otlp_metrics`] exports, read off the same tables, in the other
    /// exposition format the ecosystem speaks. It **adds no measurement**: every value is already
    /// recorded on the serving path, which is what makes a second export cheap rather than a second
    /// cost.
    ///
    /// Three things the format requires and the JSON does not, so they are decided here:
    ///
    /// * **A name is `[a-zA-Z_:][a-zA-Z0-9_:]*`**, so `beck.events.appended` cannot be spelled. The
    ///   dots become underscores, which is the ecosystem's own transliteration of an OTLP name.
    /// * **A counter's name ends `_total`.** OpenMetrics requires it; the older Prometheus text
    ///   format merely prefers it, so satisfying the stricter reader satisfies both.
    /// * **Durations are seconds.** The histograms hold microseconds and Prometheus's convention is
    ///   base units, so the values are divided by a million and the names end `_seconds`. The
    ///   *boundary* is unchanged: a bucket holding observations at or below 1 µs holds the same ones
    ///   at or below 1e-6 s.
    ///
    /// `# UNIT` and `# EOF` are OpenMetrics lines an older 0.0.4 parser reads as comments, so one
    /// body serves both readers rather than content-negotiating between them.
    pub fn openmetrics(&self, service: &str) -> String {
        use std::fmt::Write;
        let mut out = String::new();
        // The service every sample is about. The JSON hangs it on a `resource`; a text exposition
        // has no resource, so it is a label — `job` being the name a scraper already uses for it.
        let job = escape_label(service);

        for m in self.scalars() {
            let name = prometheus_name(m.name);
            let name = if m.monotonic {
                format!("{name}_total")
            } else {
                name
            };
            let _ = writeln!(
                out,
                "# TYPE {name} {}",
                if m.monotonic { "counter" } else { "gauge" }
            );
            let _ = writeln!(out, "# HELP {name} {}", escape_help(m.help));
            let _ = writeln!(out, "{name}{{job=\"{job}\"}} {}", m.value);
        }

        for (dotted, h) in self.histograms() {
            let name = format!("{}_seconds", prometheus_name(dotted));
            let _ = writeln!(out, "# TYPE {name} histogram");
            let _ = writeln!(out, "# UNIT {name} seconds");
            let _ = writeln!(out, "# HELP {name} how long this took, in seconds");
            // Cumulative, which is what the format means by a bucket: the stored counts are per
            // bucket and the last is the overflow, so the running total is the answer and its final
            // value is the `+Inf` bucket — equal, by construction, to `_count`.
            let counts = h.counts();
            let bounds = Histogram::bounds();
            let mut running = 0u64;
            for (i, c) in counts.iter().enumerate() {
                running += c;
                let le = match bounds.get(i) {
                    Some(us) => canonical_number(us / 1.0e6),
                    None => "+Inf".to_string(),
                };
                let _ = writeln!(out, "{name}_bucket{{job=\"{job}\",le=\"{le}\"}} {running}");
            }
            let _ = writeln!(
                out,
                "{name}_sum{{job=\"{job}\"}} {}",
                canonical_number(h.sum_us() as f64 / 1.0e6)
            );
            let _ = writeln!(out, "{name}_count{{job=\"{job}\"}} {}", h.count());
        }

        // OpenMetrics requires it, and it is the one line that distinguishes a complete body from a
        // truncated one — which is why the specification has it at all.
        out.push_str("# EOF\n");
        out
    }

    /// OTLP/HTTP JSON for logs — the body of a POST to `/v1/logs`.
    pub fn otlp_logs(&self, service: &str, limit: usize) -> J {
        let records: Vec<J> = self
            .records(limit)
            .into_iter()
            .map(|r| {
                let mut attributes = vec![json!({
                    "key": "code.namespace",
                    "value": { "stringValue": r.target }
                })];
                // The correlation key. Not a trace id: with this, the exact state is reproducible.
                if let Some(seq) = r.seq {
                    attributes.push(json!({
                        "key": "beck.seq",
                        "value": { "intValue": seq.to_string() }
                    }));
                }
                json!({
                    "timeUnixNano": r.at_unix_nanos.to_string(),
                    "severityNumber": severity_number(r.level),
                    "severityText": r.level,
                    "body": { "stringValue": r.message },
                    "attributes": attributes,
                })
            })
            .collect();

        json!({
            "resourceLogs": [{
                "resource": { "attributes": resource_attributes(service) },
                "scopeLogs": [{ "scope": { "name": "beck" }, "logRecords": records }]
            }]
        })
    }
}

/// OTel's severity numbers: DEBUG 5, INFO 9, WARN 13, ERROR 17.
fn severity_number(level: &str) -> u8 {
    match level {
        "TRACE" => 1,
        "DEBUG" => 5,
        "WARN" => 13,
        "ERROR" => 17,
        _ => 9,
    }
}

/// One scalar metric, as both exports need it: the name, what it means, and its value.
///
/// `monotonic` is the whole distinction between the two kinds — a counter only rises and a gauge
/// moves either way — and both exports read it rather than being told twice.
struct Scalar {
    name: &'static str,
    help: &'static str,
    monotonic: bool,
    value: u64,
}

/// An OTLP name as a Prometheus one: `beck.log.head` → `beck_log_head`.
///
/// A Prometheus metric name is `[a-zA-Z_:][a-zA-Z0-9_:]*`, so a dot cannot appear in one. Replacing
/// it with an underscore is what the ecosystem's own OTLP-to-Prometheus translation does, which
/// matters more than it looks: a dashboard written against a collector's output and one written
/// against this endpoint should name the same series.
fn prometheus_name(dotted: &str) -> String {
    dotted
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '_' || c == ':' {
                c
            } else {
                '_'
            }
        })
        .collect()
}

/// A float as OpenMetrics' **Canonical Numbers** rule renders it.
///
/// The specification does not leave this to the runtime, because an `le` is a *label value* and an
/// end user reads it: "the target rendering is equivalent to the default Go rendering of float64
/// values (i.e. `%g`), with a `.0` appended in case there is no decimal point or exponent". Its own
/// examples pin the two thresholds — `0.0001` and `1e-05`, `100000.0` and `1e+06` — so exponent
/// form is used exactly when the decimal exponent is below -4 or at least 6, with the exponent
/// itself signed and at least two digits.
///
/// Rust's own two formats are each half of this: `{}` never uses exponent form and `{:e}` always
/// does, and neither pads the exponent. So `{:e}` supplies the shortest mantissa and the exponent,
/// and this chooses between them.
fn canonical_number(x: f64) -> String {
    // `{:e}` is always `<mantissa>e<exp>`, and the mantissa is the shortest that round-trips —
    // which is what "default Go rendering" means by its digits.
    let sci = format!("{x:e}");
    let (mantissa, exp) = sci.split_once('e').expect("`{:e}` always has an exponent");
    let exp: i32 = exp.parse().expect("`{:e}` always has an integer exponent");
    // Below -4 or at least 6, which are the two boundaries the examples above pin.
    if !(-4..6).contains(&exp) {
        return format!(
            "{mantissa}e{}{:02}",
            if exp < 0 { '-' } else { '+' },
            exp.abs()
        );
    }
    let decimal = format!("{x}");
    if decimal.contains('.') {
        decimal
    } else {
        format!("{decimal}.0")
    }
}

/// `HELP` text, escaped as the format requires: a backslash and a newline are the two that matter.
fn escape_help(help: &str) -> String {
    help.replace('\\', "\\\\").replace('\n', "\\n")
}

/// A label value, escaped as the format requires — the quote as well, since it closes the value.
fn escape_label(value: &str) -> String {
    value
        .replace('\\', "\\\\")
        .replace('"', "\\\"")
        .replace('\n', "\\n")
}

fn resource_attributes(service: &str) -> J {
    json!([
        { "key": "service.name", "value": { "stringValue": service } },
        { "key": "telemetry.sdk.name", "value": { "stringValue": "beck" } },
        { "key": "telemetry.sdk.language", "value": { "stringValue": "rust" } },
    ])
}

pub fn now_unix_nanos() -> u64 {
    beck_core::clock::process_clock().now_nanos()
}

fn start_unix_nanos() -> u64 {
    static START: OnceLock<u64> = OnceLock::new();
    *START.get_or_init(now_unix_nanos)
}

/// The process-wide instruments.
pub fn telemetry() -> &'static Telemetry {
    static T: OnceLock<Telemetry> = OnceLock::new();
    T.get_or_init(|| {
        // Stamp the start when the instruments come into existence, so a cumulative metric's
        // window begins when the process began rather than when someone first asked for it.
        start_unix_nanos();
        Telemetry::default()
    })
}

/// Time a block and record it, returning what the block returned.
pub fn timed<T>(h: &Histogram, f: impl FnOnce() -> T) -> T {
    let started = std::time::Instant::now();
    let out = f();
    h.record(started.elapsed());
    out
}

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

    #[test]
    fn histogram_buckets_are_powers_of_two_and_bound_the_value() {
        let h = Histogram::default();
        for us in [0u64, 1, 3, 100, 5_000, 900_000, 90_000_000] {
            h.record_us(us);
        }
        assert_eq!(h.count(), 7);
        assert_eq!(h.sum_us(), 1 + 3 + 100 + 5_000 + 900_000 + 90_000_000);
        // Every observation lands in a bucket, including the one that overflows the range.
        assert_eq!(h.counts().iter().sum::<u64>(), 7);
        // A bucketed quantile is an upper bound, and must not understate.
        let h2 = Histogram::default();
        for _ in 0..99 {
            h2.record_us(10);
        }
        h2.record_us(100_000);
        assert!(h2.quantile_us(0.50) >= 10, "p50 understated");
        assert!(h2.quantile_us(0.99) >= 10);
        assert!(
            h2.quantile_us(1.0) >= 100_000,
            "the max must not be understated: {}",
            h2.quantile_us(1.0)
        );
    }

    #[test]
    fn the_ring_is_bounded_and_keeps_the_newest() {
        let t = Telemetry::default();
        for i in 0..RING_CAPACITY + 500 {
            t.log("INFO", "test", format!("message {i}"), Some(i as u64));
        }
        let records = t.records(RING_CAPACITY * 2);
        assert_eq!(records.len(), RING_CAPACITY, "the ring is not bounded");
        assert_eq!(
            records[0].message,
            format!("message {}", RING_CAPACITY + 499),
            "records() must be newest first"
        );
        assert_eq!(records[0].seq, Some((RING_CAPACITY + 499) as u64));
    }

    /// `canonical_number` against **the specification's own published values**.
    ///
    /// Every number below is copied out of OpenMetrics 1.0.0 — the two "Exposers SHOULD produce
    /// output for…" lists in *Considerations: Canonical Numbers*, and the `le` values of the
    /// histogram example that section governs. They are the oracle for the same reason `clbg/`
    /// rebuilds its constants from the Game's own output: a rendering checked against a rule
    /// somebody restated is a rendering checked against their reading of it.
    ///
    /// The two thresholds are what this is really about. `0.0001` renders as a decimal and `1e-05`
    /// does not; `100000.0` renders as a decimal and `1e+06` does not. Both boundaries are one
    /// comparison away from being wrong in a way no round trip could see, because a parser reads
    /// either spelling perfectly well — it is the *human* comparing two dashboards who cannot.
    #[test]
    fn the_le_rendering_is_the_specifications_own() {
        for (value, want) in [
            // "the values 0.0 up to 10.0 in 0.001 increments"
            (0.0, "0.0"),
            (0.001, "0.001"),
            (0.002, "0.002"),
            (0.01, "0.01"),
            (0.1, "0.1"),
            (0.9, "0.9"),
            (0.95, "0.95"),
            (0.99, "0.99"),
            (0.999, "0.999"),
            (1.0, "1.0"),
            (1.7, "1.7"),
            (10.0, "10.0"),
            // "the values 1e-10 up to 1e+10 in powers of ten"
            (1e-10, "1e-10"),
            (1e-9, "1e-09"),
            (1e-5, "1e-05"),
            (0.0001, "0.0001"),
            (100000.0, "100000.0"),
            (1e6, "1e+06"),
            (1e10, "1e+10"),
            // …and the wide, deliberately atypical `le` values of the histogram example.
            (1e23, "1e+23"),
            (1.1e23, "1.1e+23"),
        ] {
            assert_eq!(
                canonical_number(value),
                want,
                "the specification renders {value:?} as `{want}`"
            );
        }
    }

    /// The exposition obeys the rules OpenMetrics states, checked as rules rather than by a parser.
    ///
    /// **Nothing in this workspace is a Prometheus scraper**, so this is not a foreign reader
    /// accepting the body — it is the specification's own MUSTs, encoded, which is the same position
    /// [`adr/0030`](../../../../../docs/adr/0030-the-webassembly-emitter-writes-its-own-bytes.md) takes
    /// about a format with no local reader. What it buys over a round trip is that a writer checked
    /// by its own reader agrees with itself: the rules below are about the *bytes*, and a reader
    /// written here would have been written to accept whatever these emit.
    #[test]
    fn the_exposition_obeys_the_rules_the_specification_states() {
        let t = Telemetry::default();
        t.events_appended.add(7);
        t.sessions.set(2);
        t.fold.record_us(3);
        t.fold.record_us(9_000);
        let body = t.openmetrics("todo");

        // "Expositions MUST end with EOF and SHOULD end with 'EOF\n'."
        assert!(body.ends_with("# EOF\n"), "{body}");
        // "Line endings ... MUST NOT contain carriage returns."
        assert!(!body.contains('\r'));

        let mut types: std::collections::BTreeMap<String, String> = Default::default();
        let mut units: std::collections::BTreeMap<String, String> = Default::default();
        let mut samples: Vec<(String, String, String)> = Vec::new();
        for line in body.lines() {
            if let Some(rest) = line.strip_prefix("# TYPE ") {
                let (name, kind) = rest.split_once(' ').expect("`# TYPE <name> <kind>`");
                assert!(
                    types.insert(name.to_string(), kind.to_string()).is_none(),
                    "MUST NOT be more than one of each type of metadata line: {name}"
                );
            } else if let Some(rest) = line.strip_prefix("# UNIT ") {
                let (name, unit) = rest.split_once(' ').expect("`# UNIT <name> <unit>`");
                // "an underscore and the unit MUST be the suffix of the MetricFamily name"
                assert!(
                    name.ends_with(&format!("_{unit}")),
                    "`{name}` does not end with `_{unit}`"
                );
                units.insert(name.to_string(), unit.to_string());
            } else if line.starts_with("# HELP ") || line == "# EOF" {
                // Metadata, and the terminator.
            } else {
                // "Aside from this metadata and the EOF line ... you MUST NOT expose lines
                // beginning with a #."
                assert!(!line.starts_with('#'), "stray comment line: {line}");
                let (name, value) = line.rsplit_once(' ').expect("`<series> <value>`");
                let (name, labels) = match name.split_once('{') {
                    Some((n, l)) => (n, l.trim_end_matches('}')),
                    None => (name, ""),
                };
                // The ABNF for a metric name, and the reason a dot cannot survive transliteration.
                assert!(
                    name.starts_with(|c: char| c.is_ascii_alphabetic() || c == '_' || c == ':')
                        && name
                            .chars()
                            .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == ':'),
                    "`{name}` is not a metric name the ABNF admits"
                );
                samples.push((name.to_string(), labels.to_string(), value.to_string()));
            }
        }

        assert!(!types.is_empty() && !samples.is_empty());
        // "A counter's Total Value Sample MetricName MUST have the suffix `_total`."
        for (family, kind) in &types {
            if kind == "counter" {
                assert!(family.ends_with("_total"), "counter `{family}`");
            }
        }
        // Every sample belongs to a family that declared its type — the rule that makes the
        // exposition self-describing rather than a list of numbers.
        for (name, _, _) in &samples {
            assert!(
                types.keys().any(|f| name == f
                    || ["_bucket", "_sum", "_count"]
                        .iter()
                        .any(|s| name == &format!("{f}{s}"))),
                "`{name}` has no `# TYPE` line"
            );
        }

        // "Buckets MUST be sorted in number increasing order of `le`", and "if and only if a Sum
        // Value is present ... the +Inf Bucket value MUST also appear ... with the suffix `_count`".
        for (family, kind) in &types {
            if kind != "histogram" {
                continue;
            }
            let buckets: Vec<(&str, u64)> = samples
                .iter()
                .filter(|(n, _, _)| n == &format!("{family}_bucket"))
                .map(|(_, labels, v)| {
                    let le = labels
                        .split(',')
                        .find_map(|l| l.strip_prefix("le=\""))
                        .expect("a bucket has an `le`")
                        .trim_end_matches('"');
                    (le, v.parse::<u64>().expect("a bucket count"))
                })
                .collect();
            assert_eq!(buckets.last().expect("buckets").0, "+Inf", "{family}");
            let mut previous = f64::NEG_INFINITY;
            for (le, _) in &buckets {
                let n = if *le == "+Inf" {
                    f64::INFINITY
                } else {
                    le.parse().expect("an `le` is a number")
                };
                assert!(n > previous, "{family}: {le} does not increase");
                previous = n;
            }
            let counts: Vec<u64> = buckets.iter().map(|(_, c)| *c).collect();
            assert!(
                counts.windows(2).all(|w| w[1] >= w[0]),
                "{family}: buckets are cumulative"
            );
            let total = samples
                .iter()
                .find(|(n, _, _)| n == &format!("{family}_count"))
                .expect("a histogram with a sum has a count")
                .2
                .parse::<u64>()
                .expect("a count");
            assert_eq!(
                total,
                *counts.last().expect("buckets"),
                "{family}: `_count` is the `+Inf` bucket"
            );
        }
    }

    /// The two exports carry the same numbers, because they read the same table.
    ///
    /// They share [`Telemetry::scalars`] now, so this cannot drift by one export being edited — but
    /// the *transliteration* can, and a name that reaches Prometheus as something else is a
    /// dashboard that silently stops matching a collector's.
    #[test]
    fn both_exports_carry_the_same_scalars() {
        let t = Telemetry::default();
        t.events_appended.add(7);
        t.sessions.set(2);
        let json = t.otlp_metrics("todo");
        let text = t.openmetrics("todo");

        let metrics = json["resourceMetrics"][0]["scopeMetrics"][0]["metrics"]
            .as_array()
            .expect("metrics")
            .clone();
        assert!(metrics.len() >= 20, "{}", metrics.len());
        for m in &metrics {
            let dotted = m["name"].as_str().expect("a name");
            let (value, suffix) = if let Some(p) = m.get("sum") {
                (
                    p["dataPoints"][0]["asInt"].as_str().map(str::to_string),
                    "_total",
                )
            } else if let Some(p) = m.get("gauge") {
                (p["dataPoints"][0]["asInt"].as_str().map(str::to_string), "")
            } else {
                // A histogram: renamed to seconds, and held by the rules test above.
                continue;
            };
            let want = format!(
                "{}{suffix}{{job=\"todo\"}} {}",
                dotted.replace('.', "_"),
                value.expect("an integer point")
            );
            assert!(
                text.lines().any(|l| l == want),
                "the JSON has `{dotted}` and the text has no `{want}`"
            );
        }
    }

    #[test]
    fn the_metrics_body_is_shaped_like_otlp() {
        // Not a schema validation — a guard that the field names stay the ones a collector reads.
        // Getting `aggregationTemporality` or `asInt`-as-a-string wrong produces a body that is
        // accepted and silently dropped, which is the failure mode worth a test.
        let t = Telemetry::default();
        t.events_appended.add(7);
        t.fold.record_us(1_500);
        let body = t.otlp_metrics("todo");

        let metrics = &body["resourceMetrics"][0]["scopeMetrics"][0]["metrics"];
        let by_name = |n: &str| {
            metrics
                .as_array()
                .unwrap()
                .iter()
                .find(|m| m["name"] == n)
                .unwrap_or_else(|| panic!("no metric {n}"))
                .clone()
        };

        let appended = by_name("beck.events.appended");
        assert_eq!(
            appended["sum"]["dataPoints"][0]["asInt"], "7",
            "int64 fields are JSON strings"
        );
        assert_eq!(appended["sum"]["aggregationTemporality"], 2, "CUMULATIVE");
        assert_eq!(appended["sum"]["isMonotonic"], true);

        let fold = by_name("beck.fold.duration");
        let point = &fold["histogram"]["dataPoints"][0];
        assert_eq!(point["count"], "1");
        assert_eq!(point["sum"], 1500.0);
        assert_eq!(
            point["bucketCounts"].as_array().unwrap().len(),
            point["explicitBounds"].as_array().unwrap().len() + 1,
            "OTLP requires exactly one more bucket count than bound"
        );

        assert_eq!(
            body["resourceMetrics"][0]["resource"]["attributes"][0]["value"]["stringValue"],
            "todo"
        );

        // A cumulative point's window must not run backwards. It did, on the first export, because
        // the start was initialised lazily *after* `now` was read — and the failure mode is a
        // collector silently dropping the point.
        for m in metrics.as_array().unwrap() {
            let point = m
                .get("sum")
                .or_else(|| m.get("histogram"))
                .map(|k| &k["dataPoints"][0]);
            if let Some(p) = point {
                let start: u64 = p["startTimeUnixNano"].as_str().unwrap().parse().unwrap();
                let now: u64 = p["timeUnixNano"].as_str().unwrap().parse().unwrap();
                assert!(
                    start <= now,
                    "{} reports a window ending before it began",
                    m["name"]
                );
            }
        }
    }

    #[test]
    fn a_log_record_carries_the_sequence_number_it_is_about() {
        // The claim this module rests on: correlation is `seq`, so a record in any backend points
        // at a state `beck replay --to` can reproduce.
        let t = Telemetry::default();
        t.log("ERROR", "beck_rt::app", "append failed".into(), Some(41));
        t.log("INFO", "beck_rt::http", "listening".into(), None);

        let body = t.otlp_logs("todo", 10);
        let records = body["resourceLogs"][0]["scopeLogs"][0]["logRecords"]
            .as_array()
            .unwrap()
            .clone();
        assert_eq!(records.len(), 2);

        let failure = records
            .iter()
            .find(|r| r["severityText"] == "ERROR")
            .unwrap();
        assert_eq!(failure["severityNumber"], 17);
        assert_eq!(failure["body"]["stringValue"], "append failed");
        let seq = failure["attributes"]
            .as_array()
            .unwrap()
            .iter()
            .find(|a| a["key"] == "beck.seq")
            .expect("an error about an append must say which one");
        assert_eq!(seq["value"]["intValue"], "41");

        // …and a record with no sequence number does not invent one.
        let listening = records
            .iter()
            .find(|r| r["severityText"] == "INFO")
            .unwrap();
        assert!(
            !listening["attributes"]
                .as_array()
                .unwrap()
                .iter()
                .any(|a| a["key"] == "beck.seq"),
            "a record about no particular state must not claim one"
        );
    }
}