//! Cluster membership + KV state broadcast via chitchat SWIM gossip. //! //! Wraps `chitchat::spawn_chitchat` with clawstor-specific keys and a //! typed [`PeerView`] readout. Bootstrapped from static //! `[[cluster.peers]]` seed nodes in the config; membership extends //! dynamically as nodes come and go. //! //! # Transport //! //! Chitchat's built-in UDP transport listens on the [`ClusterConfig`] //! bind address. RPC transport (Phase 1c, QUIC) will use a different //! UDP port so gossip and RPC don't collide — QUIC is UDP too. //! //! # State advertised per node //! //! Well-known keys enumerated in [`keys`]. Peers publish their zone, //! 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; use chitchat::{ spawn_chitchat, Chitchat, ChitchatConfig, ChitchatHandle, ChitchatId, FailureDetectorConfig, }; use std::net::SocketAddr; use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::sync::Mutex; /// Well-known kv keys every clawstor node broadcasts under its own state. pub mod keys { /// Human-facing node name (mirrors the `[[cluster.peers]].name` from /// each peer's own config). pub const NODE_NAME: &str = "clawstor.name"; /// Zone tag: `"fabric-10g"`, `"lan-1g"`, `"roaming"`, or a custom label. pub const ZONE: &str = "clawstor.zone"; /// LAN RPC socket (host:port) — set when the node has a LAN bind. pub const RPC_ADDR_LAN: &str = "clawstor.rpc.lan"; /// Tailscale RPC socket (host:port) — set when the node has a Tailscale bind. pub const RPC_ADDR_TAILSCALE: &str = "clawstor.rpc.tailscale"; /// Bytes currently used in the local hot tier. pub const HOT_USED_BYTES: &str = "clawstor.hot.used"; /// Configured maximum for the local hot tier. pub const HOT_MAX_BYTES: &str = "clawstor.hot.max"; /// Comma-separated list of `org/repo` names this node serves warm. /// Feeds runner-label dynamics in later phases. 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"; /// Field finding 2026-07-12: `rustc --version --verbose` short form /// — the "release" line only, e.g. `1.97.0`. Fingerprints depend on /// the full verbose output, so a mismatch here is a strong hint /// that two nodes will silo their caches. Peer-visible via /// `PeerView.rustc_release` and `cluster-peer-status`. pub const RUSTC_RELEASE: &str = "clawstor.rustc.release"; } /// Cluster identifier — every node in the same fleet must agree on this /// string. Chitchat drops messages that carry a mismatched cluster id. const DEFAULT_CLUSTER_ID: &str = "clawstor"; /// How often chitchat runs its gossip round. 500ms is tighter than the /// crate's default (1s) — fits a small fleet where fast propagation matters /// more than saving inter-node bytes. const DEFAULT_GOSSIP_INTERVAL: Duration = Duration::from_millis(500); /// After a peer stops responding for this long the failure detector /// declares it dead. Kept short (10s) so a rebooted peer is noticed /// quickly by the placement policy. const DEFAULT_DEAD_NODE_GRACE: Duration = Duration::from_secs(10); /// Grace period for tombstoned KVs before hard GC. Long enough that a /// briefly-disconnected node still catches up on deletions on rejoin. const DEFAULT_MARKED_FOR_DELETION_GRACE: Duration = Duration::from_secs(60); /// A snapshot of one peer's advertised state, gathered from gossip. /// /// `alive` reflects the failure detector's phi-accrual decision at read /// 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). /// /// `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, pub rpc_lan: Option, pub rpc_tailscale: Option, pub hot_used_bytes: Option, pub hot_max_bytes: Option, pub warm_projects: Vec, pub uptime_unix: Option, 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, /// Phase 5i: cumulative `GetRef` misses. pub cache_get_ref_misses: Option, /// Phase 5i: cumulative bytes served out of this node's blob store. pub cache_blob_get_bytes: Option, /// Phase 5i: cumulative bytes ingested into this node's blob store. pub cache_blob_put_bytes: Option, /// Field finding 2026-07-12: peer's `rustc --version` release /// string. `None` while the peer boots or when it can't invoke /// rustc. Used to surface toolchain drift that would otherwise /// silently silo caches. pub rustc_release: Option, } impl PeerView { /// Fraction of hot tier used, when both used + max are known. pub fn hot_fill_ratio(&self) -> Option { match (self.hot_used_bytes, self.hot_max_bytes) { (Some(used), Some(max)) if max > 0 => Some(used as f64 / max as f64), _ => 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 { 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 /// the background task via [`ChitchatHandle::abort`]. pub struct ClusterGossip { chitchat: Arc>, handle: ChitchatHandle, } impl std::fmt::Debug for ClusterGossip { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("ClusterGossip").finish_non_exhaustive() } } /// Wrap a config value in the shape chitchat expects (`(String, String)`). fn kv(key: &str, value: impl Into) -> (String, String) { (key.to_string(), value.into()) } /// Turn `ClusterConfig` peers into the string-form seed list chitchat wants. /// Prefers LAN address (fast path) but falls back to Tailscale if that's the /// only reachable one — a roaming peer can still be a bootstrap seed. fn seed_list(cluster: &ClusterConfig) -> Vec { cluster .peers .iter() .filter_map(|p| p.lan_addr.or(p.tailscale_addr)) .map(|a| a.to_string()) .collect() } /// Now, as a Unix seconds timestamp. Falls back to 0 pre-1970 (never /// happens in practice but keeps this pure). fn unix_now() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) .map(|d| d.as_secs()) .unwrap_or(0) } impl ClusterGossip { /// Bootstrap chitchat: bind UDP transport, publish initial state, /// contact seed peers, return the running service. The returned /// `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) -> Result { let local_name = local_name.into(); if local_name.is_empty() { bail!("cluster gossip requires a non-empty local node name"); } cluster .validate() .context("validating cluster config before gossip bootstrap")?; let bind_addr = cluster .bind_lan .or(cluster.bind_tailscale) .context("cluster has no bind address; cannot bootstrap gossip")?; // ChitchatId generation must monotonically increase every restart // so peers can tell "same node, new incarnation" apart from "stuck". // The unix timestamp gives us that naturally. let generation = unix_now(); let chitchat_id = ChitchatId::new(local_name.clone(), generation, bind_addr); let mut initial_kvs = vec![ kv(keys::NODE_NAME, &local_name), kv(keys::ZONE, &cluster.zone), kv(keys::UPTIME_UNIX, generation.to_string()), ]; // Advertise the RPC address (QUIC), NOT the gossip address. // Gossip lives on `bind_lan`; RPC lives on `bind_lan.port + 1` // (or the explicit `bind_rpc_lan` override) so both protocols // — both UDP-based — don't collide. if let Some(rpc_lan) = cluster.rpc_lan() { initial_kvs.push(kv(keys::RPC_ADDR_LAN, rpc_lan.to_string())); } if let Some(rpc_ts) = cluster.rpc_tailscale() { initial_kvs.push(kv(keys::RPC_ADDR_TAILSCALE, rpc_ts.to_string())); } let config = ChitchatConfig { cluster_id: DEFAULT_CLUSTER_ID.to_string(), chitchat_id, gossip_interval: DEFAULT_GOSSIP_INTERVAL, listen_addr: bind_addr, seed_nodes: seed_list(cluster), failure_detector_config: FailureDetectorConfig { dead_node_grace_period: DEFAULT_DEAD_NODE_GRACE, ..FailureDetectorConfig::default() }, marked_for_deletion_grace_period: DEFAULT_MARKED_FOR_DELETION_GRACE, catchup_callback: None, extra_liveness_predicate: None, }; let handle = spawn_chitchat(config, initial_kvs, &UdpTransport) .await .context("spawning chitchat gossip service")?; let chitchat = handle.chitchat(); Ok(Self { chitchat, handle }) } /// Set an arbitrary key on our own advertised state. Peers see it on /// the next gossip round. pub async fn set(&self, key: &str, value: impl Into) { let value = value.into(); let mut cc = self.chitchat.lock().await; cc.self_node_state().set(key, &value); } /// Publish current hot-tier usage. Called periodically by the daemon. pub async fn set_hot_used(&self, bytes: u64) { self.set(keys::HOT_USED_BYTES, bytes.to_string()).await; } /// Publish configured hot-tier maximum. Called once at bootstrap and /// whenever config reloads change it. pub async fn set_hot_max(&self, bytes: u64) { 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()); } /// Field finding 2026-07-12: publish the local rustc release /// string. Called at daemon startup so peers can flag mismatches /// before wasting a build on a cache that will silo. pub async fn set_rustc_release(&self, release: impl Into) { self.set(keys::RUSTC_RELEASE, release).await; } /// 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>(&self, projects: &[S]) { let joined = projects .iter() .map(|s| s.as_ref()) .collect::>() .join(","); self.set(keys::WARM_PROJECTS, joined).await; } /// This node's identity as chitchat sees it (name, generation, gossip /// address). Useful for logs and self-filtering during peer reads. pub async fn self_chitchat_id(&self) -> ChitchatId { self.chitchat.lock().await.self_chitchat_id().clone() } /// All known peers other than self, with their advertised state and /// liveness. Includes peers currently in the grace period (dead but /// not yet garbage-collected). pub async fn peers(&self) -> Vec { let cc = self.chitchat.lock().await; let self_id = cc.self_chitchat_id().clone(); let live: std::collections::HashSet = cc.live_nodes().cloned().collect(); let mut views = Vec::new(); for id in cc.live_nodes().chain(cc.dead_nodes()) { if *id == self_id { continue; } if let Some(state) = cc.node_state(id) { views.push(peer_view_from_state(id, state, live.contains(id))); } } views } /// Look up one peer by advertised name. Returns `None` if we've never /// heard from a peer with that name. pub async fn peer(&self, name: &str) -> Option { self.peers().await.into_iter().find(|p| p.name == name) } /// Live peers in a given zone. pub async fn peers_in_zone(&self, zone: &str) -> Vec { self.peers() .await .into_iter() .filter(|p| p.alive && p.zone == zone) .collect() } /// Graceful shutdown: aborts the gossip task. Peers observe this node /// as dead within `dead_node_grace_period`. pub fn shutdown(self) { self.handle.abort(); } } /// Extract a well-known key from a peer's chitchat state. Returns `None` /// when the key hasn't been advertised (early bootstrap, or the peer /// simply doesn't publish it). fn get_str(state: &chitchat::NodeState, key: &str) -> Option { state.get(key).map(|s| s.to_string()) } fn get_u64(state: &chitchat::NodeState, key: &str) -> Option { state.get(key).and_then(|s| s.parse().ok()) } fn get_socket(state: &chitchat::NodeState, key: &str) -> Option { state.get(key).and_then(|s| s.parse().ok()) } 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). let name = get_str(state, keys::NODE_NAME).unwrap_or_else(|| id.node_id.to_string()); let zone = get_str(state, keys::ZONE).unwrap_or_default(); let warm_projects = get_str(state, keys::WARM_PROJECTS) .map(|s| { s.split(',') .filter_map(|p| { let t = p.trim(); if t.is_empty() { None } else { Some(t.to_string()) } }) .collect() }) .unwrap_or_default(); PeerView { name, zone, rpc_lan: get_socket(state, keys::RPC_ADDR_LAN), rpc_tailscale: get_socket(state, keys::RPC_ADDR_TAILSCALE), hot_used_bytes: get_u64(state, keys::HOT_USED_BYTES), hot_max_bytes: get_u64(state, keys::HOT_MAX_BYTES), 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), rustc_release: get_str(state, keys::RUSTC_RELEASE), } } #[cfg(test)] mod tests { use super::*; use crate::config::PeerEntry; use std::sync::atomic::{AtomicU16, Ordering}; use std::time::Instant; /// Fixed port range for gossip tests, so multiple parallel tests within /// the same binary don't collide. 41000-41999 is above the ephemeral /// range on most Linux boxes (32768-60999 minus the top part) yet still /// out of the well-known-service ranges. Two ports per test. static NEXT_PORT: AtomicU16 = AtomicU16::new(41001); fn next_lan_port() -> u16 { NEXT_PORT.fetch_add(1, Ordering::Relaxed) } fn loopback(port: u16) -> SocketAddr { format!("127.0.0.1:{port}").parse().unwrap() } /// Wait until the gossip service both knows about the named peer AND /// 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 { let start = Instant::now(); loop { if let Some(v) = gossip.peer(name).await { if v.alive { return true; } } if start.elapsed() >= deadline { return gossip.peer(name).await.map(|v| v.alive).unwrap_or(false); } tokio::time::sleep(Duration::from_millis(100)).await; } } #[tokio::test] async fn bootstrap_publishes_our_own_state() { let port = next_lan_port(); let cfg = ClusterConfig { zone: "fabric-10g".into(), bind_lan: Some(loopback(port)), bind_tailscale: None, peers: vec![], bind_rpc_lan: None, 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; assert_eq!(id.node_id.as_ref(), "solo"); assert_eq!(id.gossip_advertise_addr, loopback(port)); g.shutdown(); } #[tokio::test] async fn bootstrap_fails_when_local_name_empty() { let cfg = ClusterConfig { zone: "fabric-10g".into(), bind_lan: Some(loopback(next_lan_port())), bind_tailscale: None, peers: vec![], bind_rpc_lan: None, bind_rpc_tailscale: None, tls: None, blob_store_root: None, prom_bind: None, }; let err = ClusterGossip::bootstrap(&cfg, "") .await .unwrap_err() .to_string(); assert!(err.contains("non-empty local node name"), "err: {err}"); } #[tokio::test] async fn bootstrap_fails_when_no_bind_address() { let cfg = ClusterConfig { zone: "fabric-10g".into(), bind_lan: None, bind_tailscale: None, peers: vec![], bind_rpc_lan: None, 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() ); assert!( err_chain.contains("no bind address"), "expected 'no bind address' in error chain, got: {err_chain}" ); } #[tokio::test] async fn two_node_cluster_converges_and_shares_state() { let port_a = next_lan_port(); let port_b = next_lan_port(); let addr_a = loopback(port_a); let addr_b = loopback(port_b); // Node A: no seeds — starts alone. 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, prom_bind: None, }; // Node B: uses A as seed. 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, prom_bind: None, }; let gossip_a = ClusterGossip::bootstrap(&cfg_a, "a").await.unwrap(); let gossip_b = ClusterGossip::bootstrap(&cfg_b, "b").await.unwrap(); // Publish some state we'll verify propagated. gossip_a.set_hot_used(1024).await; gossip_a.set_hot_max(1_000_000).await; gossip_a .set_warm_projects(&["osobh/clawverse", "osobh/clawmates"]) .await; // Wait for gossip convergence AND phi-accrual to classify both // 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; 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; assert!(b_sees_a_alive, "B should see A alive within 10s"); // Verify the state B sees for A matches what A published. RPC // address is gossip port + 1 (see rpc_lan()) — B must see that. let b_view_of_a = gossip_b.peer("a").await.expect("B has A"); assert_eq!(b_view_of_a.zone, "fabric-10g"); assert_eq!( b_view_of_a.rpc_lan, Some(loopback(port_a + 1)), "advertised RPC addr = gossip port + 1" ); assert!(b_view_of_a.alive); assert_eq!(b_view_of_a.hot_used_bytes, Some(1024)); 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(),] ); assert_eq!(b_view_of_a.hot_fill_ratio(), Some(1024.0 / 1_000_000.0)); // And what A sees for B — zone should be lan-1g. let a_view_of_b = gossip_a.peer("b").await.expect("A has B"); assert_eq!(a_view_of_b.zone, "lan-1g"); assert_eq!(a_view_of_b.rpc_lan, Some(loopback(port_b + 1))); // Zone filtering. let a_fabric = gossip_a.peers_in_zone("fabric-10g").await; assert!(a_fabric.is_empty(), "A alone in its zone from its view"); let a_lan_1g = gossip_a.peers_in_zone("lan-1g").await; assert_eq!(a_lan_1g.len(), 1); assert_eq!(a_lan_1g[0].name, "b"); gossip_a.shutdown(); gossip_b.shutdown(); } #[tokio::test] async fn peers_excludes_self() { let port = next_lan_port(); let cfg = ClusterConfig { zone: "fabric-10g".into(), bind_lan: Some(loopback(port)), bind_tailscale: None, peers: vec![], bind_rpc_lan: None, 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. let peers = g.peers().await; assert!(peers.is_empty(), "solo cluster reports zero peers"); 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, prom_bind: 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, prom_bind: 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, rustc_release: 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 { name: "x".into(), zone: "z".into(), rpc_lan: None, rpc_tailscale: None, hot_used_bytes: None, hot_max_bytes: Some(1000), 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, rustc_release: None, }; assert_eq!(base.hot_fill_ratio(), None, "no used → None"); let with_zero_max = PeerView { hot_used_bytes: Some(500), hot_max_bytes: Some(0), ..base.clone() }; assert_eq!( with_zero_max.hot_fill_ratio(), None, "max=0 → None (no div by zero)" ); let half_full = PeerView { hot_used_bytes: Some(500), hot_max_bytes: Some(1000), ..base }; assert_eq!(half_full.hot_fill_ratio(), Some(0.5)); } }