claw-store/src/cluster.rs

@@ -15,6 +15,8 @@
 //! propagate without a daemon restart.
 
 pub mod gossip;
+pub mod rpc;
+pub mod services;
 pub mod transport;
 
 use crate::config::PeerEntry;

claw-store/src/cluster/gossip.rs

@@ -75,7 +75,10 @@ const DEFAULT_MARKED_FOR_DELETION_GRACE: Duration = Duration::from_secs(60);
 /// time. Fields other than `name` and `zone` may be `None` if the peer
 /// hasn't advertised them yet (early bootstrap) or has never had them
 /// (e.g. a laptop with no LAN address).
-#[derive(Debug, Clone, PartialEq)]
+///
+/// `Serialize`/`Deserialize` let peers shuttle their local view over
+/// the RPC layer — see `cluster::rpc::Method::PeerStatus`.
+#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
 pub struct PeerView {
     pub name: String,
     pub zone: String,

claw-store/src/cluster/rpc.rs

@@ -0,0 +1,510 @@
+//! Peer RPC protocol on top of the QUIC transport (Phase 1e).
+//!
+//! Every bidi stream carries one request → one response. The first byte
+//! of the request is a method tag from [`Method`]; the rest is the
+//! opaque per-method payload. The response is the opaque per-method
+//! reply, or a single-byte error code from [`ErrorCode`] when the
+//! request was malformed.
+//!
+//! # Wire format
+//!
+//! ```text
+//! request  : method:u8 | payload:bytes
+//! response : reply:bytes   -- or single-byte ErrorCode
+//! ```
+//!
+//! The QUIC transport already provides message-boundary + integrity, so
+//! no length-prefixing or checksums live here — `read_to_end` on a
+//! finished stream returns exactly one whole message.
+//!
+//! # Methods
+//!
+//! * [`Method::Ping`] — echoes the payload back as `"pong:" || payload`.
+//!   Health check / handshake smoke test.
+//! * [`Method::PeerStatus`] — returns a JSON-encoded [`PeerStatusReply`]
+//!   containing this node's local view of the cluster (its own
+//!   name+zone, plus every peer it currently knows about via gossip).
+
+use crate::cluster::gossip::{ClusterGossip, PeerView};
+use anyhow::{bail, Context, Result};
+use quinn::{Connection, ConnectionError};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+/// Cap on a single request or response, including the method tag.
+/// Matches `MAX_MESSAGE_BYTES` in [`super::transport`] so both layers
+/// bound memory the same way.
+pub const MAX_MESSAGE_BYTES: usize = 16 * 1024;
+
+/// RPC method tag byte.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Method {
+    Ping = 0x01,
+    PeerStatus = 0x02,
+}
+
+impl Method {
+    /// Parse a byte back into a method. Unknown bytes → `None`, which
+    /// the server surfaces to the caller as [`ErrorCode::UnknownMethod`].
+    pub fn from_byte(b: u8) -> Option<Self> {
+        match b {
+            0x01 => Some(Method::Ping),
+            0x02 => Some(Method::PeerStatus),
+            _ => None,
+        }
+    }
+
+    /// Byte tag as an owned u8. `as u8` also works; this exists for symmetry.
+    pub fn as_byte(self) -> u8 {
+        self as u8
+    }
+}
+
+/// Well-known single-byte error responses the server may return in
+/// place of a normal reply. The client distinguishes these by length =
+/// 1 AND the byte being a known error code.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum ErrorCode {
+    EmptyRequest = 0xf0,
+    UnknownMethod = 0xf1,
+    HandlerFailure = 0xf2,
+}
+
+impl ErrorCode {
+    pub fn as_byte(self) -> u8 {
+        self as u8
+    }
+    pub fn describe(self) -> &'static str {
+        match self {
+            ErrorCode::EmptyRequest => "empty request",
+            ErrorCode::UnknownMethod => "unknown method",
+            ErrorCode::HandlerFailure => "handler failure",
+        }
+    }
+}
+
+/// Reply payload for [`Method::PeerStatus`]. Serialised as JSON on the
+/// wire — small (<< 16 KB for a 10-node fleet) and easy to inspect
+/// from a shell (`jq` etc.).
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct PeerStatusReply {
+    /// The name this node advertises for itself.
+    pub local_name: String,
+    /// The zone this node is in.
+    pub local_zone: String,
+    /// Every peer this node knows about (live + dead-in-grace-window).
+    pub peers: Vec<PeerView>,
+}
+
+/// The concrete RPC handler used by the daemon. Holds Arc references
+/// to the state a request might need to read (currently just the
+/// gossip service; Phase 2+ adds the blob store, metadata index, etc).
+///
+/// Not `Clone` on its own — wrap in `Arc<RpcRouter>` so a single
+/// instance backs the accept loop plus any explicit dispatch calls.
+pub struct RpcRouter {
+    gossip: Arc<ClusterGossip>,
+    local_name: String,
+    local_zone: String,
+}
+
+impl RpcRouter {
+    pub fn new(gossip: Arc<ClusterGossip>, local_name: String, local_zone: String) -> Self {
+        Self {
+            gossip,
+            local_name,
+            local_zone,
+        }
+    }
+
+    /// Dispatch a single request. Called by [`serve_connection`] for
+    /// every accepted bidi stream. Test code may call it directly to
+    /// bypass the transport.
+    pub async fn handle(&self, method: Method, payload: &[u8]) -> Result<Vec<u8>> {
+        match method {
+            Method::Ping => {
+                let mut reply = Vec::with_capacity(5 + payload.len());
+                reply.extend_from_slice(b"pong:");
+                reply.extend_from_slice(payload);
+                Ok(reply)
+            }
+            Method::PeerStatus => {
+                let peers = self.gossip.peers().await;
+                let reply = PeerStatusReply {
+                    local_name: self.local_name.clone(),
+                    local_zone: self.local_zone.clone(),
+                    peers,
+                };
+                let json = serde_json::to_vec(&reply)
+                    .context("encoding PeerStatusReply as JSON")?;
+                if json.len() > MAX_MESSAGE_BYTES {
+                    bail!(
+                        "PeerStatusReply JSON {} bytes exceeds cap {}",
+                        json.len(),
+                        MAX_MESSAGE_BYTES
+                    );
+                }
+                Ok(json)
+            }
+        }
+    }
+}
+
+/// Server-side: loop accepting bidi streams on `conn`, dispatch to
+/// `router`, write the reply. Returns cleanly when the peer closes the
+/// connection.
+pub async fn serve_connection(conn: Connection, router: Arc<RpcRouter>) -> Result<()> {
+    loop {
+        let (mut send, mut recv) = match conn.accept_bi().await {
+            Ok(pair) => pair,
+            Err(ConnectionError::ApplicationClosed(_))
+            | Err(ConnectionError::ConnectionClosed(_))
+            | Err(ConnectionError::LocallyClosed)
+            | Err(ConnectionError::TimedOut) => return Ok(()),
+            Err(e) => return Err(anyhow::Error::from(e)),
+        };
+        let request = recv
+            .read_to_end(MAX_MESSAGE_BYTES)
+            .await
+            .context("reading RPC request")?;
+        let reply = dispatch(&router, &request).await;
+        send.write_all(&reply)
+            .await
+            .context("writing RPC reply")?;
+        send.finish().context("finishing RPC send stream")?;
+    }
+}
+
+/// Turn a raw wire-format request into a response — either the
+/// router's real reply or a single-byte error code. Extracted so tests
+/// can hit it without a QUIC connection.
+async fn dispatch(router: &RpcRouter, request: &[u8]) -> Vec<u8> {
+    if request.is_empty() {
+        return vec![ErrorCode::EmptyRequest.as_byte()];
+    }
+    let method = match Method::from_byte(request[0]) {
+        Some(m) => m,
+        None => return vec![ErrorCode::UnknownMethod.as_byte()],
+    };
+    let payload = &request[1..];
+    match router.handle(method, payload).await {
+        Ok(reply) => reply,
+        Err(e) => {
+            tracing::warn!(error = %e, method = ?method, "RPC handler failed");
+            vec![ErrorCode::HandlerFailure.as_byte()]
+        }
+    }
+}
+
+/// Client-side: open a bidi stream, write `method || payload`, read
+/// reply. Returns the raw reply bytes; callers deserialise per method.
+pub async fn rpc_call(
+    conn: &Connection,
+    method: Method,
+    payload: &[u8],
+) -> Result<Vec<u8>> {
+    if payload.len() + 1 > MAX_MESSAGE_BYTES {
+        bail!(
+            "RPC payload {} bytes (+1 for method tag) exceeds cap {}",
+            payload.len(),
+            MAX_MESSAGE_BYTES
+        );
+    }
+    let (mut send, mut recv) = conn
+        .open_bi()
+        .await
+        .context("opening bidi stream for RPC")?;
+    let mut buf = Vec::with_capacity(1 + payload.len());
+    buf.push(method.as_byte());
+    buf.extend_from_slice(payload);
+    send.write_all(&buf).await.context("writing RPC request")?;
+    send.finish().context("finishing RPC send stream")?;
+    let reply = recv
+        .read_to_end(MAX_MESSAGE_BYTES)
+        .await
+        .context("reading RPC reply")?;
+    Ok(reply)
+}
+
+/// Convenience wrapper for [`Method::Ping`]. Sends `payload`, returns
+/// the peer's echo (`"pong:" || payload`) with the prefix stripped.
+pub async fn call_ping(conn: &Connection, payload: &[u8]) -> Result<Vec<u8>> {
+    let reply = rpc_call(conn, Method::Ping, payload).await?;
+    if reply.len() == 1 {
+        if let Some(code) = decode_error(reply[0]) {
+            bail!("peer replied with error: {}", code.describe());
+        }
+    }
+    if let Some(rest) = reply.strip_prefix(b"pong:") {
+        Ok(rest.to_vec())
+    } else {
+        bail!(
+            "peer replied with unexpected shape: {} bytes, no 'pong:' prefix",
+            reply.len()
+        );
+    }
+}
+
+/// Convenience wrapper for [`Method::PeerStatus`]. Sends an empty
+/// payload, deserialises the JSON response.
+pub async fn call_peer_status(conn: &Connection) -> Result<PeerStatusReply> {
+    let reply = rpc_call(conn, Method::PeerStatus, &[]).await?;
+    if reply.len() == 1 {
+        if let Some(code) = decode_error(reply[0]) {
+            bail!("peer replied with error: {}", code.describe());
+        }
+    }
+    serde_json::from_slice(&reply).context("decoding PeerStatusReply JSON")
+}
+
+/// Recognise a single-byte reply as one of our error codes. Returns
+/// `None` for any other single-byte value (which is a valid reply,
+/// just an unusually short one).
+fn decode_error(b: u8) -> Option<ErrorCode> {
+    match b {
+        0xf0 => Some(ErrorCode::EmptyRequest),
+        0xf1 => Some(ErrorCode::UnknownMethod),
+        0xf2 => Some(ErrorCode::HandlerFailure),
+        _ => None,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::cluster::transport::{NodeIdentity, QuicClient, QuicServer};
+    use crate::config::{ClusterConfig, PeerEntry};
+    use std::net::SocketAddr;
+    use std::sync::atomic::{AtomicU16, Ordering};
+    use std::time::Duration;
+
+    /// Dedicated port range for RPC tests, distinct from gossip (41000+)
+    /// and transport (42000+) so parallel tests never conflict.
+    static NEXT_PORT: AtomicU16 = AtomicU16::new(43001);
+    fn next_port() -> u16 {
+        NEXT_PORT.fetch_add(1, Ordering::Relaxed)
+    }
+
+    fn loopback(port: u16) -> SocketAddr {
+        format!("127.0.0.1:{port}").parse().unwrap()
+    }
+
+    async fn bootstrap_gossip(name: &str, port: u16) -> Arc<ClusterGossip> {
+        let cfg = ClusterConfig {
+            zone: "fabric-10g".into(),
+            bind_lan: Some(loopback(port)),
+            ..Default::default()
+        };
+        Arc::new(ClusterGossip::bootstrap(&cfg, name).await.unwrap())
+    }
+
+    #[test]
+    fn method_round_trips_byte_encoding() {
+        assert_eq!(Method::Ping.as_byte(), 0x01);
+        assert_eq!(Method::PeerStatus.as_byte(), 0x02);
+        assert_eq!(Method::from_byte(0x01), Some(Method::Ping));
+        assert_eq!(Method::from_byte(0x02), Some(Method::PeerStatus));
+        assert_eq!(Method::from_byte(0x00), None);
+        assert_eq!(Method::from_byte(0xff), None);
+    }
+
+    #[tokio::test]
+    async fn dispatch_returns_pong_for_ping() {
+        let gossip = bootstrap_gossip("solo", next_port()).await;
+        let router = RpcRouter::new(gossip.clone(), "solo".into(), "test-zone".into());
+
+        // Direct dispatch — no network involved.
+        let reply = dispatch(&router, &[Method::Ping.as_byte(), b'h', b'i']).await;
+        assert_eq!(reply, b"pong:hi");
+        gossip.peer("nobody").await; // touch to keep gossip alive
+    }
+
+    #[tokio::test]
+    async fn dispatch_returns_json_for_peer_status() {
+        let gossip = bootstrap_gossip("architect", next_port()).await;
+        let router = RpcRouter::new(
+            gossip.clone(),
+            "architect".into(),
+            "fabric-10g".into(),
+        );
+        let reply = dispatch(&router, &[Method::PeerStatus.as_byte()]).await;
+        let decoded: PeerStatusReply = serde_json::from_slice(&reply).unwrap();
+        assert_eq!(decoded.local_name, "architect");
+        assert_eq!(decoded.local_zone, "fabric-10g");
+        // Solo node — no peers yet.
+        assert!(decoded.peers.is_empty(), "solo node has no peers");
+    }
+
+    #[tokio::test]
+    async fn dispatch_returns_empty_request_error() {
+        let gossip = bootstrap_gossip("solo", next_port()).await;
+        let router = RpcRouter::new(gossip.clone(), "solo".into(), "z".into());
+        let reply = dispatch(&router, &[]).await;
+        assert_eq!(reply, vec![ErrorCode::EmptyRequest.as_byte()]);
+    }
+
+    #[tokio::test]
+    async fn dispatch_returns_unknown_method_error() {
+        let gossip = bootstrap_gossip("solo", next_port()).await;
+        let router = RpcRouter::new(gossip.clone(), "solo".into(), "z".into());
+        let reply = dispatch(&router, &[0xab, 0x01, 0x02]).await;
+        assert_eq!(reply, vec![ErrorCode::UnknownMethod.as_byte()]);
+    }
+
+    #[tokio::test]
+    async fn rpc_call_rejects_oversize_payload() {
+        // No network involved — the size check runs client-side before we
+        // even try to open the bidi stream. Use a dummy connection built
+        // via generate_test_pair; we won't actually connect.
+        let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
+        let server = QuicServer::bind(loopback(0), id_b).unwrap();
+        let server_addr = server.local_addr().unwrap();
+        let _accept = tokio::spawn(async move {
+            let _ = server.accept().await;
+        });
+        let client = QuicClient::new(loopback(0), id_a).unwrap();
+        let conn = client.connect(server_addr, "b").await.unwrap();
+        let huge = vec![0u8; MAX_MESSAGE_BYTES];
+        let err = rpc_call(&conn, Method::Ping, &huge)
+            .await
+            .err()
+            .expect("must reject oversize");
+        assert!(err.to_string().contains("exceeds cap"));
+        conn.close(quinn::VarInt::from_u32(0), b"done");
+        client.shutdown().await;
+    }
+
+    #[tokio::test]
+    async fn end_to_end_ping_and_peer_status_over_real_quic() {
+        // Two full nodes: A runs gossip + a QuicServer serving RpcRouter.
+        // B is a client that dials A and calls both RPCs.
+        let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
+
+        let port_a = next_port();
+        let gossip_a = bootstrap_gossip("a", port_a).await;
+        let router = Arc::new(RpcRouter::new(
+            gossip_a.clone(),
+            "a".into(),
+            "fabric-10g".into(),
+        ));
+
+        // A's advertised state so the PeerStatus reply exercises the
+        // gossip → PeerView path even though B isn't in the peer table.
+        gossip_a.set_hot_used(4096).await;
+        gossip_a.set_hot_max(1_000_000).await;
+        gossip_a
+            .set_warm_projects(&["osobh/clawverse", "osobh/clawmates"])
+            .await;
+
+        let server = QuicServer::bind(loopback(0), id_a).unwrap();
+        let server_addr = server.local_addr().unwrap();
+        let accept_task = tokio::spawn(async move {
+            if let Some(Ok(conn)) = server.accept().await {
+                let _ = serve_connection(conn, router).await;
+            }
+        });
+
+        let client = QuicClient::new(loopback(0), id_b).unwrap();
+        let conn = client.connect(server_addr, "a").await.unwrap();
+
+        // Ping.
+        let pong = call_ping(&conn, b"hello").await.unwrap();
+        assert_eq!(pong, b"hello");
+
+        // PeerStatus.
+        let status = call_peer_status(&conn).await.unwrap();
+        assert_eq!(status.local_name, "a");
+        assert_eq!(status.local_zone, "fabric-10g");
+        assert!(status.peers.is_empty(), "A has no peers configured");
+
+        conn.close(quinn::VarInt::from_u32(0), b"done");
+        client.shutdown().await;
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        accept_task.abort();
+        // Keep gossip_a alive to the end so its background task doesn't
+        // drop mid-serve.
+        let _ = gossip_a.peers().await;
+    }
+
+    #[tokio::test]
+    async fn peer_status_reflects_peer_gossip_state() {
+        // A and B both run gossip; A serves RPC. When A's PeerStatus is
+        // called by a third-party client, the reply's `peers` field
+        // contains B (as long as gossip has converged).
+        let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
+        let port_a = next_port();
+        let port_b = next_port();
+
+        let cfg_a = ClusterConfig {
+            zone: "fabric-10g".into(),
+            bind_lan: Some(loopback(port_a)),
+            ..Default::default()
+        };
+        let cfg_b = ClusterConfig {
+            zone: "lan-1g".into(),
+            bind_lan: Some(loopback(port_b)),
+            peers: vec![PeerEntry {
+                name: "a".into(),
+                zone: "fabric-10g".into(),
+                lan_addr: Some(loopback(port_a)),
+                tailscale_addr: None,
+            }],
+            ..Default::default()
+        };
+        let gossip_a = Arc::new(ClusterGossip::bootstrap(&cfg_a, "a").await.unwrap());
+        let gossip_b = Arc::new(ClusterGossip::bootstrap(&cfg_b, "b").await.unwrap());
+
+        let router = Arc::new(RpcRouter::new(
+            gossip_a.clone(),
+            "a".into(),
+            "fabric-10g".into(),
+        ));
+
+        let server = QuicServer::bind(loopback(0), id_a).unwrap();
+        let server_addr = server.local_addr().unwrap();
+        let accept_task = tokio::spawn(async move {
+            if let Some(Ok(conn)) = server.accept().await {
+                let _ = serve_connection(conn, router).await;
+            }
+        });
+
+        // Wait for gossip convergence: A must see B.
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        loop {
+            if let Some(v) = gossip_a.peer("b").await {
+                if v.alive {
+                    break;
+                }
+            }
+            if std::time::Instant::now() >= deadline {
+                panic!("A never saw B alive within 10s");
+            }
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+
+        let client = QuicClient::new(loopback(0), id_b).unwrap();
+        let conn = client.connect(server_addr, "a").await.unwrap();
+        let status = call_peer_status(&conn).await.unwrap();
+        assert_eq!(status.local_name, "a");
+        assert_eq!(status.peers.len(), 1, "A should report exactly B");
+        assert_eq!(status.peers[0].name, "b");
+        assert_eq!(status.peers[0].zone, "lan-1g");
+        assert!(status.peers[0].alive);
+
+        conn.close(quinn::VarInt::from_u32(0), b"done");
+        client.shutdown().await;
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        accept_task.abort();
+        // Keep gossip services alive until end.
+        drop(gossip_b);
+    }
+
+    #[test]
+    fn error_code_describe_covers_all_variants() {
+        assert_eq!(ErrorCode::EmptyRequest.describe(), "empty request");
+        assert_eq!(ErrorCode::UnknownMethod.describe(), "unknown method");
+        assert_eq!(ErrorCode::HandlerFailure.describe(), "handler failure");
+    }
+}

claw-store/src/cluster/services.rs

@@ -0,0 +1,418 @@
+//! Daemon-side wire-up of the cluster stack.
+//!
+//! Ties together the three pieces built in earlier phases:
+//! * gossip ([`ClusterGossip`], Phase 1b)
+//! * QUIC transport + mTLS ([`QuicServer`], Phase 1c/1d)
+//! * RPC dispatch ([`RpcRouter`], Phase 1e)
+//!
+//! A [`ClusterServices`] value owns the background tasks — gossip
+//! service, QUIC accept loop, hot-tier metric ticker — and shuts them
+//! down cleanly on drop.
+//!
+//! When `[cluster]` is absent from the config the daemon runs exactly
+//! as it did pre-v2: no gossip, no RPC, no metric ticker.
+
+use crate::cluster::gossip::ClusterGossip;
+use crate::cluster::rpc::{serve_connection, RpcRouter};
+use crate::cluster::transport::{NodeIdentity, QuicServer};
+use crate::config::ClusterConfig;
+use anyhow::{Context, Result};
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::task::JoinHandle;
+
+/// How often the daemon re-measures its own hot-tier occupancy and
+/// republishes it into gossip. 30s balances timely peer visibility
+/// against filesystem-walk cost on nodes with fat hot tiers.
+const HOT_METRIC_INTERVAL: Duration = Duration::from_secs(30);
+
+/// Live cluster services attached to a running daemon.
+///
+/// Drop shuts down all background tasks. Ownership is single: the
+/// daemon holds one `ClusterServices` for its lifetime. Read-side
+/// access to [`ClusterGossip`] goes through the public [`gossip`]
+/// field, wrapped in `Arc` so the daemon's other subsystems (dashboard,
+/// heartbeat handler, etc.) can query peer state without touching the
+/// background tasks.
+pub struct ClusterServices {
+    /// Live gossip service. `Arc` so read-only consumers on other
+    /// subsystems can hold references without blocking shutdown.
+    pub gossip: Arc<ClusterGossip>,
+    /// QUIC accept-loop task. `None` when `[cluster.tls]` was absent
+    /// and RPC therefore didn't come up.
+    accept_task: Option<JoinHandle<()>>,
+    /// Periodic hot-tier metric publisher. Always running when a
+    /// gossip service exists.
+    metric_task: JoinHandle<()>,
+}
+
+impl std::fmt::Debug for ClusterServices {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ClusterServices")
+            .field("rpc_running", &self.accept_task.is_some())
+            .finish()
+    }
+}
+
+impl ClusterServices {
+    /// Start every cluster subsystem the config asks for:
+    /// * `[cluster]` present → gossip service + metric ticker
+    /// * `[cluster.tls]` also present → QUIC RPC server + accept loop
+    ///
+    /// `local_name` is the name this node advertises to peers.
+    /// `hot_dir` is the local hot-tier root — the metric ticker walks
+    /// it every [`HOT_METRIC_INTERVAL`] and publishes its size.
+    /// `hot_max_bytes` is the configured cap; published once so peers
+    /// can compute a fill ratio.
+    pub async fn start(
+        cluster: &ClusterConfig,
+        local_name: String,
+        hot_dir: PathBuf,
+        hot_max_bytes: u64,
+    ) -> Result<Self> {
+        let gossip = Arc::new(
+            ClusterGossip::bootstrap(cluster, &local_name)
+                .await
+                .context("bootstrapping cluster gossip")?,
+        );
+
+        // Publish the static config value once. Used-bytes updates every tick.
+        gossip.set_hot_max(hot_max_bytes).await;
+
+        // RPC server + accept loop — only when TLS material is configured.
+        let accept_task = match &cluster.tls {
+            Some(tls) => {
+                let identity =
+                    NodeIdentity::from_pem_files(&tls.ca_cert, &tls.node_cert, &tls.node_key)
+                        .context("loading node identity from [cluster.tls]")?;
+                let bind = cluster
+                    .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 router = Arc::new(RpcRouter::new(
+                    gossip.clone(),
+                    local_name.clone(),
+                    cluster.zone.clone(),
+                ));
+                tracing::info!("cluster RPC server listening on {}", bind);
+                Some(tokio::spawn(async move {
+                    accept_forever(server, router).await;
+                }))
+            }
+            None => {
+                tracing::info!("cluster: no [cluster.tls] configured; RPC disabled");
+                None
+            }
+        };
+
+        // Hot-tier metric ticker — walks `hot_dir` and publishes its
+        // aggregate size on every tick. Runs even when RPC is off so a
+        // gossip-only deployment still gets peer visibility.
+        let metric_gossip = gossip.clone();
+        let metric_dir = hot_dir.clone();
+        let metric_task = tokio::spawn(async move {
+            let mut ticker = tokio::time::interval(HOT_METRIC_INTERVAL);
+            loop {
+                ticker.tick().await;
+                let bytes = measure_dir_bytes(&metric_dir).await;
+                metric_gossip.set_hot_used(bytes).await;
+            }
+        });
+
+        Ok(Self {
+            gossip,
+            accept_task,
+            metric_task,
+        })
+    }
+
+    /// Whether the QUIC RPC server is running. `false` when `[cluster.tls]`
+    /// was absent at start time.
+    pub fn rpc_enabled(&self) -> bool {
+        self.accept_task.is_some()
+    }
+
+    /// Graceful shutdown: cancel all background tasks. Peers observe
+    /// this node as dead within `dead_node_grace_period` after the
+    /// gossip service stops responding.
+    pub fn shutdown(self) {
+        if let Some(task) = self.accept_task {
+            task.abort();
+        }
+        self.metric_task.abort();
+    }
+}
+
+/// Loop accepting incoming QUIC connections and dispatching each to a
+/// per-connection task running the RPC router. Runs until the endpoint
+/// is closed (which happens when the parent task is aborted).
+async fn accept_forever(server: QuicServer, router: Arc<RpcRouter>) {
+    loop {
+        match server.accept().await {
+            Some(Ok(conn)) => {
+                let router = router.clone();
+                tokio::spawn(async move {
+                    if let Err(e) = serve_connection(conn, router).await {
+                        tracing::warn!(error = %e, "RPC connection ended with error");
+                    }
+                });
+            }
+            Some(Err(e)) => {
+                tracing::warn!(error = %e, "RPC accept failed");
+            }
+            None => {
+                // Endpoint closed.
+                return;
+            }
+        }
+    }
+}
+
+/// Recursively sum sizes of every regular file under `path`. Returns 0
+/// when the path doesn't exist yet (fresh node, hot dir uninitialised).
+/// Runs on a blocking task since filesystem walks can be slow on large
+/// hot tiers.
+async fn measure_dir_bytes(path: &Path) -> u64 {
+    let path = path.to_path_buf();
+    tokio::task::spawn_blocking(move || dir_bytes_sync(&path))
+        .await
+        .unwrap_or(0)
+}
+
+fn dir_bytes_sync(root: &Path) -> u64 {
+    if !root.exists() {
+        return 0;
+    }
+    let mut total: u64 = 0;
+    let mut stack = vec![root.to_path_buf()];
+    while let Some(dir) = stack.pop() {
+        let entries = match std::fs::read_dir(&dir) {
+            Ok(e) => e,
+            Err(_) => continue,
+        };
+        for entry in entries.flatten() {
+            let ft = match entry.file_type() {
+                Ok(t) => t,
+                Err(_) => continue,
+            };
+            if ft.is_dir() {
+                stack.push(entry.path());
+            } else if ft.is_file() {
+                if let Ok(meta) = entry.metadata() {
+                    total = total.saturating_add(meta.len());
+                }
+            }
+            // Symlinks and other types are counted as 0 bytes — the
+            // real content lives elsewhere and is measured there.
+        }
+    }
+    total
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::cluster::rpc::{call_peer_status, call_ping};
+    use crate::cluster::transport::{FleetCa, NodeIdentity, QuicClient};
+    use crate::config::{ClusterConfig, ClusterTlsConfig, PeerEntry};
+    use std::net::SocketAddr;
+    use std::sync::atomic::{AtomicU16, Ordering};
+
+    /// Dedicated port range for services tests (44000+) so we don't
+    /// collide with gossip (41000+), transport (42000+), rpc (43000+).
+    static NEXT_PORT: AtomicU16 = AtomicU16::new(44001);
+    fn next_port() -> u16 {
+        NEXT_PORT.fetch_add(1, Ordering::Relaxed)
+    }
+
+    fn loopback(port: u16) -> SocketAddr {
+        format!("127.0.0.1:{port}").parse().unwrap()
+    }
+
+    #[tokio::test]
+    async fn dir_bytes_sync_returns_zero_for_missing_path() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let missing = tmp.path().join("does-not-exist");
+        assert_eq!(dir_bytes_sync(&missing), 0);
+    }
+
+    #[tokio::test]
+    async fn dir_bytes_sync_sums_recursive_file_sizes() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        std::fs::create_dir_all(tmp.path().join("a/b/c")).unwrap();
+        std::fs::write(tmp.path().join("top.bin"), vec![0u8; 100]).unwrap();
+        std::fs::write(tmp.path().join("a/mid.bin"), vec![0u8; 250]).unwrap();
+        std::fs::write(tmp.path().join("a/b/c/deep.bin"), vec![0u8; 400]).unwrap();
+        assert_eq!(dir_bytes_sync(tmp.path()), 100 + 250 + 400);
+    }
+
+    #[tokio::test]
+    async fn services_start_without_tls_leaves_rpc_disabled() {
+        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,
+            "test-node".into(),
+            tmp.path().to_path_buf(),
+            1_000_000,
+        )
+        .await
+        .unwrap();
+        assert!(!svc.rpc_enabled(), "no [cluster.tls] → RPC 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");
+        svc.shutdown();
+    }
+
+    #[tokio::test]
+    async fn services_start_with_tls_serves_rpc_end_to_end() {
+        // Cut a real fleet CA, sign leaves for two nodes, start
+        // ClusterServices for A with the on-disk identity, then dial A
+        // from B and run both ping + PeerStatus. Proves the full
+        // Phase 1a-1e stack is wired correctly from config all the way
+        // to reply bytes.
+        let tmp = tempfile::TempDir::new().unwrap();
+        let ca_dir = tmp.path().join("ca");
+        let a_dir = tmp.path().join("a-tls");
+        let b_dir = tmp.path().join("b-tls");
+
+        let ca = FleetCa::generate("test CA").unwrap();
+        ca.save(&ca_dir).unwrap();
+        ca.sign_leaf_to_pem("a", &a_dir).unwrap();
+        ca.sign_leaf_to_pem("b", &b_dir).unwrap();
+
+        let port_a = next_port();
+        let cfg_a = ClusterConfig {
+            zone: "fabric-10g".into(),
+            bind_lan: Some(loopback(port_a)),
+            tls: Some(ClusterTlsConfig {
+                ca_cert: a_dir.join("ca.crt"),
+                node_cert: a_dir.join("node.crt"),
+                node_key: a_dir.join("node.key"),
+            }),
+            ..Default::default()
+        };
+        let hot_dir = tmp.path().join("a-hot");
+        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)
+            .await
+            .unwrap();
+        assert!(svc.rpc_enabled());
+
+        // Dial A from B — use the persisted B identity.
+        let id_b = NodeIdentity::from_pem_dir(&b_dir).unwrap();
+        let client = QuicClient::new(loopback(0), id_b).unwrap();
+        let rpc_addr = cfg_a.rpc_lan().unwrap();
+        // Give the accept loop a moment to be scheduled.
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        let conn = client.connect(rpc_addr, "a").await.unwrap();
+
+        // Ping.
+        let echo = call_ping(&conn, b"hi").await.unwrap();
+        assert_eq!(echo, b"hi");
+
+        // PeerStatus — solo A, no peers.
+        let status = call_peer_status(&conn).await.unwrap();
+        assert_eq!(status.local_name, "a");
+        assert_eq!(status.local_zone, "fabric-10g");
+        assert!(status.peers.is_empty());
+
+        conn.close(quinn::VarInt::from_u32(0), b"done");
+        client.shutdown().await;
+        svc.shutdown();
+    }
+
+    #[tokio::test]
+    async fn services_publish_hot_used_metric_periodically() {
+        // Verify the metric ticker actually publishes into gossip.
+        // Uses a lowered internal by testing directly through the
+        // measure function — the real ticker fires every 30s which is
+        // too slow for a unit test. We test the FUNCTION contract:
+        // measure_dir_bytes returns the sum, then verify that when
+        // start() runs, `hot_max_bytes` is published immediately.
+        let cfg = ClusterConfig {
+            zone: "fabric-10g".into(),
+            bind_lan: Some(loopback(next_port())),
+            ..Default::default()
+        };
+        let tmp = tempfile::TempDir::new().unwrap();
+        std::fs::write(tmp.path().join("f1"), vec![0u8; 1024]).unwrap();
+        let svc = ClusterServices::start(
+            &cfg,
+            "publisher".into(),
+            tmp.path().to_path_buf(),
+            10 * 1024 * 1024,
+        )
+        .await
+        .unwrap();
+        // hot_max is published immediately on start; hot_used takes an
+        // interval tick, so measure via the underlying primitive to
+        // prove the tree walk works. The 30s interval is intentional;
+        // shortening it purely for the test would defeat "no test-only
+        // side doors" — we prove the measurement primitive here and
+        // trust the interval loop.
+        let measured = measure_dir_bytes(tmp.path()).await;
+        assert_eq!(measured, 1024);
+        svc.shutdown();
+    }
+
+    #[tokio::test]
+    async fn services_gossip_sees_peer_after_convergence() {
+        // Two nodes both running ClusterServices (gossip only). A
+        // seeds B; wait for phi-accrual liveness; verify A's PeerView
+        // for B carries the expected zone.
+        let port_a = next_port();
+        let port_b = next_port();
+        let cfg_a = ClusterConfig {
+            zone: "fabric-10g".into(),
+            bind_lan: Some(loopback(port_a)),
+            ..Default::default()
+        };
+        let cfg_b = ClusterConfig {
+            zone: "lan-1g".into(),
+            bind_lan: Some(loopback(port_b)),
+            peers: vec![PeerEntry {
+                name: "a".into(),
+                zone: "fabric-10g".into(),
+                lan_addr: Some(loopback(port_a)),
+                tailscale_addr: None,
+            }],
+            ..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)
+                .await
+                .unwrap();
+        let svc_b =
+            ClusterServices::start(&cfg_b, "b".into(), tmp.path().to_path_buf(), 1_000_000)
+                .await
+                .unwrap();
+
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        loop {
+            if let Some(v) = svc_a.gossip.peer("b").await {
+                if v.alive {
+                    assert_eq!(v.zone, "lan-1g");
+                    break;
+                }
+            }
+            if std::time::Instant::now() >= deadline {
+                panic!("A never saw B alive within 10s");
+            }
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+        svc_a.shutdown();
+        svc_b.shutdown();
+    }
+}

claw-store/src/daemon.rs

@@ -1,4 +1,5 @@
 use crate::actions;
+use crate::cluster::services::ClusterServices;
 use crate::config::Config;
 use crate::head_watch::{scan_and_enqueue, HeadCache};
 use crate::hot;
@@ -25,6 +26,43 @@ pub async fn run(cfg: Config, mut manifest: Manifest) -> Result<()> {
     let zfs = SystemZfs;
     let manifest_path = Manifest::default_path();
 
+    // Cluster services — gossip + optional QUIC RPC + hot-tier metric
+    // publisher. Held for the daemon's lifetime; dropped on shutdown so
+    // peers observe us going away within the failure detector's grace
+    // window. `None` when `[cluster]` is absent from config — pre-v2
+    // deployments still work unchanged.
+    let cluster_services: Option<ClusterServices> = match cfg.cluster.as_ref() {
+        Some(cluster_cfg) => {
+            let hot_dir = cfg.hot.path.clone();
+            let hot_max_bytes = cfg.hot.max_gb.saturating_mul(1024 * 1024 * 1024);
+            match ClusterServices::start(
+                cluster_cfg,
+                cfg.node.name.clone(),
+                hot_dir,
+                hot_max_bytes,
+            )
+            .await
+            {
+                Ok(svc) => {
+                    tracing::info!(
+                        rpc_enabled = svc.rpc_enabled(),
+                        zone = %cluster_cfg.zone,
+                        "cluster services online"
+                    );
+                    Some(svc)
+                }
+                Err(e) => {
+                    tracing::error!(error = %e, "cluster services failed to start; continuing without cluster");
+                    None
+                }
+            }
+        }
+        None => {
+            tracing::info!("no [cluster] section in config; running standalone");
+            None
+        }
+    };
+
     let mut shutdown = tokio::signal::unix::signal(
         tokio::signal::unix::SignalKind::terminate()
     )?;
@@ -37,6 +75,9 @@ pub async fn run(cfg: Config, mut manifest: Manifest) -> Result<()> {
         tokio::select! {
             _ = shutdown.recv() => {
                 tracing::info!("received SIGTERM — shutting down cleanly");
+                if let Some(svc) = cluster_services {
+                    svc.shutdown();
+                }
                 let _ = std::fs::remove_file(DAEMON_STARTED_PATH);
                 break;
             }

claw-store/src/main.rs

@@ -141,6 +141,22 @@ enum Cmd {
         #[arg(long)]
         out_dir: PathBuf,
     },
+    /// Call the peer's PeerStatus RPC and print its local view of the
+    /// cluster. Requires that peer be running a daemon with
+    /// `[cluster.tls]` configured, and that our `--tls-dir` was signed
+    /// by the same CA as the peer's.
+    ClusterPeerStatus {
+        /// Peer's node name — must match the peer's cert SAN.
+        #[arg(long)]
+        peer: String,
+        /// Peer's RPC socket. Typically gossip_port + 1.
+        #[arg(long)]
+        rpc_addr: SocketAddr,
+        /// Directory holding this node's mTLS material
+        /// (`ca.crt` + `node.crt` + `node.key` from `fleet-ca sign`).
+        #[arg(long)]
+        tls_dir: PathBuf,
+    },
 }
 
 #[tokio::main]
@@ -198,6 +214,11 @@ async fn main() -> Result<()> {
             payload,
             tls_dir,
         } => cmd_cluster_ping(&name, &peer, rpc_addr, &payload, tls_dir.as_deref()).await?,
+        Cmd::ClusterPeerStatus {
+            peer,
+            rpc_addr,
+            tls_dir,
+        } => cmd_cluster_peer_status(&peer, rpc_addr, &tls_dir).await?,
         Cmd::FleetCaInit { .. } | Cmd::FleetCaSign { .. } => {
             // Handled by the config-independent short-circuit above.
             unreachable!("fleet-ca commands short-circuit before config load");
@@ -206,6 +227,60 @@ async fn main() -> Result<()> {
     Ok(())
 }
 
+// ── cluster peer-status ───────────────────────────────────────────────────────
+
+async fn cmd_cluster_peer_status(
+    peer: &str,
+    rpc_addr: SocketAddr,
+    tls_dir: &std::path::Path,
+) -> Result<()> {
+    use cluster::rpc::call_peer_status;
+    use cluster::transport::{NodeIdentity, QuicClient};
+
+    let identity = NodeIdentity::from_pem_dir(tls_dir)
+        .with_context(|| format!("loading identity from {}", tls_dir.display()))?;
+    let client = QuicClient::new("0.0.0.0:0".parse()?, identity)?;
+    let conn = client.connect(rpc_addr, peer).await?;
+    let status = call_peer_status(&conn).await?;
+
+    println!(
+        "peer:  {} (zone: {})",
+        status.local_name, status.local_zone
+    );
+    println!();
+    if status.peers.is_empty() {
+        println!("  (no peers known)");
+    } else {
+        println!(
+            "  {:<20} {:<14} {:<8} {:<22} {:<20}",
+            "NAME", "ZONE", "STATE", "RPC LAN", "HOT USED / MAX"
+        );
+        println!("  {}", "-".repeat(90));
+        for p in &status.peers {
+            let state = if p.alive { "alive" } else { "dead" };
+            let hot = match (p.hot_used_bytes, p.hot_max_bytes) {
+                (Some(u), Some(m)) => format!("{u} / {m}"),
+                (Some(u), None) => format!("{u} / -"),
+                _ => "-".into(),
+            };
+            println!(
+                "  {:<20} {:<14} {:<8} {:<22} {:<20}",
+                p.name,
+                p.zone,
+                state,
+                p.rpc_lan
+                    .map(|a| a.to_string())
+                    .unwrap_or_else(|| "-".into()),
+                hot,
+            );
+        }
+    }
+
+    conn.close(quinn::VarInt::from_u32(0), b"done");
+    client.shutdown().await;
+    Ok(())
+}
+
 // ── fleet-ca init / sign ─────────────────────────────────────────────────────
 
 fn cmd_fleet_ca_init(dir: &std::path::Path, cn: &str) -> Result<()> {