crates/rutster/src/config.rs

@@ -276,6 +276,22 @@ pub fn tls_key_path(raw: Option<String>) -> Result<Option<PathBuf>, String> {
     }
 }
 
+/// Resolve the metrics bind address from `RUTSTER_METRICS_BIND`.
+///
+/// `None` → loopback-only `127.0.0.1:9090`. The separate listener
+/// invariant (deploy-C §5.5) means this endpoint must never be reached
+/// through the TLS-terminating edge, so the default is intentionally
+/// localhost-only — operators who want Prometheus to scrape from another
+/// host must set the variable explicitly.
+pub fn metrics_bind(raw: Option<String>) -> Result<SocketAddr, String> {
+    match raw {
+        None => Ok("127.0.0.1:9090".parse().expect("static default parses")),
+        Some(s) => s
+            .parse()
+            .map_err(|e| format!("RUTSTER_METRICS_BIND {s:?} is not a socket address: {e}")),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -302,6 +318,28 @@ mod tests {
         assert!(err.contains("RUTSTER_HTTP_BIND"));
     }
 
+    #[test]
+    fn metrics_bind_defaults_to_loopback_9090_when_unset() {
+        assert_eq!(
+            metrics_bind(None).unwrap(),
+            "127.0.0.1:9090".parse::<SocketAddr>().unwrap()
+        );
+    }
+
+    #[test]
+    fn metrics_bind_parses_override() {
+        assert_eq!(
+            metrics_bind(Some("0.0.0.0:9100".into())).unwrap(),
+            "0.0.0.0:9100".parse::<SocketAddr>().unwrap()
+        );
+    }
+
+    #[test]
+    fn metrics_bind_rejects_garbage_with_var_name_in_error() {
+        let err = metrics_bind(Some("not-an-addr".into())).unwrap_err();
+        assert!(err.contains("RUTSTER_METRICS_BIND"));
+    }
+
     #[test]
     fn port_range_parses_inclusive_pair() {
         assert_eq!(parse_port_range("10000-20000").unwrap(), (10000, 20000));

crates/rutster/src/lib.rs

@@ -25,6 +25,7 @@
 pub mod config;
 pub mod event_sink;
 pub mod media_thread;
+pub mod metrics;
 pub mod routes;
 pub mod serve;
 pub mod serve_tls;

crates/rutster/src/main.rs

@@ -162,6 +162,23 @@ async fn main() {
         ws_ping_interval,
     );
 
+    // Deploy slice C §5.5: /metrics on its own listener, never through
+    // the TLS edge. The app_state clone is the same one the main router
+    // uses; the media thread behind it is shared.
+    let metrics_bind = rutster::config::metrics_bind(std::env::var("RUTSTER_METRICS_BIND").ok())
+        .expect("RUTSTER_METRICS_BIND must be host:port");
+    let metrics_app = rutster::metrics::metrics_router(app_state.clone());
+    tokio::spawn(async move {
+        let listener = tokio::net::TcpListener::bind(metrics_bind)
+            .await
+            .expect("RUTSTER_METRICS_BIND bind failed");
+        info!(%metrics_bind, "metrics listening");
+        axum::serve(listener, metrics_app)
+            .tcp_nodelay(true)
+            .await
+            .unwrap();
+    });
+
     let app = router(app_state).merge(trunk_router);
 
     // Deploy slice C §5.4: branch to TLS or plaintext. ALWAYS-COMPILED,
@@ -173,6 +190,7 @@ async fn main() {
         .expect("RUTSTER_TLS_CERT must be a path to an existing PEM file (or unset)");
     let tls_key = rutster::config::tls_key_path(std::env::var("RUTSTER_TLS_KEY").ok())
         .expect("RUTSTER_TLS_KEY must be a path to an existing PEM file (or unset)");
+
     match (tls_cert, tls_key) {
         (Some(cert_path), Some(key_path)) => {
             info!(cert = %cert_path.display(), key = %key_path.display(), "TLS listener enabled");

crates/rutster/src/metrics.rs

@@ -0,0 +1,124 @@
+//! # Prometheus `/metrics` endpoint (deploy-C §5.5)
+//!
+//! Hand-rolled text exposition on a **separate** axum listener so it never
+//! transits the TLS-terminating edge (Caddy). The format is the
+//! Prometheus 0.0.4 text line protocol; zero new dependencies.
+//!
+//! The renderer deliberately skips `rutster_event_sink_drops_total` when
+//! the sink reports `None`: emitting "0" would falsely tell an operator
+//! the sink is healthy-with-zero-drops, when really no event sink is
+//! wired at all. This is a convention call, not a bug — see
+//! `render_prometheus_text`.
+
+use axum::Router;
+use axum::extract::State;
+use axum::http::{StatusCode, header};
+use axum::response::{IntoResponse, Response};
+use axum::routing::get;
+
+use crate::media_thread::MediaStats;
+use crate::session_map::AppState;
+
+/// Build the isolated `/metrics` router.
+///
+/// This router is intentionally NOT merged into the main `routes::router`;
+/// `/metrics` must live on its own listener so the edge never forwards it
+/// and it stays inside the trust boundary.
+pub fn metrics_router(state: AppState) -> Router {
+    Router::new()
+        .route("/metrics", get(metrics_handler))
+        .with_state(state)
+}
+
+/// GET /metrics — return a Prometheus text snapshot of the media thread.
+///
+/// Mirrors `routes::readyz` but with a 1 s timeout (vs readyz's 250 ms).
+/// Prometheus's default scrape timeout is 10 s; 1 s is generous while still
+/// letting a scraper mark a wedged node as stale.
+async fn metrics_handler(State(state): State<AppState>) -> Response {
+    let (reply, rx) = tokio::sync::oneshot::channel();
+    let stats = match tokio::time::timeout(std::time::Duration::from_secs(1), async {
+        state
+            .cmd_tx
+            .send(crate::media_thread::MediaCmd::Stats { reply })
+            .await
+            .ok()?;
+        rx.await.ok()
+    })
+    .await
+    {
+        Ok(Some(s)) => s,
+        // Elapsed, send failure, or dropped reply all mean the media thread
+        // cannot be sampled right now.
+        _ => {
+            return (StatusCode::SERVICE_UNAVAILABLE, "media thread unresponsive").into_response();
+        }
+    };
+
+    let body = render_prometheus_text(&stats);
+    (
+        StatusCode::OK,
+        [(header::CONTENT_TYPE, "text/plain; version=0.0.4")],
+        body,
+    )
+        .into_response()
+}
+
+/// Render a [`MediaStats`] snapshot as Prometheus 0.0.4 text.
+///
+/// Metrics that are conceptually counters get `_total` suffixes; gauges do
+/// not. `rutster_event_sink_drops_total` is emitted **only** when the sink
+/// reports a concrete drop count — `None` means "no sink present", and
+/// emitting "0" would mislead scrapers into believing a sink exists and is
+/// healthy.
+pub fn render_prometheus_text(stats: &MediaStats) -> String {
+    use std::fmt::Write;
+
+    let mut out = String::new();
+
+    let _ = writeln!(
+        out,
+        "# HELP rutster_sessions_active Current number of active sessions\n# TYPE rutster_sessions_active gauge\nrutster_sessions_active {}\n",
+        stats.sessions
+    );
+
+    let _ = writeln!(
+        out,
+        "# HELP rutster_max_sessions Admission capacity for this node\n# TYPE rutster_max_sessions gauge\nrutster_max_sessions {}\n",
+        stats.max_sessions
+    );
+
+    let _ = writeln!(
+        out,
+        "# HELP rutster_draining Non-zero when the node is draining and rejecting new sessions\n# TYPE rutster_draining gauge\nrutster_draining {}\n",
+        u64::from(stats.draining)
+    );
+
+    let _ = writeln!(
+        out,
+        "# HELP rutster_tick_overruns_total Ticks whose work exceeded the meta-tick budget\n# TYPE rutster_tick_overruns_total counter\nrutster_tick_overruns_total {}\n",
+        stats.tick_overruns
+    );
+
+    let _ = writeln!(
+        out,
+        "# HELP rutster_last_tick_micros Wall time of the last meta-tick in microseconds\n# TYPE rutster_last_tick_micros gauge\nrutster_last_tick_micros {}\n",
+        stats.last_tick_micros
+    );
+
+    let _ = writeln!(
+        out,
+        "# HELP rutster_admission_rejects_total Sessions rejected due to capacity or drain\n# TYPE rutster_admission_rejects_total counter\nrutster_admission_rejects_total {}\n",
+        stats.admission_rejects
+    );
+
+    if let Some(drops) = stats.event_sink_drops {
+        let _ = writeln!(
+            out,
+            "# HELP rutster_event_sink_drops_total Lifecycle events dropped by the event sink\n# TYPE rutster_event_sink_drops_total counter\nrutster_event_sink_drops_total {}\n",
+            drops
+        );
+    }
+
+    out
+}

crates/rutster/tests/metrics_endpoint.rs

@@ -0,0 +1,98 @@
+//! Integration tests for the isolated `/metrics` endpoint (deploy-C §5.5).
+//!
+//! `/metrics` lives on its own axum listener/router, so these tests drive
+//! `rutster::metrics::metrics_router` directly without binding a socket —
+//! the same pattern the `/readyz` tests use.
+
+use axum::body::Body;
+use axum::http::{Request, StatusCode};
+use tower::ServiceExt;
+
+/// Given no media thread behind AppState, `/metrics` must return 503.
+///
+/// The closed placeholder sender is the contract surface: a scraper sees
+/// "media thread unresponsive" and can mark the target stale.
+#[tokio::test]
+async fn metrics_endpoint_returns_prometheus_text() {
+    let app = rutster::metrics::metrics_router(rutster::session_map::AppState::default());
+    let resp = app
+        .oneshot(
+            Request::builder()
+                .uri("/metrics")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
+}
+
+/// Given a live media thread, `/metrics` returns Prometheus text with every
+/// core gauge/counter and omits the event-sink drop line (TracingEventSink
+/// reports `None`).
+///
+/// This is a synchronous `#[test]` (not `#[tokio::test]`) because
+/// `MediaThread::shutdown()` uses `blocking_send`/`blocking_recv`, which
+/// panic when called from inside a tokio runtime worker. The test creates
+/// its own runtime only to drive the async axum oneshot.
+#[test]
+fn metrics_endpoint_with_live_media_thread_returns_200() {
+    let rt = tokio::runtime::Runtime::new().expect("tokio runtime");
+    let (state, thread) = rutster::session_map::AppState::default()
+        .spawn_media_thread(rt.handle().clone())
+        .expect("spawn media thread");
+
+    let app = rutster::metrics::metrics_router(state);
+    let resp = rt.block_on(async {
+        app.oneshot(
+            Request::builder()
+                .uri("/metrics")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap()
+    });
+
+    assert_eq!(resp.status(), StatusCode::OK);
+    let ct = resp
+        .headers()
+        .get(axum::http::header::CONTENT_TYPE)
+        .expect("content-type header present")
+        .to_str()
+        .expect("content-type is valid ascii");
+    assert_eq!(ct, "text/plain; version=0.0.4");
+
+    let bytes = rt
+        .block_on(axum::body::to_bytes(resp.into_body(), 8192))
+        .unwrap();
+    let text = String::from_utf8(bytes.to_vec()).expect("metrics body is utf-8");
+
+    let expected = [
+        "# TYPE rutster_sessions_active gauge",
+        "# TYPE rutster_max_sessions gauge",
+        "# TYPE rutster_draining gauge",
+        "# TYPE rutster_tick_overruns_total counter",
+        "# TYPE rutster_last_tick_micros gauge",
+        "# TYPE rutster_admission_rejects_total counter",
+        "rutster_sessions_active",
+        "rutster_max_sessions",
+        "rutster_draining",
+        "rutster_tick_overruns_total",
+        "rutster_last_tick_micros",
+        "rutster_admission_rejects_total",
+    ];
+    for needle in expected {
+        assert!(
+            text.contains(needle),
+            "metrics output missing {needle:?}; body:\n{text}"
+        );
+    }
+
+    assert!(
+        !text.contains("rutster_event_sink_drops_total"),
+        "TracingEventSink must not emit drop counter; body:\n{text}"
+    );
+
+    thread.shutdown();
+}