claw-store/src/cluster.rs

@@ -19,6 +19,7 @@ pub mod build_cache;
 pub mod client_config;
 pub mod gossip;
 pub mod metrics;
+pub mod prom;
 pub mod refs;
 pub mod rpc;
 pub mod services;
@@ -198,10 +199,8 @@ mod tests {
         let ts = spawn_accepting_listener().await;
         // Short LAN timeout so the test runs fast even if the OS holds the
         // failed connect in a queue rather than refusing immediately.
-        let probe = LanFirstProbe::with_timeouts(
-            Duration::from_millis(50),
-            Duration::from_millis(500),
-        );
+        let probe =
+            LanFirstProbe::with_timeouts(Duration::from_millis(50), Duration::from_millis(500));
         let win = probe
             .probe(&peer_with(Some(lan_dead), Some(ts)))
             .await
@@ -214,10 +213,8 @@ mod tests {
     async fn probe_errors_when_both_routes_fail() {
         let lan_dead = free_but_unbound_addr().await;
         let ts_dead = free_but_unbound_addr().await;
-        let probe = LanFirstProbe::with_timeouts(
-            Duration::from_millis(50),
-            Duration::from_millis(50),
-        );
+        let probe =
+            LanFirstProbe::with_timeouts(Duration::from_millis(50), Duration::from_millis(50));
         let err = probe
             .probe(&peer_with(Some(lan_dead), Some(ts_dead)))
             .await

claw-store/src/cluster/gossip.rs

@@ -22,8 +22,7 @@ use crate::config::ClusterConfig;
 use anyhow::{bail, Context, Result};
 use chitchat::transport::UdpTransport;
 use chitchat::{
-    spawn_chitchat, Chitchat, ChitchatConfig, ChitchatHandle, ChitchatId,
-    FailureDetectorConfig,
+    spawn_chitchat, Chitchat, ChitchatConfig, ChitchatHandle, ChitchatId, FailureDetectorConfig,
 };
 use std::net::SocketAddr;
 use std::sync::Arc;
@@ -182,10 +181,7 @@ impl ClusterGossip {
     /// `ClusterGossip` MUST be held for the daemon's lifetime; dropping
     /// it aborts the gossip task and the node effectively leaves the
     /// cluster (peers observe it as dead within `dead_node_grace_period`).
-    pub async fn bootstrap(
-        cluster: &ClusterConfig,
-        local_name: impl Into<String>,
-    ) -> Result<Self> {
+    pub async fn bootstrap(cluster: &ClusterConfig, local_name: impl Into<String>) -> Result<Self> {
         let local_name = local_name.into();
         if local_name.is_empty() {
             bail!("cluster gossip requires a non-empty local node name");
@@ -354,11 +350,7 @@ fn get_socket(state: &chitchat::NodeState, key: &str) -> Option<SocketAddr> {
     state.get(key).and_then(|s| s.parse().ok())
 }
 
-fn peer_view_from_state(
-    id: &ChitchatId,
-    state: &chitchat::NodeState,
-    alive: bool,
-) -> PeerView {
+fn peer_view_from_state(id: &ChitchatId, state: &chitchat::NodeState, alive: bool) -> PeerView {
     // Fall back to the chitchat node_id when the peer hasn't published a
     // separate NODE_NAME yet (should be almost never, but keeps startup
     // races graceful).
@@ -369,7 +361,11 @@ fn peer_view_from_state(
             s.split(',')
                 .filter_map(|p| {
                     let t = p.trim();
-                    if t.is_empty() { None } else { Some(t.to_string()) }
+                    if t.is_empty() {
+                        None
+                    } else {
+                        Some(t.to_string())
+                    }
                 })
                 .collect()
         })
@@ -415,11 +411,7 @@ mod tests {
     /// the failure detector has classified it as alive. Phi-accrual needs
     /// a handful of heartbeat samples before a newly-discovered node
     /// flips to live — this wait covers that ramp-up.
-    async fn wait_until_peer_alive(
-        gossip: &ClusterGossip,
-        name: &str,
-        deadline: Duration,
-    ) -> bool {
+    async fn wait_until_peer_alive(gossip: &ClusterGossip, name: &str, deadline: Duration) -> bool {
         let start = Instant::now();
         loop {
             if let Some(v) = gossip.peer(name).await {
@@ -446,6 +438,7 @@ mod tests {
             bind_rpc_tailscale: None,
             tls: None,
             blob_store_root: None,
+            prom_bind: None,
         };
         let g = ClusterGossip::bootstrap(&cfg, "solo").await.unwrap();
         let id = g.self_chitchat_id().await;
@@ -465,6 +458,7 @@ mod tests {
             bind_rpc_tailscale: None,
             tls: None,
             blob_store_root: None,
+            prom_bind: None,
         };
         let err = ClusterGossip::bootstrap(&cfg, "")
             .await
@@ -484,10 +478,14 @@ mod tests {
             bind_rpc_tailscale: None,
             tls: None,
             blob_store_root: None,
+            prom_bind: None,
         };
         // ClusterConfig::validate rejects this first — that's what we want:
         // the daemon should refuse to bootstrap gossip on a malformed config.
-        let err_chain = format!("{:#}", ClusterGossip::bootstrap(&cfg, "solo").await.unwrap_err());
+        let err_chain = format!(
+            "{:#}",
+            ClusterGossip::bootstrap(&cfg, "solo").await.unwrap_err()
+        );
         assert!(
             err_chain.contains("no bind address"),
             "expected 'no bind address' in error chain, got: {err_chain}"
@@ -511,6 +509,7 @@ mod tests {
             bind_rpc_tailscale: None,
             tls: None,
             blob_store_root: None,
+            prom_bind: None,
         };
         // Node B: uses A as seed.
         let cfg_b = ClusterConfig {
@@ -527,6 +526,7 @@ mod tests {
             bind_rpc_tailscale: None,
             tls: None,
             blob_store_root: None,
+            prom_bind: None,
         };
 
         let gossip_a = ClusterGossip::bootstrap(&cfg_a, "a").await.unwrap();
@@ -543,11 +543,9 @@ mod tests {
         // peers as live. Allow a generous 10s since phi-accrual needs a
         // handful of heartbeat samples (500ms gossip interval → typically
         // 2-4s to flip to alive on first sight).
-        let a_sees_b_alive =
-            wait_until_peer_alive(&gossip_a, "b", Duration::from_secs(10)).await;
+        let a_sees_b_alive = wait_until_peer_alive(&gossip_a, "b", Duration::from_secs(10)).await;
         assert!(a_sees_b_alive, "A should see B alive within 10s");
-        let b_sees_a_alive =
-            wait_until_peer_alive(&gossip_b, "a", Duration::from_secs(10)).await;
+        let b_sees_a_alive = wait_until_peer_alive(&gossip_b, "a", Duration::from_secs(10)).await;
         assert!(b_sees_a_alive, "B should see A alive within 10s");
 
         // Verify the state B sees for A matches what A published. RPC
@@ -564,10 +562,7 @@ mod tests {
         assert_eq!(b_view_of_a.hot_max_bytes, Some(1_000_000));
         assert_eq!(
             b_view_of_a.warm_projects,
-            vec![
-                "osobh/clawverse".to_string(),
-                "osobh/clawmates".to_string(),
-            ]
+            vec!["osobh/clawverse".to_string(), "osobh/clawmates".to_string(),]
         );
         assert_eq!(b_view_of_a.hot_fill_ratio(), Some(1024.0 / 1_000_000.0));
 
@@ -599,6 +594,7 @@ mod tests {
             bind_rpc_tailscale: None,
             tls: None,
             blob_store_root: None,
+            prom_bind: None,
         };
         let g = ClusterGossip::bootstrap(&cfg, "solo").await.unwrap();
         // Solo cluster — peers() must never include self.
@@ -626,6 +622,7 @@ mod tests {
             bind_rpc_tailscale: None,
             tls: None,
             blob_store_root: None,
+            prom_bind: None,
         };
         let cfg_b = ClusterConfig {
             zone: "lan-1g".into(),
@@ -641,6 +638,7 @@ mod tests {
             bind_rpc_tailscale: None,
             tls: None,
             blob_store_root: None,
+            prom_bind: None,
         };
         let g_a = ClusterGossip::bootstrap(&cfg_a, "a").await.unwrap();
         let g_b = ClusterGossip::bootstrap(&cfg_b, "b").await.unwrap();
@@ -767,5 +765,4 @@ mod tests {
         };
         assert_eq!(half_full.hot_fill_ratio(), Some(0.5));
     }
-
 }

claw-store/src/cluster/metrics.rs

@@ -147,6 +147,115 @@ impl MetricsReply {
         }
     }
 
+    /// Phase 5j: encode this snapshot in the Prometheus text exposition
+    /// format (v0.0.4 — the same one every scraper understands). One
+    /// `# HELP`, one `# TYPE`, one sample line per metric.
+    ///
+    /// All counters are exposed as `counter` type (monotonically
+    /// increasing since `started_unix`); the timestamp itself is a
+    /// `gauge`. Line ordering is stable so text-diff comparisons on a
+    /// scrape work — useful for debugging without a Prometheus server.
+    pub fn to_prometheus(&self) -> String {
+        // Preallocate — the output is always small (~1 KiB) but avoids
+        // reallocs mid-format.
+        let mut out = String::with_capacity(1024);
+        fn counter(out: &mut String, name: &str, help: &str, value: u64) {
+            out.push_str("# HELP ");
+            out.push_str(name);
+            out.push(' ');
+            out.push_str(help);
+            out.push('\n');
+            out.push_str("# TYPE ");
+            out.push_str(name);
+            out.push_str(" counter\n");
+            out.push_str(name);
+            out.push(' ');
+            out.push_str(&value.to_string());
+            out.push('\n');
+        }
+        fn gauge(out: &mut String, name: &str, help: &str, value: u64) {
+            out.push_str("# HELP ");
+            out.push_str(name);
+            out.push(' ');
+            out.push_str(help);
+            out.push('\n');
+            out.push_str("# TYPE ");
+            out.push_str(name);
+            out.push_str(" gauge\n");
+            out.push_str(name);
+            out.push(' ');
+            out.push_str(&value.to_string());
+            out.push('\n');
+        }
+        gauge(
+            &mut out,
+            "clawstor_counters_started_unix",
+            "Unix seconds when this counter set started (reset on daemon restart).",
+            self.started_unix,
+        );
+        counter(
+            &mut out,
+            "clawstor_cache_get_ref_hits_total",
+            "GetRef lookups that returned a cached blob id.",
+            self.get_ref_hits,
+        );
+        counter(
+            &mut out,
+            "clawstor_cache_get_ref_misses_total",
+            "GetRef lookups that missed.",
+            self.get_ref_misses,
+        );
+        counter(
+            &mut out,
+            "clawstor_cache_get_tag_hits_total",
+            "GetTag lookups that returned a value.",
+            self.get_tag_hits,
+        );
+        counter(
+            &mut out,
+            "clawstor_cache_get_tag_misses_total",
+            "GetTag lookups that missed.",
+            self.get_tag_misses,
+        );
+        counter(
+            &mut out,
+            "clawstor_cache_blob_get_bytes_total",
+            "Bytes served from the local blob store via BlobGet/GetChunk.",
+            self.blob_get_bytes,
+        );
+        counter(
+            &mut out,
+            "clawstor_cache_blob_put_bytes_total",
+            "Bytes ingested into the local blob store via BlobPut.",
+            self.blob_put_bytes,
+        );
+        counter(
+            &mut out,
+            "clawstor_cache_get_chunk_hits_total",
+            "GetChunk requests that returned bytes.",
+            self.get_chunk_hits,
+        );
+        counter(
+            &mut out,
+            "clawstor_cache_get_chunk_misses_total",
+            "GetChunk requests for unknown chunks.",
+            self.get_chunk_misses,
+        );
+        counter(
+            &mut out,
+            "clawstor_cache_has_chunk_hits_total",
+            "HasChunk probes that found the chunk locally (dedup save).",
+            self.has_chunk_hits,
+        );
+        counter(
+            &mut out,
+            "clawstor_cache_has_chunk_misses_total",
+            "HasChunk probes for chunks not present locally.",
+            self.has_chunk_misses,
+        );
+        out
+    }
+
     /// Hit rate for `HasChunk` probes — the dominant signal for how
     /// much dedup we're saving in partial-sync operations.
     pub fn has_chunk_hit_rate(&self) -> Option<f64> {
@@ -226,10 +335,7 @@ mod tests {
         }
         m.record_get_ref_miss();
         let rate = m.snapshot().get_ref_hit_rate().unwrap();
-        assert!(
-            (rate - 0.75).abs() < 1e-9,
-            "expected 0.75, got {rate}"
-        );
+        assert!((rate - 0.75).abs() < 1e-9, "expected 0.75, got {rate}");
     }
 
     #[test]
@@ -243,6 +349,64 @@ mod tests {
         assert_eq!(round, original);
     }
 
+    #[test]
+    fn to_prometheus_emits_every_counter_with_correct_type() {
+        let m = CacheMetrics::new();
+        for _ in 0..7 {
+            m.record_get_ref_hit();
+        }
+        m.record_get_ref_miss();
+        m.record_get_tag_hit();
+        m.record_blob_get_bytes(4096);
+        m.record_blob_put_bytes(8192);
+        m.record_get_chunk_hit();
+        m.record_get_chunk_miss();
+        m.record_has_chunk_hit();
+        m.record_has_chunk_miss();
+
+        let text = m.snapshot().to_prometheus();
+
+        // Every metric name must appear as a `# TYPE` line so scrapers
+        // classify it correctly. Missing a TYPE line breaks Grafana.
+        for (name, kind) in [
+            ("clawstor_counters_started_unix", "gauge"),
+            ("clawstor_cache_get_ref_hits_total", "counter"),
+            ("clawstor_cache_get_ref_misses_total", "counter"),
+            ("clawstor_cache_get_tag_hits_total", "counter"),
+            ("clawstor_cache_get_tag_misses_total", "counter"),
+            ("clawstor_cache_blob_get_bytes_total", "counter"),
+            ("clawstor_cache_blob_put_bytes_total", "counter"),
+            ("clawstor_cache_get_chunk_hits_total", "counter"),
+            ("clawstor_cache_get_chunk_misses_total", "counter"),
+            ("clawstor_cache_has_chunk_hits_total", "counter"),
+            ("clawstor_cache_has_chunk_misses_total", "counter"),
+        ] {
+            let type_line = format!("# TYPE {name} {kind}");
+            assert!(
+                text.contains(&type_line),
+                "missing type line {type_line} in:\n{text}"
+            );
+        }
+
+        // Sample values land verbatim on their own line.
+        assert!(text.contains("clawstor_cache_get_ref_hits_total 7\n"));
+        assert!(text.contains("clawstor_cache_get_ref_misses_total 1\n"));
+        assert!(text.contains("clawstor_cache_blob_get_bytes_total 4096\n"));
+        assert!(text.contains("clawstor_cache_blob_put_bytes_total 8192\n"));
+    }
+
+    #[test]
+    fn to_prometheus_valid_when_all_counters_zero() {
+        // A fresh daemon should still expose a valid scrape — zero
+        // counters are meaningful (they let alerts fire on absence of
+        // change, not just presence of misses).
+        let text = CacheMetrics::new().snapshot().to_prometheus();
+        assert!(text.contains("clawstor_cache_get_ref_hits_total 0\n"));
+        assert!(text.contains("clawstor_cache_blob_get_bytes_total 0\n"));
+        // No trailing garbage or partial lines.
+        assert!(text.ends_with('\n'));
+    }
+
     #[test]
     fn snapshots_across_threads_are_consistent_up_to_relaxed_ordering() {
         // Fire many increments from N threads and verify the final

claw-store/src/cluster/prom.rs

@@ -0,0 +1,272 @@
+//! Prometheus `/metrics` endpoint (Phase 5j).
+//!
+//! Exposes the same [`CacheMetrics`] counters that back `GetMetrics`
+//! and gossip, but in the Prometheus text exposition format so any
+//! existing scraper can consume them without knowing anything about
+//! QUIC or chitchat.
+//!
+//! # Wire shape
+//!
+//! One GET route:
+//!
+//! ```text
+//! GET /metrics
+//! → 200 OK, content-type text/plain; version=0.0.4
+//!   <the counters>
+//! ```
+//!
+//! Every other path returns 404. The scrape is stateless — no auth,
+//! no session — because the endpoint is intended for `127.0.0.1`
+//! binding by default, or a Tailscale-only bind at operator choice.
+//! A reverse proxy in front is the recommended pattern for
+//! internet-exposed scrapers.
+
+use crate::cluster::metrics::CacheMetrics;
+use anyhow::{Context, Result};
+use axum::{extract::State, http::header, response::IntoResponse, routing::get, Router};
+use std::net::SocketAddr;
+use std::sync::Arc;
+use tokio::net::TcpListener;
+use tokio::sync::oneshot;
+use tokio::task::JoinHandle;
+
+/// Prometheus text exposition v0.0.4 content-type. Every scraper in
+/// the wild parses this MIME correctly — a plain `text/plain` also
+/// works but the versioned string is the canonical value.
+const PROM_CONTENT_TYPE: &str = "text/plain; version=0.0.4; charset=utf-8";
+
+/// Running Prometheus HTTP server. Drop / [`PromServer::shutdown`] cancels
+/// the underlying accept loop and closes the listening socket.
+pub struct PromServer {
+    /// Address we actually bound. Copies of `SocketAddr` are cheap;
+    /// exposing it lets callers log or tests connect without a race.
+    bind_addr: SocketAddr,
+    task: JoinHandle<()>,
+    shutdown_tx: Option<oneshot::Sender<()>>,
+}
+
+impl std::fmt::Debug for PromServer {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("PromServer")
+            .field("bind_addr", &self.bind_addr)
+            .finish_non_exhaustive()
+    }
+}
+
+impl PromServer {
+    /// Bind the HTTP server, spawn the accept loop, return. The
+    /// returned handle is the sole owner of the background task; drop
+    /// it (or call [`Self::shutdown`]) to stop serving.
+    ///
+    /// `metrics` is shared with the RPC router so both endpoints
+    /// observe the same counters — no risk of drift between the
+    /// `GetMetrics` RPC and a scrape landing at the same instant.
+    pub async fn bind(addr: SocketAddr, metrics: Arc<CacheMetrics>) -> Result<Self> {
+        let listener = TcpListener::bind(addr)
+            .await
+            .with_context(|| format!("binding prometheus listener at {addr}"))?;
+        let bind_addr = listener
+            .local_addr()
+            .context("reading prometheus listener bound address")?;
+        let router = Router::new()
+            .route("/metrics", get(metrics_handler))
+            .with_state(metrics);
+        let (tx, rx) = oneshot::channel::<()>();
+        let task = tokio::spawn(async move {
+            let shutdown = async move {
+                let _ = rx.await;
+            };
+            if let Err(e) = axum::serve(listener, router)
+                .with_graceful_shutdown(shutdown)
+                .await
+            {
+                tracing::warn!(error = %e, "prometheus server exited with error");
+            }
+        });
+        Ok(Self {
+            bind_addr,
+            task,
+            shutdown_tx: Some(tx),
+        })
+    }
+
+    /// The address the server actually bound to. Differs from the
+    /// requested `addr` only when the caller passed a `0` port
+    /// (OS-assigned) — useful in tests.
+    pub fn local_addr(&self) -> SocketAddr {
+        self.bind_addr
+    }
+
+    /// Fire the graceful-shutdown signal and await the accept task.
+    /// Idempotent — a second call after the receiver's already been
+    /// dropped is a no-op.
+    pub async fn shutdown(mut self) {
+        if let Some(tx) = self.shutdown_tx.take() {
+            let _ = tx.send(());
+        }
+        // Best-effort await. If it panicked we've already logged.
+        let _ = self.task.await;
+    }
+
+    /// Abort the accept task without waiting. Used by `ClusterServices`
+    /// during a synchronous drop where awaiting isn't an option.
+    pub fn abort(self) {
+        if let Some(tx) = self.shutdown_tx {
+            let _ = tx.send(());
+        }
+        self.task.abort();
+    }
+}
+
+/// `GET /metrics` handler — snapshot the counters, format, return.
+///
+/// Snapshotting on every scrape (rather than caching for N ms) keeps
+/// counters as fresh as possible; the underlying `Ordering::Relaxed`
+/// atomic loads are cheap enough that even a 1 Hz scrape cadence is
+/// well under a microsecond of CPU per request.
+async fn metrics_handler(State(metrics): State<Arc<CacheMetrics>>) -> impl IntoResponse {
+    let body = metrics.snapshot().to_prometheus();
+    ([(header::CONTENT_TYPE, PROM_CONTENT_TYPE)], body)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio::net::TcpStream;
+
+    /// Raw HTTP/1.1 client: writes one GET, reads the entire response
+    /// until connection close, and splits headers from body on the
+    /// double CRLF. Returns `(status, content_type, body)`.
+    ///
+    /// The server closes each response's TCP connection (default for a
+    /// non-keep-alive request), so a bounded read-to-end is safe here.
+    async fn scrape(addr: SocketAddr, path: &str) -> (u16, String, String) {
+        let mut sock = TcpStream::connect(addr).await.unwrap();
+        let req = format!("GET {path} HTTP/1.1\r\nHost: {addr}\r\nConnection: close\r\n\r\n");
+        sock.write_all(req.as_bytes()).await.unwrap();
+        let mut buf = Vec::with_capacity(4096);
+        sock.read_to_end(&mut buf).await.unwrap();
+        let raw = String::from_utf8_lossy(&buf).to_string();
+        let (head, body) = raw.split_once("\r\n\r\n").unwrap_or((&raw, ""));
+        let status = head
+            .lines()
+            .next()
+            .and_then(|l| l.split_whitespace().nth(1))
+            .and_then(|s| s.parse().ok())
+            .unwrap_or(0);
+        let ct = head
+            .lines()
+            .find_map(|l| {
+                let (k, v) = l.split_once(':')?;
+                if k.eq_ignore_ascii_case("content-type") {
+                    Some(v.trim().to_string())
+                } else {
+                    None
+                }
+            })
+            .unwrap_or_default();
+        (status, ct, body.to_string())
+    }
+
+    #[tokio::test]
+    async fn scrape_returns_prometheus_text_with_correct_content_type() {
+        let m = Arc::new(CacheMetrics::new());
+        for _ in 0..3 {
+            m.record_get_ref_hit();
+        }
+        m.record_get_ref_miss();
+        m.record_blob_get_bytes(1024);
+        let bind = "127.0.0.1:0".parse().unwrap();
+        let server = PromServer::bind(bind, m.clone()).await.unwrap();
+        let addr = server.local_addr();
+        assert_ne!(addr.port(), 0, "kernel must assign a real port");
+
+        let (status, ct, body) = scrape(addr, "/metrics").await;
+        assert_eq!(status, 200);
+        assert!(ct.starts_with("text/plain"), "unexpected content-type {ct}");
+        // Exposition body carries the counters at the values we set.
+        // Body may be chunked (transfer-encoding: chunked) — check via
+        // substring rather than exact equality.
+        assert!(
+            body.contains("clawstor_cache_get_ref_hits_total 3"),
+            "body missing get_ref_hits=3:\n{body}"
+        );
+        assert!(
+            body.contains("clawstor_cache_get_ref_misses_total 1"),
+            "body missing get_ref_misses=1:\n{body}"
+        );
+        assert!(
+            body.contains("clawstor_cache_blob_get_bytes_total 1024"),
+            "body missing blob_get_bytes=1024:\n{body}"
+        );
+        server.shutdown().await;
+    }
+
+    #[tokio::test]
+    async fn scrape_shows_live_updates_between_requests() {
+        // Prove there's no cache — a second scrape after an increment
+        // must show the new value.
+        let m = Arc::new(CacheMetrics::new());
+        let bind = "127.0.0.1:0".parse().unwrap();
+        let server = PromServer::bind(bind, m.clone()).await.unwrap();
+        let addr = server.local_addr();
+
+        let (_, _, first) = scrape(addr, "/metrics").await;
+        assert!(first.contains("clawstor_cache_get_ref_hits_total 0"));
+
+        for _ in 0..5 {
+            m.record_get_ref_hit();
+        }
+        let (_, _, second) = scrape(addr, "/metrics").await;
+        assert!(
+            second.contains("clawstor_cache_get_ref_hits_total 5"),
+            "expected 5 after increments:\n{second}"
+        );
+        server.shutdown().await;
+    }
+
+    #[tokio::test]
+    async fn unknown_paths_return_404() {
+        // Only /metrics is a valid path; anything else is a 404 (no
+        // catch-all router that would tempt someone into treating this
+        // as a general HTTP server).
+        let m = Arc::new(CacheMetrics::new());
+        let bind = "127.0.0.1:0".parse().unwrap();
+        let server = PromServer::bind(bind, m.clone()).await.unwrap();
+        let addr = server.local_addr();
+        let (status, _, _) = scrape(addr, "/not-a-thing").await;
+        assert_eq!(status, 404);
+        server.shutdown().await;
+    }
+
+    #[tokio::test]
+    async fn shutdown_stops_serving() {
+        // After shutdown() the port must refuse new connections.
+        let m = Arc::new(CacheMetrics::new());
+        let bind = "127.0.0.1:0".parse().unwrap();
+        let server = PromServer::bind(bind, m.clone()).await.unwrap();
+        let addr = server.local_addr();
+        server.shutdown().await;
+        // Give the OS a beat to release the socket.
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        // Either the connect fails or the connection is closed with 0
+        // bytes read — both count as "not serving".
+        let res = tokio::time::timeout(
+            std::time::Duration::from_millis(500),
+            TcpStream::connect(addr),
+        )
+        .await;
+        match res {
+            Err(_) => { /* timeout — good */ }
+            Ok(Err(_)) => { /* connect refused — good */ }
+            Ok(Ok(mut sock)) => {
+                // Connected — but a read must return 0 bytes because
+                // the accept loop is gone.
+                let mut buf = [0u8; 16];
+                let n = sock.read(&mut buf).await.unwrap_or(0);
+                assert_eq!(n, 0, "server accepted a connection after shutdown");
+            }
+        }
+    }
+}

claw-store/src/cluster/services.rs

@@ -14,6 +14,7 @@
 
 use crate::cluster::blob::BlobStore;
 use crate::cluster::gossip::ClusterGossip;
+use crate::cluster::prom::PromServer;
 use crate::cluster::refs::RefStore;
 use crate::cluster::rpc::{serve_connection, RpcRouter};
 use crate::cluster::tags::TagStore;
@@ -75,6 +76,10 @@ pub struct ClusterServices {
     /// Phase 5i: periodic cache-metric publisher. `None` when RPC is
     /// not running (no router → no counters to publish).
     cache_metric_task: Option<JoinHandle<()>>,
+    /// Phase 5j: Prometheus `/metrics` HTTP server. `None` when
+    /// `cluster.prom_bind` was absent or no router exists (nothing to
+    /// scrape).
+    prom_server: Option<PromServer>,
 }
 
 impl std::fmt::Debug for ClusterServices {
@@ -161,8 +166,7 @@ impl ClusterServices {
         // present, so an in-process caller (dashboard, tests) can hold
         // it. But we only spawn the accept loop when TLS is up.
         let router: Option<Arc<RpcRouter>> = if cluster.tls.is_some() {
-            let mut r =
-                RpcRouter::new(gossip.clone(), local_name.clone(), cluster.zone.clone());
+            let mut r = RpcRouter::new(gossip.clone(), local_name.clone(), cluster.zone.clone());
             if let Some(store) = &blob_store {
                 r = r.with_blob_store(store.clone());
             }
@@ -186,8 +190,7 @@ impl ClusterServices {
                     .rpc_lan()
                     .or_else(|| cluster.rpc_tailscale())
                     .context("no RPC bind address (need bind_lan or bind_tailscale)")?;
-                let server = QuicServer::bind(bind, identity)
-                    .context("binding QUIC RPC server")?;
+                let server = QuicServer::bind(bind, identity).context("binding QUIC RPC server")?;
                 let router = router.clone();
                 tracing::info!("cluster RPC server listening on {}", bind);
                 Some(tokio::spawn(async move {
@@ -237,6 +240,28 @@ impl ClusterServices {
             None
         };
 
+        // Phase 5j: Prometheus `/metrics` server. Only meaningful when
+        // we have a router — otherwise there are no counters to expose.
+        // `prom_bind` unset => no server, no port opened.
+        let prom_server = match (&router, cluster.prom_bind) {
+            (Some(router), Some(addr)) => {
+                let server = PromServer::bind(addr, router.metrics().clone())
+                    .await
+                    .with_context(|| format!("binding Prometheus /metrics at {addr}"))?;
+                tracing::info!("Prometheus /metrics listening on {}", server.local_addr());
+                Some(server)
+            }
+            (Some(_), None) => {
+                tracing::info!("cluster: prom_bind unset; Prometheus endpoint disabled");
+                None
+            }
+            (None, Some(_)) => {
+                tracing::warn!("cluster: prom_bind set but RPC not running; nothing to scrape");
+                None
+            }
+            (None, None) => None,
+        };
+
         Ok(Self {
             gossip,
             blob_store,
@@ -246,9 +271,16 @@ impl ClusterServices {
             accept_task,
             metric_task,
             cache_metric_task,
+            prom_server,
         })
     }
 
+    /// Address the Prometheus `/metrics` server actually bound to.
+    /// `None` when the server was never started.
+    pub fn prom_addr(&self) -> Option<std::net::SocketAddr> {
+        self.prom_server.as_ref().map(|s| s.local_addr())
+    }
+
     /// Whether a local blob store was configured at start time.
     pub fn blob_store_enabled(&self) -> bool {
         self.blob_store.is_some()
@@ -276,6 +308,9 @@ impl ClusterServices {
         if let Some(task) = self.cache_metric_task {
             task.abort();
         }
+        if let Some(server) = self.prom_server {
+            server.abort();
+        }
     }
 }
 
@@ -405,7 +440,10 @@ mod tests {
         .await
         .unwrap();
         assert!(!svc.rpc_enabled(), "no [cluster.tls] → RPC disabled");
-        assert!(!svc.blob_store_enabled(), "no blob root → blob store disabled");
+        assert!(
+            !svc.blob_store_enabled(),
+            "no blob root → blob store disabled"
+        );
         // Gossip must still be functional.
         let self_id = svc.gossip.self_chitchat_id().await;
         assert_eq!(self_id.node_id.as_ref(), "test-node");
@@ -444,8 +482,7 @@ mod tests {
         std::fs::create_dir_all(&hot_dir).unwrap();
         std::fs::write(hot_dir.join("blob"), vec![0u8; 4096]).unwrap();
 
-        let svc =
-            ClusterServices::start(&cfg_a, "a".into(), hot_dir.clone(), 1_000_000, None)
+        let svc = ClusterServices::start(&cfg_a, "a".into(), hot_dir.clone(), 1_000_000, None)
             .await
             .unwrap();
         assert!(svc.rpc_enabled());
@@ -532,12 +569,22 @@ mod tests {
             ..Default::default()
         };
         let tmp = tempfile::TempDir::new().unwrap();
-        let svc_a =
-            ClusterServices::start(&cfg_a, "a".into(), tmp.path().to_path_buf(), 1_000_000, None)
+        let svc_a = ClusterServices::start(
+            &cfg_a,
+            "a".into(),
+            tmp.path().to_path_buf(),
+            1_000_000,
+            None,
+        )
         .await
         .unwrap();
-        let svc_b =
-            ClusterServices::start(&cfg_b, "b".into(), tmp.path().to_path_buf(), 1_000_000, None)
+        let svc_b = ClusterServices::start(
+            &cfg_b,
+            "b".into(),
+            tmp.path().to_path_buf(),
+            1_000_000,
+            None,
+        )
         .await
         .unwrap();
 
@@ -558,6 +605,90 @@ mod tests {
         svc_b.shutdown();
     }
 
+    #[tokio::test]
+    async fn services_expose_prometheus_endpoint_when_configured() {
+        // Phase 5j: cluster.prom_bind is honoured, the endpoint serves
+        // live counters that reflect in-process router activity.
+        use crate::cluster::transport::FleetCa;
+        use crate::config::ClusterTlsConfig;
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+        use tokio::net::TcpStream;
+
+        let tmp = tempfile::TempDir::new().unwrap();
+        let ca_dir = tmp.path().join("ca");
+        let a_tls_dir = tmp.path().join("a-tls");
+        let ca = FleetCa::generate("test CA").unwrap();
+        ca.save(&ca_dir).unwrap();
+        ca.sign_leaf_to_pem("a", &a_tls_dir).unwrap();
+
+        let cfg = ClusterConfig {
+            zone: "fabric-10g".into(),
+            bind_lan: Some(loopback(next_port())),
+            tls: Some(ClusterTlsConfig {
+                ca_cert: a_tls_dir.join("ca.crt"),
+                node_cert: a_tls_dir.join("node.crt"),
+                node_key: a_tls_dir.join("node.key"),
+            }),
+            prom_bind: Some("127.0.0.1:0".parse().unwrap()),
+            ..Default::default()
+        };
+        let hot_dir = tmp.path().join("hot");
+        std::fs::create_dir_all(&hot_dir).unwrap();
+
+        let svc = ClusterServices::start(&cfg, "a".into(), hot_dir, 1_000_000, None)
+            .await
+            .unwrap();
+        let prom = svc.prom_addr().expect("prom server bound");
+        assert_ne!(prom.port(), 0, "kernel must assign a real port");
+
+        // Drive some counters directly through the router.
+        let router = svc.router.as_ref().unwrap();
+        for _ in 0..4 {
+            router.metrics().record_get_ref_hit();
+        }
+        router.metrics().record_blob_get_bytes(2048);
+
+        // Scrape.
+        let mut sock = TcpStream::connect(prom).await.unwrap();
+        let req = format!("GET /metrics HTTP/1.1\r\nHost: {prom}\r\nConnection: close\r\n\r\n");
+        sock.write_all(req.as_bytes()).await.unwrap();
+        let mut buf = Vec::new();
+        sock.read_to_end(&mut buf).await.unwrap();
+        let body = String::from_utf8_lossy(&buf);
+        assert!(
+            body.contains("clawstor_cache_get_ref_hits_total 4"),
+            "expected 4 hits in scrape: {body}"
+        );
+        assert!(
+            body.contains("clawstor_cache_blob_get_bytes_total 2048"),
+            "expected 2048 blob GET bytes: {body}"
+        );
+        svc.shutdown();
+    }
+
+    #[tokio::test]
+    async fn services_leave_prom_server_off_when_unconfigured() {
+        // Absence of prom_bind means no HTTP server; prom_addr() is
+        // None. Guards against accidentally leaking a metrics port.
+        let cfg = ClusterConfig {
+            zone: "fabric-10g".into(),
+            bind_lan: Some(loopback(next_port())),
+            ..Default::default()
+        };
+        let tmp = tempfile::TempDir::new().unwrap();
+        let svc = ClusterServices::start(
+            &cfg,
+            "quiet".into(),
+            tmp.path().to_path_buf(),
+            1_000_000,
+            None,
+        )
+        .await
+        .unwrap();
+        assert!(svc.prom_addr().is_none(), "prom server must not start");
+        svc.shutdown();
+    }
+
     #[tokio::test]
     async fn services_publish_cache_metrics_to_gossip_on_start() {
         // Phase 5i: when RPC is up, the initial cache-metric publish
@@ -601,26 +732,14 @@ mod tests {
         let hot_dir = tmp.path().join("hot");
         std::fs::create_dir_all(&hot_dir).unwrap();
 
-        let svc_a = ClusterServices::start(
-            &cfg_a,
-            "a".into(),
-            hot_dir.clone(),
-            1_000_000,
-            None,
-        )
+        let svc_a = ClusterServices::start(&cfg_a, "a".into(), hot_dir.clone(), 1_000_000, None)
             .await
             .unwrap();
         // Drive counters BEFORE start() would be a chicken-and-egg —
         // instead, drive them AFTER start and check the periodic
         // republish. First, verify the initial-publish (all zeros) has
         // reached B: the four keys should be Some(0), not None.
-        let svc_b = ClusterServices::start(
-            &cfg_b,
-            "b".into(),
-            hot_dir,
-            1_000_000,
-            None,
-        )
+        let svc_b = ClusterServices::start(&cfg_b, "b".into(), hot_dir, 1_000_000, None)
             .await
             .unwrap();
 

claw-store/src/config.rs

@@ -6,7 +6,10 @@ use std::path::{Path, PathBuf};
 
 #[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
 #[serde(rename_all = "lowercase")]
-pub enum NodeRole { Primary, Secondary }
+pub enum NodeRole {
+    Primary,
+    Secondary,
+}
 
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct NodeConfig {
@@ -143,6 +146,14 @@ pub struct ClusterConfig {
     /// gossip + peer-status but returns `NotConfigured` for Blob RPCs.
     #[serde(default)]
     pub blob_store_root: Option<PathBuf>,
+    /// Phase 5j: optional bind address for the Prometheus `/metrics`
+    /// endpoint. When set, the daemon spins up a tiny HTTP server on
+    /// this address serving the same counters exposed via `GetMetrics`
+    /// and gossip. Typical value is `127.0.0.1:7702` (Prometheus scrapes
+    /// via the LAN listener the node advertises to its scrape target
+    /// group). Absent means "no scrape endpoint".
+    #[serde(default)]
+    pub prom_bind: Option<SocketAddr>,
 }
 
 /// Compute the default RPC address for a gossip address: same IP, port + 1.
@@ -182,7 +193,8 @@ impl ClusterConfig {
     /// explicit `bind_rpc_lan` override; otherwise derives from `bind_lan`
     /// with port + 1.
     pub fn rpc_lan(&self) -> Option<SocketAddr> {
-        self.bind_rpc_lan.or_else(|| self.bind_lan.and_then(default_rpc_addr))
+        self.bind_rpc_lan
+            .or_else(|| self.bind_lan.and_then(default_rpc_addr))
     }
 
     /// The Tailscale address this node listens on for RPC (QUIC).
@@ -367,10 +379,7 @@ tailscale_addr = "100.64.1.5:7701"
         );
 
         let laptop = cluster.peer("laptop").expect("laptop peer");
-        assert!(
-            laptop.lan_addr.is_none(),
-            "roaming peer has no LAN address"
-        );
+        assert!(laptop.lan_addr.is_none(), "roaming peer has no LAN address");
         assert!(laptop.tailscale_addr.is_some());
 
         cluster.validate().expect("valid cluster config");
@@ -402,12 +411,10 @@ tailscale_addr = "100.64.1.5:7701"
             bind_rpc_tailscale: None,
             tls: None,
             blob_store_root: None,
+            prom_bind: None,
         };
         let err = cluster.validate().unwrap_err().to_string();
-        assert!(
-            err.contains("no bind address"),
-            "unexpected error: {err}"
-        );
+        assert!(err.contains("no bind address"), "unexpected error: {err}");
     }
 
     #[test]
@@ -434,6 +441,7 @@ tailscale_addr = "100.64.1.5:7701"
             bind_rpc_tailscale: None,
             tls: None,
             blob_store_root: None,
+            prom_bind: None,
         };
         let err = cluster.validate().unwrap_err().to_string();
         assert!(