liminal_server/
metrics.rs

1//! Server-side metric recording over the process-global liminal registry.
2//!
3//! [`init`] installs a global [`MetricsRegistry`] — which flips the library's
4//! metrics gate on for this process — and registers the first-wave server
5//! families, caching their handles. Every recording helper no-ops until `init`
6//! has run, so a standalone liminal library user (who never calls `init`) pays
7//! nothing and the registry stays disabled.
8
9use std::sync::OnceLock;
10
11use liminal::metrics::{
12    CounterHandle, GaugeHandle, MetricsRegistry, global_registry, install_global_registry,
13};
14
15const CONNECTIONS_ACTIVE: &str = "liminal_connections_active";
16const PUBLISHES_TOTAL: &str = "liminal_publishes_total";
17const DELIVERIES_TOTAL: &str = "liminal_deliveries_total";
18
19static SERVER_METRICS: OnceLock<ServerMetrics> = OnceLock::new();
20
21/// Cached handles for the first-wave server metrics.
22#[derive(Clone, Debug)]
23struct ServerMetrics {
24    connections_active: GaugeHandle,
25    publishes_total: CounterHandle,
26    deliveries_total: CounterHandle,
27}
28
29/// Enables metrics for this server process and registers the server families.
30///
31/// Idempotent: a second call is a no-op. Called once at server startup so the
32/// `/metrics` endpoint has data to render; the recording helpers below stay
33/// inert until this runs.
34pub fn init() {
35    if SERVER_METRICS.get().is_some() {
36        return;
37    }
38    let Some(registry) = global_or_install() else {
39        return;
40    };
41    if let Some(metrics) = ServerMetrics::register(registry) {
42        let _ = SERVER_METRICS.set(metrics);
43    }
44}
45
46/// Records the spawn of a supervised connection (`liminal_connections_active`
47/// gauge increment). Paired with [`connection_closed`] on every teardown route.
48pub fn connection_spawned() {
49    if let Some(metrics) = SERVER_METRICS.get() {
50        metrics.connections_active.increment();
51    }
52}
53
54/// Records the teardown of a supervised connection (`liminal_connections_active`
55/// gauge decrement). Paired with [`connection_spawned`].
56pub fn connection_closed() {
57    if let Some(metrics) = SERVER_METRICS.get() {
58        metrics.connections_active.decrement();
59    }
60}
61
62/// Records one accepted publish on the services publish path
63/// (`liminal_publishes_total`).
64pub fn publish_accepted() {
65    if let Some(metrics) = SERVER_METRICS.get() {
66        metrics.publishes_total.increment();
67    }
68}
69
70/// Records `count` genuine subscriber deliveries from a single publish
71/// (`liminal_deliveries_total`). A publish that reached no subscriber records
72/// nothing.
73pub fn deliveries_recorded(count: u64) {
74    if count == 0 {
75        return;
76    }
77    if let Some(metrics) = SERVER_METRICS.get() {
78        metrics.deliveries_total.increment_by(count);
79    }
80}
81
82impl ServerMetrics {
83    fn register(registry: &MetricsRegistry) -> Option<Self> {
84        let connections_active = registry
85            .register_gauge(CONNECTIONS_ACTIVE, no_labels())
86            .ok()?;
87        let publishes_total = registry
88            .register_counter(PUBLISHES_TOTAL, no_labels())
89            .ok()?;
90        let deliveries_total = registry
91            .register_counter(DELIVERIES_TOTAL, no_labels())
92            .ok()?;
93        Some(Self {
94            connections_active,
95            publishes_total,
96            deliveries_total,
97        })
98    }
99}
100
101const fn no_labels() -> std::iter::Empty<(&'static str, &'static str)> {
102    std::iter::empty()
103}
104
105/// Returns the process-global registry, installing a fresh one when none exists.
106///
107/// Enabling the gate here (rather than in the library) keeps standalone liminal
108/// users on the disabled fast path; the server is the sole installer.
109fn global_or_install() -> Option<&'static MetricsRegistry> {
110    if let Some(registry) = global_registry() {
111        return Some(registry);
112    }
113    // Best-effort install; if a concurrent caller won the race we still read the
114    // now-installed registry back below.
115    let _ = install_global_registry(MetricsRegistry::new());
116    global_registry()
117}
118
119#[cfg(test)]
120mod tests {
121    use super::{
122        CONNECTIONS_ACTIVE, DELIVERIES_TOTAL, PUBLISHES_TOTAL, connection_spawned,
123        deliveries_recorded, init, publish_accepted,
124    };
125    use liminal::metrics::{global_registry, render};
126
127    #[test]
128    fn init_registers_the_three_server_families_on_the_global_registry()
129    -> Result<(), Box<dyn std::error::Error>> {
130        init();
131        connection_spawned();
132        publish_accepted();
133        deliveries_recorded(2);
134
135        let registry =
136            global_registry().ok_or("init must install and enable the global registry")?;
137        let exposition = render(&registry.snapshot());
138
139        assert!(exposition.contains(CONNECTIONS_ACTIVE));
140        assert!(exposition.contains(PUBLISHES_TOTAL));
141        assert!(exposition.contains(DELIVERIES_TOTAL));
142
143        Ok(())
144    }
145}
liminal_server/metrics.rs

liminal_server/
metrics.rs
