claw-store/src/cluster/gossip.rs

@@ -17,6 +17,7 @@
 //! RPC endpoints, hot-tier occupancy, and the warm-tier projects they
 //! serve — the last of which drives Gitea runner label dynamics later.
 
+use crate::cluster::metrics::MetricsReply;
 use crate::config::ClusterConfig;
 use anyhow::{bail, Context, Result};
 use chitchat::transport::UdpTransport;
@@ -49,6 +50,18 @@ pub mod keys {
     pub const WARM_PROJECTS: &str = "clawstor.warm.projects";
     /// Unix timestamp (seconds) when this node's daemon started.
     pub const UPTIME_UNIX: &str = "clawstor.uptime.unix";
+    /// Phase 5i: cumulative `GetRef` hits observed by this node's cache
+    /// since counters were reset. Placement policy divides by
+    /// `hits + misses` to derive a hit rate without an RPC roundtrip.
+    pub const CACHE_GET_REF_HITS: &str = "clawstor.cache.get_ref.hits";
+    /// Phase 5i: cumulative `GetRef` misses.
+    pub const CACHE_GET_REF_MISSES: &str = "clawstor.cache.get_ref.misses";
+    /// Phase 5i: cumulative bytes served out of this node's blob store
+    /// via `BlobGet` / `GetChunk`. Byte-volume signal for load balancing.
+    pub const CACHE_BLOB_GET_BYTES: &str = "clawstor.cache.blob_get.bytes";
+    /// Phase 5i: cumulative bytes ingested into this node's blob store
+    /// via `BlobPut`.
+    pub const CACHE_BLOB_PUT_BYTES: &str = "clawstor.cache.blob_put.bytes";
 }
 
 /// Cluster identifier — every node in the same fleet must agree on this
@@ -89,6 +102,15 @@ pub struct PeerView {
     pub warm_projects: Vec<String>,
     pub uptime_unix: Option<u64>,
     pub alive: bool,
+    /// Phase 5i: cumulative `GetRef` hits — `None` when the peer hasn't
+    /// yet published its cache counters (fresh daemon, or no RPC).
+    pub cache_get_ref_hits: Option<u64>,
+    /// Phase 5i: cumulative `GetRef` misses.
+    pub cache_get_ref_misses: Option<u64>,
+    /// Phase 5i: cumulative bytes served out of this node's blob store.
+    pub cache_blob_get_bytes: Option<u64>,
+    /// Phase 5i: cumulative bytes ingested into this node's blob store.
+    pub cache_blob_put_bytes: Option<u64>,
 }
 
 impl PeerView {
@@ -99,6 +121,20 @@ impl PeerView {
             _ => None,
         }
     }
+
+    /// Phase 5i: `GetRef` hit rate derived from gossiped counters.
+    /// Returns `None` when the peer hasn't published counters yet, or
+    /// when zero lookups have been recorded (0/0 is undefined).
+    pub fn cache_get_ref_hit_rate(&self) -> Option<f64> {
+        let hits = self.cache_get_ref_hits?;
+        let misses = self.cache_get_ref_misses?;
+        let total = hits.saturating_add(misses);
+        if total == 0 {
+            None
+        } else {
+            Some(hits as f64 / total as f64)
+        }
+    }
 }
 
 /// Running gossip service — owns the chitchat handle. Drop shuts down
@@ -226,6 +262,25 @@ impl ClusterGossip {
         self.set(keys::HOT_MAX_BYTES, bytes.to_string()).await;
     }
 
+    /// Phase 5i: publish this node's cache-metric snapshot into gossip.
+    ///
+    /// Peers see the four raw counters — `GetRef` hits + misses and blob
+    /// GET/PUT byte totals — on the next gossip round and derive hit
+    /// rate locally via [`PeerView::cache_get_ref_hit_rate`]. Called
+    /// periodically by [`crate::cluster::services::ClusterServices`]
+    /// once RPC has come up.
+    pub async fn set_cache_metrics(&self, snap: &MetricsReply) {
+        // Batch all four writes under one lock — chitchat serialises
+        // per-node kv writes, so a single locked block avoids an
+        // interleaving where peers see hits without misses.
+        let mut cc = self.chitchat.lock().await;
+        let state = cc.self_node_state();
+        state.set(keys::CACHE_GET_REF_HITS, snap.get_ref_hits.to_string());
+        state.set(keys::CACHE_GET_REF_MISSES, snap.get_ref_misses.to_string());
+        state.set(keys::CACHE_BLOB_GET_BYTES, snap.blob_get_bytes.to_string());
+        state.set(keys::CACHE_BLOB_PUT_BYTES, snap.blob_put_bytes.to_string());
+    }
+
     /// Publish the list of warm-tier `org/repo` projects this node serves.
     /// Later phases use this to bias runner scheduling.
     pub async fn set_warm_projects<S: AsRef<str>>(&self, projects: &[S]) {
@@ -329,6 +384,10 @@ fn peer_view_from_state(
         warm_projects,
         uptime_unix: get_u64(state, keys::UPTIME_UNIX),
         alive,
+        cache_get_ref_hits: get_u64(state, keys::CACHE_GET_REF_HITS),
+        cache_get_ref_misses: get_u64(state, keys::CACHE_GET_REF_MISSES),
+        cache_blob_get_bytes: get_u64(state, keys::CACHE_BLOB_GET_BYTES),
+        cache_blob_put_bytes: get_u64(state, keys::CACHE_BLOB_PUT_BYTES),
     }
 }
 
@@ -548,6 +607,129 @@ mod tests {
         g.shutdown();
     }
 
+    #[tokio::test]
+    async fn set_cache_metrics_publishes_all_four_counters_to_peer() {
+        // Phase 5i: A publishes cache metrics; B's PeerView for A must
+        // carry all four raw counters + derive the hit rate correctly.
+        use crate::cluster::metrics::CacheMetrics;
+
+        let port_a = next_lan_port();
+        let port_b = next_lan_port();
+        let addr_a = loopback(port_a);
+        let addr_b = loopback(port_b);
+        let cfg_a = ClusterConfig {
+            zone: "fabric-10g".into(),
+            bind_lan: Some(addr_a),
+            bind_tailscale: None,
+            peers: vec![],
+            bind_rpc_lan: None,
+            bind_rpc_tailscale: None,
+            tls: None,
+            blob_store_root: None,
+        };
+        let cfg_b = ClusterConfig {
+            zone: "lan-1g".into(),
+            bind_lan: Some(addr_b),
+            bind_tailscale: None,
+            peers: vec![PeerEntry {
+                name: "a".into(),
+                zone: "fabric-10g".into(),
+                lan_addr: Some(addr_a),
+                tailscale_addr: None,
+            }],
+            bind_rpc_lan: None,
+            bind_rpc_tailscale: None,
+            tls: None,
+            blob_store_root: None,
+        };
+        let g_a = ClusterGossip::bootstrap(&cfg_a, "a").await.unwrap();
+        let g_b = ClusterGossip::bootstrap(&cfg_b, "b").await.unwrap();
+
+        // Feed A's metrics — 3 hits, 1 miss ⇒ 0.75 hit rate — then
+        // publish. Byte counters test the u64 path independently.
+        let m = CacheMetrics::new();
+        for _ in 0..3 {
+            m.record_get_ref_hit();
+        }
+        m.record_get_ref_miss();
+        m.record_blob_get_bytes(4096);
+        m.record_blob_put_bytes(8192);
+        g_a.set_cache_metrics(&m.snapshot()).await;
+
+        assert!(
+            wait_until_peer_alive(&g_b, "a", Duration::from_secs(10)).await,
+            "B must see A alive"
+        );
+
+        // Wait for the specific cache keys to converge — liveness is
+        // faster than kv propagation on the first sight of a new peer.
+        let deadline = Instant::now() + Duration::from_secs(10);
+        loop {
+            if let Some(v) = g_b.peer("a").await {
+                if v.cache_get_ref_hits.is_some() {
+                    assert_eq!(v.cache_get_ref_hits, Some(3));
+                    assert_eq!(v.cache_get_ref_misses, Some(1));
+                    assert_eq!(v.cache_blob_get_bytes, Some(4096));
+                    assert_eq!(v.cache_blob_put_bytes, Some(8192));
+                    let rate = v.cache_get_ref_hit_rate().unwrap();
+                    assert!((rate - 0.75).abs() < 1e-9, "hit rate = {rate}");
+                    break;
+                }
+            }
+            if Instant::now() >= deadline {
+                panic!("cache metrics never propagated to B within 10s");
+            }
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+        g_a.shutdown();
+        g_b.shutdown();
+    }
+
+    #[test]
+    fn peer_view_cache_hit_rate_none_when_counters_missing() {
+        let base = PeerView {
+            name: "x".into(),
+            zone: "z".into(),
+            rpc_lan: None,
+            rpc_tailscale: None,
+            hot_used_bytes: None,
+            hot_max_bytes: None,
+            warm_projects: vec![],
+            uptime_unix: None,
+            alive: true,
+            cache_get_ref_hits: None,
+            cache_get_ref_misses: None,
+            cache_blob_get_bytes: None,
+            cache_blob_put_bytes: None,
+        };
+        assert_eq!(base.cache_get_ref_hit_rate(), None, "no counters → None");
+
+        // Hits published but not misses — treat as None too. Prevents a
+        // half-write from producing a bogus 1.0 hit rate.
+        let half = PeerView {
+            cache_get_ref_hits: Some(5),
+            ..base.clone()
+        };
+        assert_eq!(half.cache_get_ref_hit_rate(), None, "hits only → None");
+
+        // Both zero — division would be undefined; must be None.
+        let zero = PeerView {
+            cache_get_ref_hits: Some(0),
+            cache_get_ref_misses: Some(0),
+            ..base.clone()
+        };
+        assert_eq!(zero.cache_get_ref_hit_rate(), None, "0/0 → None");
+
+        // Real ratio.
+        let real = PeerView {
+            cache_get_ref_hits: Some(9),
+            cache_get_ref_misses: Some(1),
+            ..base
+        };
+        let rate = real.cache_get_ref_hit_rate().unwrap();
+        assert!((rate - 0.9).abs() < 1e-9, "expected 0.9, got {rate}");
+    }
+
     #[test]
     fn peer_view_hot_fill_ratio_handles_missing_or_zero() {
         let base = PeerView {
@@ -560,6 +742,10 @@ mod tests {
             warm_projects: vec![],
             uptime_unix: None,
             alive: true,
+            cache_get_ref_hits: None,
+            cache_get_ref_misses: None,
+            cache_blob_get_bytes: None,
+            cache_blob_put_bytes: None,
         };
         assert_eq!(base.hot_fill_ratio(), None, "no used → None");
 

claw-store/src/cluster/services.rs

@@ -30,6 +30,13 @@ use tokio::task::JoinHandle;
 /// against filesystem-walk cost on nodes with fat hot tiers.
 const HOT_METRIC_INTERVAL: Duration = Duration::from_secs(30);
 
+/// Phase 5i: how often the daemon publishes its cache-metric snapshot
+/// into gossip. Cache counters change on every request so freshness
+/// matters, but writing to chitchat too often bloats gossip payloads.
+/// 60s is the sweet spot — a placement engine polling on a 5-minute
+/// cadence still sees at-worst-60s-stale counters.
+const CACHE_METRIC_INTERVAL: Duration = Duration::from_secs(60);
+
 /// Live cluster services attached to a running daemon.
 ///
 /// Drop shuts down all background tasks. Ownership is single: the
@@ -55,12 +62,19 @@ pub struct ClusterServices {
     /// `DeleteTag` / `ListTags` — the human-readable pin layer over
     /// raw refs. Opened alongside the blob store for the same reason.
     pub tag_store: Option<Arc<TagStore>>,
+    /// RPC router. Held so the cache-metric ticker can call
+    /// `router.metrics()` without duplicating an `Arc<CacheMetrics>`
+    /// field. `None` when RPC never came up (no `[cluster.tls]`).
+    pub router: Option<Arc<RpcRouter>>,
     /// 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<()>,
+    /// Phase 5i: periodic cache-metric publisher. `None` when RPC is
+    /// not running (no router → no counters to publish).
+    cache_metric_task: Option<JoinHandle<()>>,
 }
 
 impl std::fmt::Debug for ClusterServices {
@@ -143,8 +157,28 @@ impl ClusterServices {
         };
 
         // RPC server + accept loop — only when TLS material is configured.
-        let accept_task = match &cluster.tls {
-            Some(tls) => {
+        // Router is built even when TLS is absent iff a blob store is
+        // 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());
+            if let Some(store) = &blob_store {
+                r = r.with_blob_store(store.clone());
+            }
+            if let Some(store) = &ref_store {
+                r = r.with_ref_store(store.clone());
+            }
+            if let Some(store) = &tag_store {
+                r = r.with_tag_store(store.clone());
+            }
+            Some(Arc::new(r))
+        } else {
+            None
+        };
+
+        let accept_task = match (&cluster.tls, &router) {
+            (Some(tls), Some(router)) => {
                 let identity =
                     NodeIdentity::from_pem_files(&tls.ca_cert, &tls.node_cert, &tls.node_key)
                         .context("loading node identity from [cluster.tls]")?;
@@ -154,27 +188,13 @@ impl ClusterServices {
                     .context("no RPC bind address (need bind_lan or bind_tailscale)")?;
                 let server = QuicServer::bind(bind, identity)
                     .context("binding QUIC RPC server")?;
-                let mut router = RpcRouter::new(
-                    gossip.clone(),
-                    local_name.clone(),
-                    cluster.zone.clone(),
-                );
-                if let Some(store) = &blob_store {
-                    router = router.with_blob_store(store.clone());
-                }
-                if let Some(store) = &ref_store {
-                    router = router.with_ref_store(store.clone());
-                }
-                if let Some(store) = &tag_store {
-                    router = router.with_tag_store(store.clone());
-                }
-                let router = Arc::new(router);
+                let router = router.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
             }
@@ -194,13 +214,38 @@ impl ClusterServices {
             }
         });
 
+        // Phase 5i: cache-metric ticker. Only meaningful when the
+        // router exists, because counters only increment inside RPC
+        // dispatch. Publish once immediately so peers don't wait a full
+        // interval for the first read even when no requests have fired.
+        let cache_metric_task = if let Some(router) = &router {
+            let router = router.clone();
+            let gossip = gossip.clone();
+            gossip.set_cache_metrics(&router.metrics().snapshot()).await;
+            Some(tokio::spawn(async move {
+                let mut ticker = tokio::time::interval(CACHE_METRIC_INTERVAL);
+                // First tick fires immediately (tokio interval default),
+                // covered by the initial publish above — skip it here.
+                ticker.tick().await;
+                loop {
+                    ticker.tick().await;
+                    let snap = router.metrics().snapshot();
+                    gossip.set_cache_metrics(&snap).await;
+                }
+            }))
+        } else {
+            None
+        };
+
         Ok(Self {
             gossip,
             blob_store,
             ref_store,
             tag_store,
+            router,
             accept_task,
             metric_task,
+            cache_metric_task,
         })
     }
 
@@ -228,6 +273,9 @@ impl ClusterServices {
             task.abort();
         }
         self.metric_task.abort();
+        if let Some(task) = self.cache_metric_task {
+            task.abort();
+        }
     }
 }
 
@@ -510,6 +558,125 @@ mod tests {
         svc_b.shutdown();
     }
 
+    #[tokio::test]
+    async fn services_publish_cache_metrics_to_gossip_on_start() {
+        // Phase 5i: when RPC is up, the initial cache-metric publish
+        // fires immediately at start(). A peer B seeded from A must see
+        // A's four cache counter keys within gossip convergence time,
+        // even though no RPC requests have been issued.
+        use crate::cluster::transport::FleetCa;
+        use crate::config::ClusterTlsConfig;
+
+        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 port_a = next_port();
+        let port_b = next_port();
+        let cfg_a = ClusterConfig {
+            zone: "fabric-10g".into(),
+            bind_lan: Some(loopback(port_a)),
+            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"),
+            }),
+            ..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 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,
+        )
+        .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,
+        )
+        .await
+        .unwrap();
+
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        loop {
+            if let Some(v) = svc_b.gossip.peer("a").await {
+                if v.cache_get_ref_hits == Some(0) && v.alive {
+                    // Initial publish converged. Sanity-check all four.
+                    assert_eq!(v.cache_get_ref_misses, Some(0));
+                    assert_eq!(v.cache_blob_get_bytes, Some(0));
+                    assert_eq!(v.cache_blob_put_bytes, Some(0));
+                    break;
+                }
+            }
+            if std::time::Instant::now() >= deadline {
+                panic!("initial cache-metric publish never reached B in 10s");
+            }
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+
+        // Now push counters directly through the router (in-process) and
+        // publish once more manually via gossip — proves the whole
+        // router↔gossip wiring is real, without waiting 60s for the
+        // ticker.
+        let router = svc_a.router.as_ref().expect("router when TLS up");
+        for _ in 0..4 {
+            router.metrics().record_get_ref_hit();
+        }
+        router.metrics().record_get_ref_miss();
+        router.metrics().record_blob_get_bytes(2048);
+        svc_a
+            .gossip
+            .set_cache_metrics(&router.metrics().snapshot())
+            .await;
+
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        loop {
+            if let Some(v) = svc_b.gossip.peer("a").await {
+                if v.cache_get_ref_hits == Some(4) {
+                    assert_eq!(v.cache_get_ref_misses, Some(1));
+                    assert_eq!(v.cache_blob_get_bytes, Some(2048));
+                    let rate = v.cache_get_ref_hit_rate().unwrap();
+                    assert!((rate - 0.8).abs() < 1e-9, "hit rate = {rate}");
+                    break;
+                }
+            }
+            if std::time::Instant::now() >= deadline {
+                panic!("updated cache metrics never reached B in 10s");
+            }
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+
+        svc_a.shutdown();
+        svc_b.shutdown();
+    }
+
     #[tokio::test]
     async fn services_with_blob_store_serves_blob_rpc_end_to_end() {
         // Prove that config → services → RPC path plumbs BlobStore