//! 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::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; 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); /// 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 /// 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, /// Local blob store, when `blob_store_root` was provided at start /// time. `None` means this node runs gossip-only (Blob RPCs return /// [`crate::cluster::rpc::ErrorCode::NotConfigured`]). pub blob_store: Option>, /// Local ref store (Phase 5b). Backs the `GetRef`/`PutRef` RPCs /// used by the fingerprint-keyed cargo cache. Opened automatically /// alongside the blob store — a node with one gets the other, so /// the entire Phase 5 substrate is enabled by a single config field. pub ref_store: Option>, /// Local tag store (Phase 5d). Backs `PutTag` / `GetTag` / /// `DeleteTag` / `ListTags` — the human-readable pin layer over /// raw refs. Opened alongside the blob store for the same reason. pub tag_store: Option>, /// RPC router. Held so the cache-metric ticker can call /// `router.metrics()` without duplicating an `Arc` /// field. `None` when RPC never came up (no `[cluster.tls]`). pub router: Option>, /// QUIC accept-loop task. `None` when `[cluster.tls]` was absent /// and RPC therefore didn't come up. accept_task: Option>, /// 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>, /// Phase 5j: Prometheus `/metrics` HTTP server. `None` when /// `cluster.prom_bind` was absent or no router exists (nothing to /// scrape). prom_server: Option, /// Field finding 2026-07-12: periodic orphan-chunk GC. `None` when /// `cluster.gc_interval_hours` is unset or the daemon has no blob /// store (nothing to sweep). gc_task: Option>, } 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, blob_store_root: Option, ) -> Result { 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; // Field finding 2026-07-12: publish `rustc --version` so peers // can flag toolchain drift before wasting a build on a cache // that will silo. Best-effort — a node with no rustc on PATH // simply doesn't advertise; peer-metrics prints "unknown". if let Some(release) = detect_rustc_release() { tracing::info!(rustc = %release, "publishing rustc release into gossip"); gossip.set_rustc_release(release).await; } else { tracing::info!("rustc not detected on PATH; skipping rustc.release gossip key"); } // Open the local blob store if a root path was supplied. Kept // outside the TLS branch: a node can serve blobs to callers // without RPC (via in-process API) or over RPC (once TLS is // configured too). let blob_store: Option> = match blob_store_root.as_ref() { Some(root) => { let store = BlobStore::open(root.clone()) .with_context(|| format!("opening blob store at {}", root.display()))?; tracing::info!("blob store opened at {}", root.display()); Some(Arc::new(store)) } None => { tracing::info!("no blob store configured; Blob RPCs will return NotConfigured"); None } }; // Ref store lives under `/refs-db` so the entire // Phase 5 substrate is enabled by a single config field. // Nodes without a blob store don't get a ref store either; // the fingerprint cache is meaningless without content. let ref_store: Option> = match blob_store_root.as_ref() { Some(root) => { let refs_dir = root.join("refs-db"); let store = RefStore::open(refs_dir.clone()) .with_context(|| format!("opening ref store at {}", refs_dir.display()))?; tracing::info!("ref store opened at {}", refs_dir.display()); Some(Arc::new(store)) } None => None, }; // Tag store (Phase 5d) sits next to the ref store. Same // rationale: the whole Phase 5 substrate follows blob_store_root. let tag_store: Option> = match blob_store_root.as_ref() { Some(root) => { let tags_dir = root.join("tags-db"); let store = TagStore::open(tags_dir.clone()) .with_context(|| format!("opening tag store at {}", tags_dir.display()))?; tracing::info!("tag store opened at {}", tags_dir.display()); Some(Arc::new(store)) } None => None, }; // RPC server + accept loop — only when TLS material is configured. // 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, and // ref-forwarding only activates when the outbound QUIC client // can be constructed (needs TLS material). let (router, accept_task) = match &cluster.tls { Some(tls) => { // Load identity twice — server takes ownership; outbound // client needs its own copy for TLS presentation on // ref-forwarding dials. let server_identity = NodeIdentity::from_pem_files(&tls.ca_cert, &tls.node_cert, &tls.node_key) .context("loading node identity from [cluster.tls]")?; let client_identity = NodeIdentity::from_pem_files(&tls.ca_cert, &tls.node_cert, &tls.node_key) .context("loading second node identity for outbound QUIC client")?; let bind = cluster .rpc_lan() .or_else(|| cluster.rpc_tailscale()) .context("no RPC bind address (need bind_lan or bind_tailscale)")?; // Phase 8d (2026-07-14): when BOTH LAN and tailnet // addresses are configured, we also bind a second // QuicServer on the tailnet interface. The gossip // layer already publishes `rpc_tailscale` so peers // learn to dial it; without the second bind the // advertised address just refuses connections. let bind_tailnet = match (cluster.rpc_lan(), cluster.rpc_tailscale()) { (Some(lan), Some(ts)) if lan != ts => Some(ts), _ => None, }; let outbound_client = crate::cluster::transport::QuicClient::new( "0.0.0.0:0".parse().expect("literal 0.0.0.0:0 parses"), client_identity, ) .context("binding outbound QUIC client for ref-forwarding")?; let outbound_client = Arc::new(outbound_client); 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()); } r = r.with_outbound_client(outbound_client); // Phase 9 R1a wiring: enable RepoEnsure/RepoRelease when // the daemon has a blob_store_root (which is the // canonical anchor for all fleet on-disk state). Repos // materialize under /repos//… if let Some(root) = blob_store_root.as_ref() { let repo_root = root.join("repos"); let _ = std::fs::create_dir_all(&repo_root); r = r.with_repo_root(repo_root); } let router = Arc::new(r); let server = QuicServer::bind(bind, server_identity).context("binding QUIC RPC server")?; tracing::info!( "cluster RPC server listening on {} (ref-forwarding enabled)", bind ); let router_for_accept = router.clone(); let task = tokio::spawn(async move { accept_forever(server, router_for_accept).await; }); // Phase 8d: second bind for the tailnet interface, // when configured. Shares the same identity + router // as the LAN listener — connections from either side // hit the same handlers. if let Some(ts_addr) = bind_tailnet { let ts_identity = NodeIdentity::from_pem_files(&tls.ca_cert, &tls.node_cert, &tls.node_key) .context( "loading node identity for tailnet QUIC server", )?; match QuicServer::bind(ts_addr, ts_identity) { Ok(ts_server) => { tracing::info!( "cluster RPC server also listening on {} (tailnet)", ts_addr ); let router_for_ts = router.clone(); tokio::spawn(async move { accept_forever(ts_server, router_for_ts).await; }); } Err(e) => tracing::warn!( error = %e, addr = %ts_addr, "tailnet RPC bind failed; LAN listener still active" ), } } (Some(router), Some(task)) } None => { tracing::info!("cluster: no [cluster.tls] configured; RPC disabled"); (None, 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; } }); // 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 }; // 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, }; // Field finding 2026-07-12: periodic orphan-chunk GC. // Bounded by chunks + manifests on disk; safe to run any time // and interruptible (only unreferenced chunks get deleted). let gc_task = match (&blob_store, cluster.gc_interval_hours) { (Some(store), Some(hours)) if hours > 0 => { let store = store.clone(); let tag_store_for_gc = tag_store.clone(); // Phase 7d follow-on: snapshot store is under the // same root as the blob store. Open once here so the // ticker doesn't pay the fs setup cost every tick. let snapshot_store_for_gc = blob_store_root .as_ref() .and_then(|root| { crate::cluster::snapshot::SnapshotStore::open(root.clone()).ok() }); let interval = Duration::from_secs(hours * 3600); let max_gb = cluster.blob_max_gb; Some(tokio::spawn(async move { let mut ticker = tokio::time::interval(interval); // Skip the immediate first tick — no point running GC // on a fresh daemon. ticker.tick().await; loop { ticker.tick().await; match store.gc_orphan_chunks().await { Ok(r) => tracing::info!( chunks_scanned = r.chunks_scanned, chunks_removed = r.chunks_removed, bytes_reclaimed = r.bytes_reclaimed, "auto-GC swept orphan chunks" ), Err(e) => { tracing::warn!(error = %e, "auto-GC failed; will retry next tick") } } // Field finding 2026-07-12: if configured with a // size cap, follow the orphan sweep with LRU // eviction. Orphan-only never frees blobs whose // manifest is still on disk — this is the piece // that actually bounds growth. if let Some(gb) = max_gb { let cap = gb.saturating_mul(1024 * 1024 * 1024); // Phase 4 (2026-07-13): pin-aware eviction. // Any blob referenced by a tag (stamped or // legacy) survives; pins act as retention // markers so operators can `claw-cargo pin` // a build and know it won't be evicted by // the size cap. let now_unix = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_secs()) .unwrap_or(0); let mut pinned = match &tag_store_for_gc { Some(ts) => { // Phase 4b: prune expired pins first so // the pin set reflects the wall-clock // moment we're about to evict at. let _ = ts .prune_expired_stamped_at(now_unix) .await; let raw = ts .pinned_blob_values_at(now_unix) .await .unwrap_or_default(); raw.into_iter() .map(crate::cluster::blob::BlobId::from_bytes) .collect() } None => std::collections::HashSet::new(), }; // Phase 7d follow-on: snapshots pin their // referenced blobs. Union in every blob_id // captured by any snapshot; the eviction // routine sees the merged set. let mut snapshot_pin_count = 0usize; if let Some(ss) = &snapshot_store_for_gc { if let Ok(snaps) = ss.pinned_blob_ids().await { snapshot_pin_count = snaps.len(); pinned.extend(snaps); } } match store .evict_to_size_cap_with_pins(cap, &pinned) .await { Ok(r) if r.chunks_removed > 0 => tracing::info!( chunks_removed = r.chunks_removed, bytes_reclaimed = r.bytes_reclaimed, max_gb = gb, pinned_blobs = pinned.len(), snapshot_pins = snapshot_pin_count, "auto-GC evicted LRU blobs to hit size cap" ), Ok(_) => {} // under cap already; keep quiet Err(e) => tracing::warn!( error = %e, "auto-GC eviction failed; will retry next tick" ), } } } })) } _ => None, }; Ok(Self { gossip, blob_store, ref_store, tag_store, router, accept_task, metric_task, cache_metric_task, prom_server, gc_task, }) } /// Address the Prometheus `/metrics` server actually bound to. /// `None` when the server was never started. pub fn prom_addr(&self) -> Option { 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() } /// Whether a local ref store was configured at start time. pub fn ref_store_enabled(&self) -> bool { self.ref_store.is_some() } /// 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(); if let Some(task) = self.cache_metric_task { task.abort(); } if let Some(server) = self.prom_server { server.abort(); } if let Some(task) = self.gc_task { 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) { 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 } /// Field finding 2026-07-12: probe `rustc --version` at startup so the /// daemon can advertise its toolchain over gossip. Best-effort; if /// rustc isn't on PATH we return `None` and skip the publish. /// /// Returns the "release" component only — for `rustc 1.97.0 (...)` /// that's `1.97.0`. Matches what fingerprints care about most: a bump /// in the major/minor version guarantees a different fingerprint. fn detect_rustc_release() -> Option { let output = std::process::Command::new("rustc") .arg("--version") .output() .ok()?; if !output.status.success() { return None; } let line = std::str::from_utf8(&output.stdout).ok()?.trim(); // Format: `rustc 1.97.0 (2d8144b78 2026-07-07)` — second whitespace // token is the release. line.split_whitespace().nth(1).map(|s| s.to_string()) } #[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+). /// /// Bump by 2 so each returned port `p` also implicitly reserves /// `p + 1` — RPC binds to `bind_lan.port + 1` (see /// `ClusterConfig::rpc_lan`), so returning consecutive ports would /// have one test's RPC step on the next test's gossip. static NEXT_PORT: AtomicU16 = AtomicU16::new(44001); fn next_port() -> u16 { NEXT_PORT.fetch_add(2, 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, None, ) .await .unwrap(); assert!(!svc.rpc_enabled(), "no [cluster.tls] → RPC 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"); 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, None) .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, None, ) .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, None, ) .await .unwrap(); let svc_b = ClusterServices::start( &cfg_b, "b".into(), tmp.path().to_path_buf(), 1_000_000, None, ) .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(); } #[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 // 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 // through correctly. Node A runs with a blob store; B dials it // and puts + gets a blob over real QUIC + mTLS. use crate::cluster::rpc::{call_blob_get, call_blob_put}; use crate::cluster::transport::{FleetCa, NodeIdentity, QuicClient}; 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 b_tls_dir = tmp.path().join("b-tls"); let blob_root = tmp.path().join("a-blobs"); let ca = FleetCa::generate("test CA").unwrap(); ca.save(&ca_dir).unwrap(); ca.sign_leaf_to_pem("a", &a_tls_dir).unwrap(); ca.sign_leaf_to_pem("b", &b_tls_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_tls_dir.join("ca.crt"), node_cert: a_tls_dir.join("node.crt"), node_key: a_tls_dir.join("node.key"), }), blob_store_root: Some(blob_root.clone()), ..Default::default() }; let hot_dir = tmp.path().join("a-hot"); std::fs::create_dir_all(&hot_dir).unwrap(); let svc = ClusterServices::start( &cfg_a, "a".into(), hot_dir, 1_000_000, Some(blob_root.clone()), ) .await .unwrap(); assert!(svc.rpc_enabled()); assert!(svc.blob_store_enabled()); let id_b = NodeIdentity::from_pem_dir(&b_tls_dir).unwrap(); let client = QuicClient::new(loopback(0), id_b).unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; let rpc_addr = cfg_a.rpc_lan().unwrap(); let conn = client.connect(rpc_addr, "a").await.unwrap(); let payload: &[u8] = b"cross-node blob over rpc"; let id = call_blob_put(&conn, payload).await.unwrap(); let round = call_blob_get(&conn, &id).await.unwrap().unwrap(); assert_eq!(round, payload); // Independently confirm the bytes are physically on disk in A's // store — proves the RPC didn't just echo back but actually // wrote through to BlobStore. let local_round = svc .blob_store .as_ref() .unwrap() .get_bytes(&id) .await .unwrap(); assert_eq!(local_round.as_deref(), Some(payload)); conn.close(quinn::VarInt::from_u32(0), b"done"); client.shutdown().await; svc.shutdown(); } }