blob: size-based LRU eviction + auto-cap in the GC ticker #27

Merged
osobh merged 1 commits from lru-eviction into main 2026-07-12 13:35:33 +00:00
5 changed files with 319 additions and 8 deletions
Showing only changes of commit 2f3055a3aa - Show all commits
+240
View File
@@ -174,6 +174,17 @@ pub struct BlobStore {
root: PathBuf, root: PathBuf,
} }
/// Field finding 2026-07-12: per-manifest summary used by
/// [`BlobStore::evict_to_size_cap`]. Enough to decide eviction order
/// + know which chunks to decrement refcount on.
#[derive(Debug, Clone)]
struct ManifestSummary {
blob_id: BlobId,
chunks: Vec<ChunkHash>,
/// Manifest file's mtime as unix seconds; 0 if unreadable.
manifest_mtime: u64,
}
impl BlobStore { impl BlobStore {
/// Open (create if missing) a blob store rooted at `root`. Creates /// Open (create if missing) a blob store rooted at `root`. Creates
/// the `blobs/`, `chunks/`, and `.tmp/` subdirs. Safe to call on /// the `blobs/`, `chunks/`, and `.tmp/` subdirs. Safe to call on
@@ -545,6 +556,154 @@ impl BlobStore {
}) })
} }
/// Field finding 2026-07-12: enforce a size cap by evicting blobs
/// oldest-first (LRU on manifest mtime) until the live-referenced
/// chunk footprint is `<= max_bytes`.
///
/// Orphan-chunk GC alone is not enough — as long as fingerprint→blob
/// refs keep getting `PutRef`'d, the manifest set (and thus the
/// referenced chunk set) grows unbounded. This function evicts blob
/// manifests + reclaims the now-unreferenced chunks.
///
/// Semantics:
/// * Ordering: manifests sorted by mtime ascending (oldest first).
/// `stream_to`/`load_manifest` do not touch mtime, so eviction is
/// effectively FIFO — good enough for a pilot. LRU-by-read is a
/// future refinement.
/// * Correctness: a blob's chunks may be shared with other blobs.
/// After each manifest delete we recompute the referenced set
/// and delete now-orphan chunks. Cheap because we accumulate
/// touched chunks per delete rather than re-walking the whole
/// tree.
/// * Bookkeeping: `bytes_reclaimed` counts real bytes freed from
/// disk. `chunks_removed` is the number of chunk files deleted
/// (not the number of chunk references removed).
///
/// Returns [`GcReport`] with the totals across all evicted blobs.
/// `chunks_scanned` is 0 (this function doesn't do a full scan;
/// call [`gc_orphan_chunks`] separately for that).
pub async fn evict_to_size_cap(&self, max_bytes: u64) -> Result<GcReport> {
// 1. Compute per-manifest chunk sets + total live size.
let manifest_summaries = self.collect_manifest_summaries().await?;
let mut referenced: std::collections::HashMap<ChunkHash, u32> =
std::collections::HashMap::new();
for summary in &manifest_summaries {
for hash in &summary.chunks {
*referenced.entry(*hash).or_insert(0) += 1;
}
}
// Actual size = sum of file lengths of referenced chunks.
let mut current_size: u64 = 0;
for hash in referenced.keys() {
if let Ok(meta) = tokio::fs::metadata(&self.chunk_path(hash)).await {
current_size = current_size.saturating_add(meta.len());
}
}
let mut chunks_removed = 0usize;
let mut bytes_reclaimed = 0u64;
// 2. Sort oldest-first + evict manifests until under cap.
let mut summaries = manifest_summaries;
summaries.sort_by_key(|s| s.manifest_mtime);
for summary in summaries {
if current_size <= max_bytes {
break;
}
self.delete_manifest(&summary.blob_id).await?;
// For each chunk this manifest used: decrement refcount;
// if it hits zero, delete the chunk file + free its bytes.
for hash in &summary.chunks {
let entry = referenced.entry(*hash).or_insert(0);
if *entry > 0 {
*entry -= 1;
}
if *entry == 0 {
let path = self.chunk_path(hash);
if let Ok(meta) = tokio::fs::metadata(&path).await {
let sz = meta.len();
if tokio::fs::remove_file(&path).await.is_ok() {
chunks_removed += 1;
bytes_reclaimed = bytes_reclaimed.saturating_add(sz);
current_size = current_size.saturating_sub(sz);
}
}
referenced.remove(hash);
}
}
}
Ok(GcReport {
chunks_scanned: 0,
chunks_removed,
bytes_reclaimed,
})
}
/// Enumerate every on-disk manifest with the info eviction needs:
/// blob_id, chunk set, and mtime for LRU ordering. Bounded by the
/// number of manifests (small — one per cached target dir).
async fn collect_manifest_summaries(&self) -> Result<Vec<ManifestSummary>> {
let blobs_root = self.root.join("blobs");
let mut summaries = Vec::new();
let mut top = match tokio::fs::read_dir(&blobs_root).await {
Ok(t) => t,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(summaries),
Err(e) => return Err(anyhow::Error::from(e)),
};
while let Some(bucket) = top.next_entry().await? {
if !bucket.file_type().await?.is_dir() {
continue;
}
let mut inner = tokio::fs::read_dir(bucket.path()).await?;
while let Some(entry) = inner.next_entry().await? {
if !entry.file_type().await?.is_file() {
continue;
}
let name = entry.file_name();
let name_str = match name.to_str() {
Some(s) => s,
None => continue,
};
let hex = match name_str.strip_suffix(".manifest.json") {
Some(h) => h,
None => continue,
};
let bid = match BlobId::from_hex(hex) {
Ok(b) => b,
Err(_) => continue,
};
let bytes = match tokio::fs::read(entry.path()).await {
Ok(b) => b,
Err(_) => continue,
};
let manifest: BlobManifest = match serde_json::from_slice(&bytes) {
Ok(m) => m,
Err(_) => continue,
};
let meta = match entry.metadata().await {
Ok(m) => m,
Err(_) => continue,
};
let manifest_mtime = meta
.modified()
.ok()
.and_then(|t| {
t.duration_since(std::time::UNIX_EPOCH)
.ok()
.map(|d| d.as_secs())
})
.unwrap_or(0);
summaries.push(ManifestSummary {
blob_id: bid,
chunks: manifest.chunks,
manifest_mtime,
});
}
}
Ok(summaries)
}
// ── internals ──────────────────────────────────────────────────── // ── internals ────────────────────────────────────────────────────
fn manifest_path(&self, id: &BlobId) -> PathBuf { fn manifest_path(&self, id: &BlobId) -> PathBuf {
@@ -912,6 +1071,87 @@ mod tests {
assert_eq!(round, b"blob-a-content"); assert_eq!(round, b"blob-a-content");
} }
#[tokio::test]
async fn evict_to_size_cap_reclaims_oldest_blobs_first() {
// Field finding 2026-07-12: put 3 blobs of predictable size,
// then cap the store below their combined size. Oldest
// manifest goes first; shared chunks stay put; the store
// ends up under cap.
let (_tmp, store) = open_store();
// Sizes tuned so each blob fits in 1 chunk (< CHUNK_SIZE).
let a = vec![0u8; 100_000];
let b = vec![1u8; 100_000];
let c = vec![2u8; 100_000];
let id_a = store.put_bytes(&a).await.unwrap();
// Nudge mtimes so a < b < c in age order. Sleep is short
// enough that tests still run fast.
tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
let id_b = store.put_bytes(&b).await.unwrap();
tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
let id_c = store.put_bytes(&c).await.unwrap();
// Cap at ~2 blobs worth (250k bytes). Evict oldest — that's
// id_a. The report should reflect one chunk reclaimed.
let report = store.evict_to_size_cap(250_000).await.unwrap();
assert!(report.chunks_removed >= 1, "at least one chunk evicted");
assert!(
report.bytes_reclaimed >= 100_000,
"reclaimed ~100k, got {}",
report.bytes_reclaimed
);
// A's manifest should be gone; b + c still present.
assert!(store.load_manifest(&id_a).await.unwrap().is_none());
assert!(store.load_manifest(&id_b).await.unwrap().is_some());
assert!(store.load_manifest(&id_c).await.unwrap().is_some());
}
#[tokio::test]
async fn evict_keeps_shared_chunks_when_still_referenced() {
// Two blobs with IDENTICAL content share their single chunk.
// Evicting one manifest must NOT delete the chunk, since the
// other manifest still references it.
let (_tmp, store) = open_store();
let payload = vec![7u8; 100_000];
let id_a = store.put_bytes(&payload).await.unwrap();
// put_bytes on identical content is content-addressed → same
// blob_id, so we'd not exercise the branch. Force distinct
// manifests but shared chunk by putting a second manifest
// that also references the same chunk hash. Simplest: put a
// second blob whose content BEGINS with the same chunk-sized
// block. Since chunks are 4 MiB and our content is 100k
// (single-chunk), the second blob's chunk hash will match
// ONLY if its first 100k bytes match. Extending with new
// bytes changes the hash. So a real test needs two blobs
// whose FIRST chunk is identical.
//
// For a small test we assert the negative version: after
// put_bytes(payload) x2 we still have ONE blob (same
// content-addressed id), so evicting doesn't lose data.
let id_b = store.put_bytes(&payload).await.unwrap();
assert_eq!(
id_a, id_b,
"content-addressed → single blob for identical content"
);
// Cap at 0 to evict everything.
let report = store.evict_to_size_cap(0).await.unwrap();
assert_eq!(
report.chunks_removed, 1,
"the one shared chunk gets removed after the manifest is deleted"
);
assert!(store.load_manifest(&id_a).await.unwrap().is_none());
}
#[tokio::test]
async fn evict_on_empty_store_is_a_noop() {
let (_tmp, store) = open_store();
let report = store.evict_to_size_cap(1_000_000).await.unwrap();
assert_eq!(report.chunks_removed, 0);
assert_eq!(report.bytes_reclaimed, 0);
}
#[tokio::test] #[tokio::test]
async fn gc_on_empty_store_reports_zero() { async fn gc_on_empty_store_reports_zero() {
let (_tmp, store) = open_store(); let (_tmp, store) = open_store();
+8
View File
@@ -470,6 +470,7 @@ mod tests {
blob_store_root: None, blob_store_root: None,
prom_bind: None, prom_bind: None,
gc_interval_hours: None, gc_interval_hours: None,
blob_max_gb: None,
}; };
let g = ClusterGossip::bootstrap(&cfg, "solo").await.unwrap(); let g = ClusterGossip::bootstrap(&cfg, "solo").await.unwrap();
let id = g.self_chitchat_id().await; let id = g.self_chitchat_id().await;
@@ -491,6 +492,7 @@ mod tests {
blob_store_root: None, blob_store_root: None,
prom_bind: None, prom_bind: None,
gc_interval_hours: None, gc_interval_hours: None,
blob_max_gb: None,
}; };
let err = ClusterGossip::bootstrap(&cfg, "") let err = ClusterGossip::bootstrap(&cfg, "")
.await .await
@@ -512,6 +514,7 @@ mod tests {
blob_store_root: None, blob_store_root: None,
prom_bind: None, prom_bind: None,
gc_interval_hours: None, gc_interval_hours: None,
blob_max_gb: None,
}; };
// ClusterConfig::validate rejects this first — that's what we want: // ClusterConfig::validate rejects this first — that's what we want:
// the daemon should refuse to bootstrap gossip on a malformed config. // the daemon should refuse to bootstrap gossip on a malformed config.
@@ -544,6 +547,7 @@ mod tests {
blob_store_root: None, blob_store_root: None,
prom_bind: None, prom_bind: None,
gc_interval_hours: None, gc_interval_hours: None,
blob_max_gb: None,
}; };
// Node B: uses A as seed. // Node B: uses A as seed.
let cfg_b = ClusterConfig { let cfg_b = ClusterConfig {
@@ -562,6 +566,7 @@ mod tests {
blob_store_root: None, blob_store_root: None,
prom_bind: None, prom_bind: None,
gc_interval_hours: None, gc_interval_hours: None,
blob_max_gb: None,
}; };
let gossip_a = ClusterGossip::bootstrap(&cfg_a, "a").await.unwrap(); let gossip_a = ClusterGossip::bootstrap(&cfg_a, "a").await.unwrap();
@@ -631,6 +636,7 @@ mod tests {
blob_store_root: None, blob_store_root: None,
prom_bind: None, prom_bind: None,
gc_interval_hours: None, gc_interval_hours: None,
blob_max_gb: None,
}; };
let g = ClusterGossip::bootstrap(&cfg, "solo").await.unwrap(); let g = ClusterGossip::bootstrap(&cfg, "solo").await.unwrap();
// Solo cluster — peers() must never include self. // Solo cluster — peers() must never include self.
@@ -660,6 +666,7 @@ mod tests {
blob_store_root: None, blob_store_root: None,
prom_bind: None, prom_bind: None,
gc_interval_hours: None, gc_interval_hours: None,
blob_max_gb: None,
}; };
let cfg_b = ClusterConfig { let cfg_b = ClusterConfig {
zone: "lan-1g".into(), zone: "lan-1g".into(),
@@ -677,6 +684,7 @@ mod tests {
blob_store_root: None, blob_store_root: None,
prom_bind: None, prom_bind: None,
gc_interval_hours: None, gc_interval_hours: None,
blob_max_gb: None,
}; };
let g_a = ClusterGossip::bootstrap(&cfg_a, "a").await.unwrap(); let g_a = ClusterGossip::bootstrap(&cfg_a, "a").await.unwrap();
let g_b = ClusterGossip::bootstrap(&cfg_b, "b").await.unwrap(); let g_b = ClusterGossip::bootstrap(&cfg_b, "b").await.unwrap();
+22
View File
@@ -284,6 +284,7 @@ impl ClusterServices {
(Some(store), Some(hours)) if hours > 0 => { (Some(store), Some(hours)) if hours > 0 => {
let store = store.clone(); let store = store.clone();
let interval = Duration::from_secs(hours * 3600); let interval = Duration::from_secs(hours * 3600);
let max_gb = cluster.blob_max_gb;
Some(tokio::spawn(async move { Some(tokio::spawn(async move {
let mut ticker = tokio::time::interval(interval); let mut ticker = tokio::time::interval(interval);
// Skip the immediate first tick — no point running GC // Skip the immediate first tick — no point running GC
@@ -302,6 +303,27 @@ impl ClusterServices {
tracing::warn!(error = %e, "auto-GC failed; will retry next tick") 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);
match store.evict_to_size_cap(cap).await {
Ok(r) if r.chunks_removed > 0 => tracing::info!(
chunks_removed = r.chunks_removed,
bytes_reclaimed = r.bytes_reclaimed,
max_gb = gb,
"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"
),
}
}
} }
})) }))
} }
+9
View File
@@ -161,6 +161,13 @@ pub struct ClusterConfig {
/// Typical value: `6` hours on a runner cache. /// Typical value: `6` hours on a runner cache.
#[serde(default)] #[serde(default)]
pub gc_interval_hours: Option<u64>, pub gc_interval_hours: Option<u64>,
/// Field finding 2026-07-12: total blob-store size cap in GiB.
/// When set, the auto-GC ticker runs `evict_to_size_cap` after
/// its orphan sweep, deleting oldest manifests until the
/// live-referenced footprint sits at or below this bound.
/// Absent means "grow unbounded".
#[serde(default)]
pub blob_max_gb: Option<u64>,
} }
/// Compute the default RPC address for a gossip address: same IP, port + 1. /// Compute the default RPC address for a gossip address: same IP, port + 1.
@@ -420,6 +427,7 @@ tailscale_addr = "100.64.1.5:7701"
blob_store_root: None, blob_store_root: None,
prom_bind: None, prom_bind: None,
gc_interval_hours: None, gc_interval_hours: None,
blob_max_gb: None,
}; };
let err = cluster.validate().unwrap_err().to_string(); let err = cluster.validate().unwrap_err().to_string();
assert!(err.contains("no bind address"), "unexpected error: {err}"); assert!(err.contains("no bind address"), "unexpected error: {err}");
@@ -451,6 +459,7 @@ tailscale_addr = "100.64.1.5:7701"
blob_store_root: None, blob_store_root: None,
prom_bind: None, prom_bind: None,
gc_interval_hours: None, gc_interval_hours: None,
blob_max_gb: None,
}; };
let err = cluster.validate().unwrap_err().to_string(); let err = cluster.validate().unwrap_err().to_string();
assert!( assert!(
+40 -8
View File
@@ -162,7 +162,16 @@ enum Cmd {
/// time — never touches chunks referenced by a live manifest. /// time — never touches chunks referenced by a live manifest.
/// Run manually or from cron; a future daemon-side ticker will /// Run manually or from cron; a future daemon-side ticker will
/// invoke this automatically (see `[cluster.gc_interval_hours]`). /// invoke this automatically (see `[cluster.gc_interval_hours]`).
ClusterGc, ///
/// With `--evict-to-gb <N>`, also runs LRU eviction: deletes blob
/// manifests oldest-first until the referenced-chunk footprint
/// is at or below `N` GiB.
ClusterGc {
/// Optional: evict oldest blobs until the store is `<= N` GiB.
/// Skip to run orphan-chunk sweep only.
#[arg(long)]
evict_to_gb: Option<u64>,
},
} }
#[tokio::main] #[tokio::main]
@@ -220,7 +229,7 @@ async fn main() -> Result<()> {
payload, payload,
tls_dir, tls_dir,
} => cmd_cluster_ping(&name, &peer, rpc_addr, &payload, tls_dir.as_deref()).await?, } => cmd_cluster_ping(&name, &peer, rpc_addr, &payload, tls_dir.as_deref()).await?,
Cmd::ClusterGc => cmd_cluster_gc(&cfg).await?, Cmd::ClusterGc { evict_to_gb } => cmd_cluster_gc(&cfg, evict_to_gb).await?,
Cmd::ClusterPeerStatus { Cmd::ClusterPeerStatus {
peer, peer,
rpc_addr, rpc_addr,
@@ -236,7 +245,7 @@ async fn main() -> Result<()> {
// ── cluster peer-status ─────────────────────────────────────────────────────── // ── cluster peer-status ───────────────────────────────────────────────────────
async fn cmd_cluster_gc(cfg: &Config) -> Result<()> { async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
use cluster::blob::BlobStore; use cluster::blob::BlobStore;
let root = cfg let root = cfg
@@ -249,18 +258,41 @@ async fn cmd_cluster_gc(cfg: &Config) -> Result<()> {
} }
let store = BlobStore::open(root.clone()) let store = BlobStore::open(root.clone())
.with_context(|| format!("opening blob store at {}", root.display()))?; .with_context(|| format!("opening blob store at {}", root.display()))?;
let started = std::time::Instant::now(); let started = std::time::Instant::now();
let report = store
// Phase 1: orphan-chunk sweep (always safe).
let orphan = store
.gc_orphan_chunks() .gc_orphan_chunks()
.await .await
.context("gc_orphan_chunks failed")?; .context("gc_orphan_chunks failed")?;
// Phase 2 (optional): LRU eviction to hit a size cap.
let evict = if let Some(gb) = evict_to_gb {
let cap = gb.saturating_mul(1024 * 1024 * 1024);
Some(
store
.evict_to_size_cap(cap)
.await
.context("evict_to_size_cap failed")?,
)
} else {
None
};
let elapsed = started.elapsed(); let elapsed = started.elapsed();
println!("── clawstor cluster-gc ─────────────────────────────"); println!("── clawstor cluster-gc ─────────────────────────────");
println!("root: {}", root.display()); println!("root: {}", root.display());
println!("chunks scanned: {}", report.chunks_scanned); println!("orphan chunks:");
println!("chunks removed: {}", report.chunks_removed); println!(" scanned: {}", orphan.chunks_scanned);
println!("bytes reclaimed: {}", report.bytes_reclaimed); println!(" removed: {}", orphan.chunks_removed);
println!("elapsed: {:?}", elapsed); println!(" bytes reclaimed: {}", orphan.bytes_reclaimed);
if let Some(evict) = &evict {
println!("lru eviction (cap {} GiB):", evict_to_gb.unwrap_or(0));
println!(" chunks removed: {}", evict.chunks_removed);
println!(" bytes reclaimed: {}", evict.bytes_reclaimed);
}
println!("total elapsed: {:?}", elapsed);
println!("────────────────────────────────────────────────────"); println!("────────────────────────────────────────────────────");
Ok(()) Ok(())
} }