Two pilot follow-ons: rustc drift warning + blob GC #26

Merged
osobh merged 1 commits from rustc-drift-warning-and-gc into main 2026-07-12 13:24:01 +00:00
6 changed files with 177 additions and 2 deletions
+60 -2
View File
@@ -47,7 +47,7 @@ use crate::cluster::build_cache::{
use crate::cluster::client_config::{ClientConfig, ResolvedClientConfig};
use crate::cluster::rpc::{
call_blob_get_stream, call_blob_put_stream, call_blob_stat, call_delete_tag, call_get_metrics,
prewarm_missing_chunks_between_parallel,
call_peer_status, prewarm_missing_chunks_between_parallel,
call_get_ref, call_get_tag, call_list_tags, call_put_ref, call_put_tag,
};
use crate::cluster::transport::{NodeIdentity, QuicClient};
@@ -495,6 +495,55 @@ async fn resolve_pin(
}
}
/// Field finding 2026-07-12: compare our rustc release against the
/// peer's advertised one on a cache miss and emit a WARN if they
/// don't match. Silos are correctness — the cache we're about to
/// upload won't be usable by any peer runner on a different rustc —
/// so surface the issue before the operator wastes a build.
///
/// Best-effort: absence of either release string is a shrug, not
/// an error.
async fn warn_on_rustc_drift(conn: &quinn::Connection) -> Result<()> {
let status = call_peer_status(conn).await?;
let peer_rustc = match status.local_rustc_release.as_deref() {
Some(r) if !r.is_empty() => r,
_ => return Ok(()),
};
let local = detect_local_rustc_release();
let local = match local.as_deref() {
Some(r) if !r.is_empty() => r,
_ => return Ok(()),
};
if peer_rustc != local {
tracing::warn!(
local_rustc = %local,
peer_rustc = %peer_rustc,
peer = %status.local_name,
"rustc release drift: fingerprint depends on rustc verbose output, \
so the cache we're about to publish will silo — any peer runner on \
a different rustc will miss on our fingerprint. Align toolchains \
via `rust-toolchain.toml` if you want cross-node cache sharing."
);
}
Ok(())
}
/// Field finding 2026-07-12: cheap local rustc release probe. Same
/// shape as the daemon's `detect_rustc_release`; kept independent so
/// claw-cargo doesn't have to link against the daemon crate. Returns
/// `None` when rustc isn't on PATH.
fn detect_local_rustc_release() -> Option<String> {
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();
line.split_whitespace().nth(1).map(|s| s.to_string())
}
// ── build ────────────────────────────────────────────────────────────
async fn cmd_build(args: BuildArgs) -> Result<()> {
@@ -523,7 +572,16 @@ async fn cmd_build(args: BuildArgs) -> Result<()> {
};
}
}
None => tracing::info!("cache MISS"),
None => {
tracing::info!("cache MISS");
// Field finding 2026-07-12: on a miss, ask the peer for
// its rustc release; if it differs from ours, warn — the
// cache we're about to populate will silo (peer's runners
// won't share this fingerprint). Cheap check, high value.
if let Err(e) = warn_on_rustc_drift(&conn).await {
tracing::debug!(error = %e, "rustc drift check skipped");
}
}
}
// Run cargo. Even on hit we run cargo — the workspace's own crates
+19
View File
@@ -312,6 +312,17 @@ impl ClusterGossip {
self.chitchat.lock().await.self_chitchat_id().clone()
}
/// Field finding 2026-07-12: read one of this node's own gossip
/// key-values. Used by [`crate::cluster::rpc::RpcRouter`] to fold
/// the local rustc release into `PeerStatusReply` so runners can
/// detect toolchain drift at build time.
pub async fn self_kv(&self, key: &str) -> Option<String> {
let cc = self.chitchat.lock().await;
let self_id = cc.self_chitchat_id().clone();
cc.node_state(&self_id)
.and_then(|s| s.get(key).map(|v| v.to_string()))
}
/// 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).
@@ -458,6 +469,7 @@ mod tests {
tls: None,
blob_store_root: None,
prom_bind: None,
gc_interval_hours: None,
};
let g = ClusterGossip::bootstrap(&cfg, "solo").await.unwrap();
let id = g.self_chitchat_id().await;
@@ -478,6 +490,7 @@ mod tests {
tls: None,
blob_store_root: None,
prom_bind: None,
gc_interval_hours: None,
};
let err = ClusterGossip::bootstrap(&cfg, "")
.await
@@ -498,6 +511,7 @@ mod tests {
tls: None,
blob_store_root: None,
prom_bind: None,
gc_interval_hours: None,
};
// ClusterConfig::validate rejects this first — that's what we want:
// the daemon should refuse to bootstrap gossip on a malformed config.
@@ -529,6 +543,7 @@ mod tests {
tls: None,
blob_store_root: None,
prom_bind: None,
gc_interval_hours: None,
};
// Node B: uses A as seed.
let cfg_b = ClusterConfig {
@@ -546,6 +561,7 @@ mod tests {
tls: None,
blob_store_root: None,
prom_bind: None,
gc_interval_hours: None,
};
let gossip_a = ClusterGossip::bootstrap(&cfg_a, "a").await.unwrap();
@@ -614,6 +630,7 @@ mod tests {
tls: None,
blob_store_root: None,
prom_bind: None,
gc_interval_hours: None,
};
let g = ClusterGossip::bootstrap(&cfg, "solo").await.unwrap();
// Solo cluster — peers() must never include self.
@@ -642,6 +659,7 @@ mod tests {
tls: None,
blob_store_root: None,
prom_bind: None,
gc_interval_hours: None,
};
let cfg_b = ClusterConfig {
zone: "lan-1g".into(),
@@ -658,6 +676,7 @@ mod tests {
tls: None,
blob_store_root: None,
prom_bind: None,
gc_interval_hours: None,
};
let g_a = ClusterGossip::bootstrap(&cfg_a, "a").await.unwrap();
let g_b = ClusterGossip::bootstrap(&cfg_b, "b").await.unwrap();
+14
View File
@@ -218,6 +218,12 @@ pub struct PeerStatusReply {
pub local_name: String,
/// The zone this node is in.
pub local_zone: String,
/// Field finding 2026-07-12: this node's own rustc release
/// (e.g. `1.97.0`). Absent when rustc isn't on the daemon's PATH.
/// Runners consult this before a cold build to detect toolchain
/// drift that would silo the produced cache.
#[serde(default)]
pub local_rustc_release: Option<String>,
/// Every peer this node knows about (live + dead-in-grace-window).
pub peers: Vec<PeerView>,
}
@@ -329,9 +335,17 @@ impl RpcRouter {
}
Method::PeerStatus => {
let peers = self.gossip.peers().await;
// Field finding 2026-07-12: expose our own rustc in
// the reply so `claw-cargo build` can warn on drift
// before wasting a full cold build.
let local_rustc_release = self
.gossip
.self_kv(crate::cluster::gossip::keys::RUSTC_RELEASE)
.await;
let reply = PeerStatusReply {
local_name: self.local_name.clone(),
local_zone: self.local_zone.clone(),
local_rustc_release,
peers,
};
let json = serde_json::to_vec(&reply)
+39
View File
@@ -80,6 +80,10 @@ pub struct ClusterServices {
/// `cluster.prom_bind` was absent or no router exists (nothing to
/// scrape).
prom_server: Option<PromServer>,
/// 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<JoinHandle<()>>,
}
impl std::fmt::Debug for ClusterServices {
@@ -273,6 +277,37 @@ impl ClusterServices {
(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 interval = Duration::from_secs(hours * 3600);
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")
}
}
}
}))
}
_ => None,
};
Ok(Self {
gossip,
blob_store,
@@ -283,6 +318,7 @@ impl ClusterServices {
metric_task,
cache_metric_task,
prom_server,
gc_task,
})
}
@@ -322,6 +358,9 @@ impl ClusterServices {
if let Some(server) = self.prom_server {
server.abort();
}
if let Some(task) = self.gc_task {
task.abort();
}
}
}
+9
View File
@@ -154,6 +154,13 @@ pub struct ClusterConfig {
/// group). Absent means "no scrape endpoint".
#[serde(default)]
pub prom_bind: Option<SocketAddr>,
/// Field finding 2026-07-12: how often the daemon runs
/// `gc_orphan_chunks` to reclaim disk from chunks no live
/// manifest references. `None` or `0` disables auto-GC — the
/// operator can still invoke `claw-store cluster-gc` by hand.
/// Typical value: `6` hours on a runner cache.
#[serde(default)]
pub gc_interval_hours: Option<u64>,
}
/// Compute the default RPC address for a gossip address: same IP, port + 1.
@@ -412,6 +419,7 @@ tailscale_addr = "100.64.1.5:7701"
tls: None,
blob_store_root: None,
prom_bind: None,
gc_interval_hours: None,
};
let err = cluster.validate().unwrap_err().to_string();
assert!(err.contains("no bind address"), "unexpected error: {err}");
@@ -442,6 +450,7 @@ tailscale_addr = "100.64.1.5:7701"
tls: None,
blob_store_root: None,
prom_bind: None,
gc_interval_hours: None,
};
let err = cluster.validate().unwrap_err().to_string();
assert!(
+36
View File
@@ -157,6 +157,12 @@ enum Cmd {
#[arg(long)]
tls_dir: PathBuf,
},
/// Field finding 2026-07-12: sweep orphan chunks from this node's
/// blob store (chunks referenced by no manifest). Safe to run any
/// time — never touches chunks referenced by a live manifest.
/// Run manually or from cron; a future daemon-side ticker will
/// invoke this automatically (see `[cluster.gc_interval_hours]`).
ClusterGc,
}
#[tokio::main]
@@ -214,6 +220,7 @@ async fn main() -> Result<()> {
payload,
tls_dir,
} => cmd_cluster_ping(&name, &peer, rpc_addr, &payload, tls_dir.as_deref()).await?,
Cmd::ClusterGc => cmd_cluster_gc(&cfg).await?,
Cmd::ClusterPeerStatus {
peer,
rpc_addr,
@@ -229,6 +236,35 @@ async fn main() -> Result<()> {
// ── cluster peer-status ───────────────────────────────────────────────────────
async fn cmd_cluster_gc(cfg: &Config) -> Result<()> {
use cluster::blob::BlobStore;
let root = cfg
.cluster
.as_ref()
.and_then(|c| c.blob_store_root.clone())
.context("cluster.blob_store_root not configured; nothing to GC")?;
if !root.is_dir() {
bail!("blob_store_root {} does not exist", root.display());
}
let store = BlobStore::open(root.clone())
.with_context(|| format!("opening blob store at {}", root.display()))?;
let started = std::time::Instant::now();
let report = store
.gc_orphan_chunks()
.await
.context("gc_orphan_chunks failed")?;
let elapsed = started.elapsed();
println!("── clawstor cluster-gc ─────────────────────────────");
println!("root: {}", root.display());
println!("chunks scanned: {}", report.chunks_scanned);
println!("chunks removed: {}", report.chunks_removed);
println!("bytes reclaimed: {}", report.bytes_reclaimed);
println!("elapsed: {:?}", elapsed);
println!("────────────────────────────────────────────────────");
Ok(())
}
async fn cmd_cluster_peer_status(
peer: &str,
rpc_addr: SocketAddr,