liminal-server 0.21.1

//! `/metrics` Prometheus exposition body over the process-global registry.
//!
//! The health server already speaks a hand-rolled HTTP; this module only
//! supplies the `GET /metrics` body and its content type, leaving the response
//! framing to [`super::endpoint`].

use liminal::metrics::{global_registry, render};

/// Prometheus text-exposition content type served on `/metrics`.
pub(super) const CONTENT_TYPE: &str = "text/plain; version=0.0.4";

/// Renders the current global registry snapshot as Prometheus 0.0.4 text,
/// followed by the dedup head counters read at this scrape
/// ([`crate::metrics::render_dedup_head_counters`]).
///
/// When metrics are disabled — no global registry installed, e.g. a health
/// server started without the server runtime having called
/// [`crate::metrics::init`] — the body is empty, dedup head counters included;
/// the endpoint still answers 200 so a scraper observes a live target.
pub(super) fn render_body() -> String {
    global_registry().map_or_else(String::new, |registry| {
        let mut body = render(&registry.snapshot());
        crate::metrics::render_dedup_head_counters(&mut body);
        body
    })
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::io::{Read, Write};
    use std::net::{SocketAddr, TcpStream};
    use std::sync::{Arc, Mutex, MutexGuard, OnceLock};
    use std::time::Duration;

    use liminal::durability::bridge::block_on;
    use liminal::durability::{
        DedupCache, DedupDecision, DedupEntry, DurabilityError, DurableStore, ScanCursor, ScanPage,
        StoredEntry, dedup_head_counts, head_checkpoint_key,
    };

    use super::{CONTENT_TYPE, render_body};
    use crate::health::checks::SharedReadinessState;
    use crate::health::endpoint::start_health_server;

    /// The idempotency key whose stream [`OneFullPageStore`] holds.
    const KEY: &str = "dedup-head-metrics";

    fn get(address: SocketAddr, path: &str) -> Result<String, Box<dyn std::error::Error>> {
        let mut stream = TcpStream::connect(address)?;
        stream.set_read_timeout(Some(Duration::from_secs(2)))?;
        let request = format!("GET {path} HTTP/1.1\r\nHost: localhost\r\n\r\n");
        stream.write_all(request.as_bytes())?;
        let mut response = String::new();
        stream.read_to_string(&mut response)?;
        Ok(response)
    }

    /// The value of the unlabelled sample `name` in an exposition body.
    fn sample(body: &str, name: &str) -> Option<u64> {
        body.lines()
            .find_map(|line| line.strip_prefix(name)?.strip_prefix(' ')?.parse().ok())
    }

    /// A dedup key stream that fills exactly the first page read of it, every
    /// row an open claim of [`KEY`], and the values a compare-and-swap keeps,
    /// all in memory.
    ///
    /// Sizing the stream by its first read makes it the shortest stream whose
    /// lookup consults the recorded head, whatever page size the dedup cache
    /// reads with (that size is private to `liminal`). Nothing on a lookup
    /// appends or scans, so both refuse by name.
    #[derive(Debug, Default)]
    struct OneFullPageStore {
        /// The stream's length: the limit of the first read from position 0.
        rows: OnceLock<u64>,
        values: Mutex<HashMap<String, u64>>,
    }

    impl OneFullPageStore {
        fn lock_values(&self) -> Result<MutexGuard<'_, HashMap<String, u64>>, DurabilityError> {
            self.values
                .lock()
                .map_err(|_| DurabilityError::ConfigError("value map poisoned".to_owned()))
        }
    }

    #[async_trait::async_trait]
    impl DurableStore for OneFullPageStore {
        async fn apply_routed_batch(
            &self,
            _: &str,
            _: Vec<liminal::durability::RoutedBatchItem>,
        ) -> Result<(), liminal::durability::DurabilityError> {
            Err(liminal::durability::DurabilityError::BatchUnsupported {
                store: std::any::type_name::<Self>(),
            })
        }

        async fn get_routed(
            &self,
            _: &str,
            _: &[u8],
        ) -> Result<Option<Vec<u8>>, liminal::durability::DurabilityError> {
            Err(liminal::durability::DurabilityError::BatchUnsupported {
                store: std::any::type_name::<Self>(),
            })
        }

        async fn append(
            &self,
            _stream_key: &str,
            _payload: Vec<u8>,
            _expected_seq: u64,
        ) -> Result<u64, DurabilityError> {
            Err(DurabilityError::ConfigError(
                "a lookup never appends to this stream".to_owned(),
            ))
        }

        async fn append_all(
            &self,
            stream_key: &str,
            payloads: Vec<Vec<u8>>,
            expected_seq: u64,
        ) -> Result<u64, DurabilityError> {
            Err(DurabilityError::ConfigError(format!(
                "OneFullPageStore is a single-append or read-only test fixture; atomic batch on {stream_key} at {expected_seq} with {} payloads is unsupported",
                payloads.len()
            )))
        }

        async fn read_from(
            &self,
            _stream_key: &str,
            offset: u64,
            limit: usize,
        ) -> Result<Vec<StoredEntry>, DurabilityError> {
            let limit = u64::try_from(limit)
                .map_err(|error| DurabilityError::ConfigError(error.to_string()))?;
            let rows = if offset == 0 {
                *self.rows.get_or_init(|| limit)
            } else {
                self.rows.get().copied().unwrap_or(0)
            };
            let payload = DedupEntry::new(KEY, None, 1).serialize()?;
            Ok((offset..rows.min(offset.saturating_add(limit)))
                .map(|sequence| StoredEntry {
                    payload: payload.clone(),
                    sequence,
                    timestamp: 0,
                })
                .collect())
        }

        async fn cas(
            &self,
            key: &str,
            old_value: u64,
            new_value: u64,
        ) -> Result<(), DurabilityError> {
            let mut values = self.lock_values()?;
            let stored = values.get(key).copied().unwrap_or(0);
            if stored != old_value {
                return Err(DurabilityError::CursorRegression {
                    stored,
                    attempted: old_value,
                });
            }
            values.insert(key.to_owned(), new_value);
            drop(values);
            Ok(())
        }

        async fn read_stream_next_seq(
            &self,
            stream_key: &str,
        ) -> Result<Option<u64>, DurabilityError> {
            Err(DurabilityError::ConfigError(format!(
                "stream head read is unsupported by OneFullPageStore for {stream_key}"
            )))
        }

        async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
            Ok(self.lock_values()?.get(key).copied())
        }

        async fn scan_page(
            &self,
            _prefix: &str,
            _cursor: Option<ScanCursor>,
            _limit: usize,
        ) -> Result<ScanPage, DurabilityError> {
            Err(DurabilityError::ConfigError(
                "a lookup never scans this store".to_owned(),
            ))
        }

        async fn flush(&self) -> Result<(), DurabilityError> {
            Ok(())
        }
    }

    #[test]
    fn render_body_exposes_the_three_server_families_after_activity() {
        crate::metrics::init();
        crate::metrics::connection_spawned();
        crate::metrics::publish_accepted();
        crate::metrics::deliveries_recorded(3);

        let body = render_body();

        // Parseable exposition: every family carries a `# TYPE` header line.
        assert!(body.contains("# TYPE liminal_connections_active gauge"));
        assert!(body.contains("# TYPE liminal_publishes_total counter"));
        assert!(body.contains("# TYPE liminal_deliveries_total counter"));
    }

    #[test]
    fn metrics_endpoint_serves_prometheus_text_with_200() -> Result<(), Box<dyn std::error::Error>>
    {
        crate::metrics::init();
        crate::metrics::publish_accepted();

        let readiness = SharedReadinessState::default();
        let server = start_health_server("127.0.0.1:0".parse()?, readiness)?;
        let response = get(server.local_addr(), "/metrics")?;
        server.shutdown()?;

        assert!(
            response.starts_with("HTTP/1.1 200 "),
            "metrics endpoint must answer 200: {response}"
        );
        assert!(response.contains(&format!("Content-Type: {CONTENT_TYPE}\r\n")));
        let Some((_headers, body)) = response.split_once("\r\n\r\n") else {
            return Err("metrics response had no header/body separator".into());
        };
        assert!(body.contains("liminal_publishes_total"));

        Ok(())
    }

    /// The dedup head counters reach the live `/metrics` under the names
    /// `DedupHeadCounts::named` gives them, each as one counter family. A head
    /// planted past the end of a key stream that fills its first page is
    /// refused and repaired by one lookup, so two counters are nonzero in this
    /// process; every scraped value lies between `dedup_head_counts()` read
    /// just before and just after the GET, which a rendered constant could
    /// not.
    #[test]
    fn metrics_endpoint_serves_the_dedup_head_counters_by_name()
    -> Result<(), Box<dyn std::error::Error>> {
        crate::metrics::init();
        let store = Arc::new(OneFullPageStore::default());
        let cache = DedupCache::new(Arc::clone(&store) as Arc<dyn DurableStore>, "metrics-route");
        // Past the end of any stream, so the lookup refuses it.
        block_on(store.cas(
            &head_checkpoint_key(&cache.stream_key_for(KEY)),
            0,
            u64::MAX,
        ))??;

        let before_lookup = dedup_head_counts();
        assert_eq!(block_on(cache.lookup(KEY))??, Some(DedupDecision::InFlight));
        let before = dedup_head_counts();
        assert!(
            before.refusals > before_lookup.refusals && before.repairs > before_lookup.repairs,
            "the planted head is refused and repaired: {before_lookup:?} -> {before:?}"
        );

        let readiness = SharedReadinessState::default();
        let server = start_health_server("127.0.0.1:0".parse()?, readiness)?;
        let response = get(server.local_addr(), "/metrics")?;
        let after = dedup_head_counts();
        server.shutdown()?;

        assert!(
            response.starts_with("HTTP/1.1 200 "),
            "metrics endpoint must answer 200: {response}"
        );
        let Some((_headers, body)) = response.split_once("\r\n\r\n") else {
            return Err("metrics response had no header/body separator".into());
        };
        assert_eq!(
            before.named().map(|(name, _)| name),
            [
                "liminal_dedup_head_refusals_total",
                "liminal_dedup_head_record_lost_races_total",
                "liminal_dedup_head_repairs_total",
            ],
            "the dedup head counters keep the server's liminal_ prefix and the _total suffix"
        );
        for ((name, low), (_, high)) in before.named().into_iter().zip(after.named()) {
            let type_prefix = format!("# TYPE {name} ");
            let types = body
                .lines()
                .filter(|line| line.starts_with(&type_prefix))
                .collect::<Vec<_>>();
            assert_eq!(
                types,
                [format!("# TYPE {name} counter")],
                "{name} is one counter family: {body}"
            );
            let scraped = sample(body, name).ok_or_else(|| format!("{name} absent: {body}"))?;
            assert!(
                (low..=high).contains(&scraped),
                "{name} scraped {scraped}, read {low} before and {high} after the GET"
            );
        }

        Ok(())
    }

    #[test]
    fn health_and_ready_are_unaffected_by_the_metrics_route()
    -> Result<(), Box<dyn std::error::Error>> {
        crate::metrics::init();

        let readiness = SharedReadinessState::default();
        let server = start_health_server("127.0.0.1:0".parse()?, readiness)?;
        let health = get(server.local_addr(), "/health")?;
        let ready = get(server.local_addr(), "/ready")?;
        server.shutdown()?;

        assert!(health.starts_with("HTTP/1.1 200 "));
        assert!(health.contains("Content-Type: application/json\r\n"));
        // Default readiness is not ready (config not loaded), so /ready stays 503 —
        // proving the new route did not disturb the existing handlers.
        assert!(ready.starts_with("HTTP/1.1 503 "));

        Ok(())
    }
}