Phase 4a: pin-aware LRU eviction #43
@@ -583,6 +583,31 @@ impl BlobStore {
|
|||||||
/// `chunks_scanned` is 0 (this function doesn't do a full scan;
|
/// `chunks_scanned` is 0 (this function doesn't do a full scan;
|
||||||
/// call [`gc_orphan_chunks`] separately for that).
|
/// call [`gc_orphan_chunks`] separately for that).
|
||||||
pub async fn evict_to_size_cap(&self, max_bytes: u64) -> Result<GcReport> {
|
pub async fn evict_to_size_cap(&self, max_bytes: u64) -> Result<GcReport> {
|
||||||
|
// No pins → every manifest is a candidate. Delegate.
|
||||||
|
self.evict_to_size_cap_with_pins(
|
||||||
|
max_bytes,
|
||||||
|
&std::collections::HashSet::new(),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Phase 4 (2026-07-13): pin-aware LRU eviction. `pinned_blobs`
|
||||||
|
/// is the set of blob IDs the caller considers protected from
|
||||||
|
/// eviction — typically the set of every blob referenced by a
|
||||||
|
/// live tag. Pinned manifests are skipped entirely; their chunks
|
||||||
|
/// stay in the referenced set so shared chunks with evicted
|
||||||
|
/// blobs also survive.
|
||||||
|
///
|
||||||
|
/// The eviction pass may go under cap earlier than a pin-free
|
||||||
|
/// pass would — pinned blobs count against `max_bytes` but can't
|
||||||
|
/// be evicted to make room, so if the pinned footprint alone
|
||||||
|
/// exceeds the cap, we return without doing anything (the caller
|
||||||
|
/// is expected to raise `blob_max_gb` or drop pins).
|
||||||
|
pub async fn evict_to_size_cap_with_pins(
|
||||||
|
&self,
|
||||||
|
max_bytes: u64,
|
||||||
|
pinned_blobs: &std::collections::HashSet<BlobId>,
|
||||||
|
) -> Result<GcReport> {
|
||||||
// 1. Compute per-manifest chunk sets + total live size.
|
// 1. Compute per-manifest chunk sets + total live size.
|
||||||
let manifest_summaries = self.collect_manifest_summaries().await?;
|
let manifest_summaries = self.collect_manifest_summaries().await?;
|
||||||
let mut referenced: std::collections::HashMap<ChunkHash, u32> =
|
let mut referenced: std::collections::HashMap<ChunkHash, u32> =
|
||||||
@@ -604,12 +629,17 @@ impl BlobStore {
|
|||||||
let mut bytes_reclaimed = 0u64;
|
let mut bytes_reclaimed = 0u64;
|
||||||
|
|
||||||
// 2. Sort oldest-first + evict manifests until under cap.
|
// 2. Sort oldest-first + evict manifests until under cap.
|
||||||
|
// Skip pinned blobs entirely — their chunks stay in
|
||||||
|
// `referenced` so any shared chunks also stay put.
|
||||||
let mut summaries = manifest_summaries;
|
let mut summaries = manifest_summaries;
|
||||||
summaries.sort_by_key(|s| s.manifest_mtime);
|
summaries.sort_by_key(|s| s.manifest_mtime);
|
||||||
for summary in summaries {
|
for summary in summaries {
|
||||||
if current_size <= max_bytes {
|
if current_size <= max_bytes {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
if pinned_blobs.contains(&summary.blob_id) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
self.delete_manifest(&summary.blob_id).await?;
|
self.delete_manifest(&summary.blob_id).await?;
|
||||||
// For each chunk this manifest used: decrement refcount;
|
// For each chunk this manifest used: decrement refcount;
|
||||||
// if it hits zero, delete the chunk file + free its bytes.
|
// if it hits zero, delete the chunk file + free its bytes.
|
||||||
@@ -1152,6 +1182,69 @@ mod tests {
|
|||||||
assert_eq!(report.bytes_reclaimed, 0);
|
assert_eq!(report.bytes_reclaimed, 0);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn evict_with_pins_protects_pinned_blobs_from_eviction() {
|
||||||
|
// Phase 4: 3 blobs, mtime-ordered a < b < c. Pin the OLDEST
|
||||||
|
// (a) — normal LRU would evict a first. With pins, a survives
|
||||||
|
// and b (the next-oldest) is evicted instead.
|
||||||
|
let (_tmp, store) = open_store();
|
||||||
|
|
||||||
|
let a_bytes = vec![0u8; 100_000];
|
||||||
|
let b_bytes = vec![1u8; 100_000];
|
||||||
|
let c_bytes = vec![2u8; 100_000];
|
||||||
|
|
||||||
|
let id_a = store.put_bytes(&a_bytes).await.unwrap();
|
||||||
|
tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
|
||||||
|
let id_b = store.put_bytes(&b_bytes).await.unwrap();
|
||||||
|
tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
|
||||||
|
let id_c = store.put_bytes(&c_bytes).await.unwrap();
|
||||||
|
|
||||||
|
let mut pinned = std::collections::HashSet::new();
|
||||||
|
pinned.insert(id_a);
|
||||||
|
|
||||||
|
// Cap at ~2 blobs. Without pins, a would be evicted; with
|
||||||
|
// the pin, b goes instead.
|
||||||
|
let report = store
|
||||||
|
.evict_to_size_cap_with_pins(250_000, &pinned)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert!(report.chunks_removed >= 1);
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
store.load_manifest(&id_a).await.unwrap().is_some(),
|
||||||
|
"pinned blob a must survive"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
store.load_manifest(&id_b).await.unwrap().is_none(),
|
||||||
|
"next-oldest unpinned b was evicted"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
store.load_manifest(&id_c).await.unwrap().is_some(),
|
||||||
|
"newest c stays"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn evict_with_pins_stops_when_pinned_footprint_dominates() {
|
||||||
|
// Every blob is pinned → nothing to evict → eviction is a
|
||||||
|
// no-op regardless of `max_bytes`.
|
||||||
|
let (_tmp, store) = open_store();
|
||||||
|
let id_a = store.put_bytes(&vec![9u8; 100_000]).await.unwrap();
|
||||||
|
let id_b = store.put_bytes(&vec![8u8; 100_000]).await.unwrap();
|
||||||
|
|
||||||
|
let mut pinned = std::collections::HashSet::new();
|
||||||
|
pinned.insert(id_a);
|
||||||
|
pinned.insert(id_b);
|
||||||
|
|
||||||
|
let report = store
|
||||||
|
.evict_to_size_cap_with_pins(0, &pinned)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(report.chunks_removed, 0);
|
||||||
|
assert!(store.load_manifest(&id_a).await.unwrap().is_some());
|
||||||
|
assert!(store.load_manifest(&id_b).await.unwrap().is_some());
|
||||||
|
}
|
||||||
|
|
||||||
#[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();
|
||||||
|
|||||||
@@ -301,6 +301,7 @@ impl ClusterServices {
|
|||||||
let gc_task = match (&blob_store, cluster.gc_interval_hours) {
|
let gc_task = match (&blob_store, cluster.gc_interval_hours) {
|
||||||
(Some(store), Some(hours)) if hours > 0 => {
|
(Some(store), Some(hours)) if hours > 0 => {
|
||||||
let store = store.clone();
|
let store = store.clone();
|
||||||
|
let tag_store_for_gc = tag_store.clone();
|
||||||
let interval = Duration::from_secs(hours * 3600);
|
let interval = Duration::from_secs(hours * 3600);
|
||||||
let max_gb = cluster.blob_max_gb;
|
let max_gb = cluster.blob_max_gb;
|
||||||
Some(tokio::spawn(async move {
|
Some(tokio::spawn(async move {
|
||||||
@@ -328,11 +329,30 @@ impl ClusterServices {
|
|||||||
// that actually bounds growth.
|
// that actually bounds growth.
|
||||||
if let Some(gb) = max_gb {
|
if let Some(gb) = max_gb {
|
||||||
let cap = gb.saturating_mul(1024 * 1024 * 1024);
|
let cap = gb.saturating_mul(1024 * 1024 * 1024);
|
||||||
match store.evict_to_size_cap(cap).await {
|
// 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 pinned = match &tag_store_for_gc {
|
||||||
|
Some(ts) => {
|
||||||
|
let raw = ts.pinned_blob_values().await.unwrap_or_default();
|
||||||
|
raw.into_iter()
|
||||||
|
.map(crate::cluster::blob::BlobId::from_bytes)
|
||||||
|
.collect()
|
||||||
|
}
|
||||||
|
None => std::collections::HashSet::new(),
|
||||||
|
};
|
||||||
|
match store
|
||||||
|
.evict_to_size_cap_with_pins(cap, &pinned)
|
||||||
|
.await
|
||||||
|
{
|
||||||
Ok(r) if r.chunks_removed > 0 => tracing::info!(
|
Ok(r) if r.chunks_removed > 0 => tracing::info!(
|
||||||
chunks_removed = r.chunks_removed,
|
chunks_removed = r.chunks_removed,
|
||||||
bytes_reclaimed = r.bytes_reclaimed,
|
bytes_reclaimed = r.bytes_reclaimed,
|
||||||
max_gb = gb,
|
max_gb = gb,
|
||||||
|
pinned_blobs = pinned.len(),
|
||||||
"auto-GC evicted LRU blobs to hit size cap"
|
"auto-GC evicted LRU blobs to hit size cap"
|
||||||
),
|
),
|
||||||
Ok(_) => {} // under cap already; keep quiet
|
Ok(_) => {} // under cap already; keep quiet
|
||||||
|
|||||||
@@ -268,6 +268,67 @@ impl TagStore {
|
|||||||
Ok(entries)
|
Ok(entries)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Phase 4 (2026-07-13): union of every 32-byte value referenced
|
||||||
|
/// by a tag in either the legacy `tags/` or the Phase-3c
|
||||||
|
/// `tags-v2/` namespace. Feeds pin-aware eviction: any blob whose
|
||||||
|
/// id appears in this set is protected from LRU eviction.
|
||||||
|
///
|
||||||
|
/// Bounded by (tags on disk × 32 bytes); a fleet with 1000 tags
|
||||||
|
/// is well under 100 KiB of memory.
|
||||||
|
pub async fn pinned_blob_values(
|
||||||
|
&self,
|
||||||
|
) -> Result<std::collections::HashSet<[u8; 32]>> {
|
||||||
|
let mut out = std::collections::HashSet::new();
|
||||||
|
// Unstamped: same walk as `list`, but we skip TagEntry hex
|
||||||
|
// encoding and just push raw 32-byte values.
|
||||||
|
let tags_root = self.root.join("tags");
|
||||||
|
if tags_root.is_dir() {
|
||||||
|
let mut top = tokio::fs::read_dir(&tags_root).await?;
|
||||||
|
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 bytes = match tokio::fs::read(entry.path()).await {
|
||||||
|
Ok(b) => b,
|
||||||
|
Err(_) => continue,
|
||||||
|
};
|
||||||
|
if let Ok((_, value)) = decode_record(&bytes) {
|
||||||
|
out.insert(value);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// Stamped: walk tags-v2/, decode as StampedTag, insert value.
|
||||||
|
let stamped_root = self.root.join("tags-v2");
|
||||||
|
if stamped_root.is_dir() {
|
||||||
|
let mut top = tokio::fs::read_dir(&stamped_root).await?;
|
||||||
|
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 bytes = match tokio::fs::read(entry.path()).await {
|
||||||
|
Ok(b) => b,
|
||||||
|
Err(_) => continue,
|
||||||
|
};
|
||||||
|
if let Ok((_, stamped)) = decode_stamped_record(&bytes) {
|
||||||
|
out.insert(stamped.value);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Ok(out)
|
||||||
|
}
|
||||||
|
|
||||||
fn tag_path(&self, key: &str) -> PathBuf {
|
fn tag_path(&self, key: &str) -> PathBuf {
|
||||||
let hash_hex = hex32(blake3::hash(key.as_bytes()).as_bytes());
|
let hash_hex = hex32(blake3::hash(key.as_bytes()).as_bytes());
|
||||||
self.root
|
self.root
|
||||||
@@ -657,6 +718,43 @@ mod tests {
|
|||||||
assert_eq!(store.get_stamped(k).await.unwrap(), Some(higher));
|
assert_eq!(store.get_stamped(k).await.unwrap(), Some(higher));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn pinned_blob_values_unions_both_stores() {
|
||||||
|
// Phase 4 (2026-07-13): pin-set feed for LRU eviction. Any
|
||||||
|
// 32-byte value referenced by ANY tag (legacy or stamped)
|
||||||
|
// must show up in the union. Duplicates dedupe naturally.
|
||||||
|
let (_tmp, store) = open();
|
||||||
|
|
||||||
|
// Legacy tag pointing at value V1.
|
||||||
|
let v1 = [0xA1; 32];
|
||||||
|
store.put("clawverse:main:latest", &v1).await.unwrap();
|
||||||
|
|
||||||
|
// Stamped tag pointing at value V2.
|
||||||
|
let v2 = [0xB2; 32];
|
||||||
|
let s2 = StampedTagValue {
|
||||||
|
value: v2,
|
||||||
|
clock: 5,
|
||||||
|
node: [0; 8],
|
||||||
|
};
|
||||||
|
store
|
||||||
|
.put_stamped("clawverse:pr-42:latest", s2)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
// Same value V1 also pinned via a stamped tag → union dedupes.
|
||||||
|
let s3 = StampedTagValue {
|
||||||
|
value: v1,
|
||||||
|
clock: 6,
|
||||||
|
node: [0; 8],
|
||||||
|
};
|
||||||
|
store.put_stamped("mirror:main", s3).await.unwrap();
|
||||||
|
|
||||||
|
let pins = store.pinned_blob_values().await.unwrap();
|
||||||
|
assert!(pins.contains(&v1));
|
||||||
|
assert!(pins.contains(&v2));
|
||||||
|
assert_eq!(pins.len(), 2, "duplicate values dedupe");
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn stamped_and_unstamped_stores_are_independent() {
|
async fn stamped_and_unstamped_stores_are_independent() {
|
||||||
// put() writes to tags/, put_stamped() writes to tags-v2/.
|
// put() writes to tags/, put_stamped() writes to tags-v2/.
|
||||||
|
|||||||
+23
-4
@@ -246,7 +246,8 @@ async fn main() -> Result<()> {
|
|||||||
// ── cluster peer-status ───────────────────────────────────────────────────────
|
// ── cluster peer-status ───────────────────────────────────────────────────────
|
||||||
|
|
||||||
async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
|
async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
|
||||||
use cluster::blob::BlobStore;
|
use cluster::blob::{BlobId, BlobStore};
|
||||||
|
use cluster::tags::TagStore;
|
||||||
|
|
||||||
let root = cfg
|
let root = cfg
|
||||||
.cluster
|
.cluster
|
||||||
@@ -267,14 +268,31 @@ async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
|
|||||||
.await
|
.await
|
||||||
.context("gc_orphan_chunks failed")?;
|
.context("gc_orphan_chunks failed")?;
|
||||||
|
|
||||||
// Phase 2 (optional): LRU eviction to hit a size cap.
|
// Phase 4 (2026-07-13): gather pin set from tag store so evictions
|
||||||
|
// respect them. Cheap even at fleet scale (one 32-byte value per
|
||||||
|
// tag).
|
||||||
|
let tags_dir = root.join("tags-db");
|
||||||
|
let pinned_blobs = if tags_dir.is_dir() {
|
||||||
|
let ts = TagStore::open(tags_dir.clone())
|
||||||
|
.with_context(|| format!("opening tag store at {}", tags_dir.display()))?;
|
||||||
|
ts.pinned_blob_values()
|
||||||
|
.await
|
||||||
|
.context("collecting pinned tag values")?
|
||||||
|
.into_iter()
|
||||||
|
.map(BlobId::from_bytes)
|
||||||
|
.collect::<std::collections::HashSet<_>>()
|
||||||
|
} else {
|
||||||
|
std::collections::HashSet::new()
|
||||||
|
};
|
||||||
|
|
||||||
|
// Phase 2 (optional): pin-aware LRU eviction to hit a size cap.
|
||||||
let evict = if let Some(gb) = evict_to_gb {
|
let evict = if let Some(gb) = evict_to_gb {
|
||||||
let cap = gb.saturating_mul(1024 * 1024 * 1024);
|
let cap = gb.saturating_mul(1024 * 1024 * 1024);
|
||||||
Some(
|
Some(
|
||||||
store
|
store
|
||||||
.evict_to_size_cap(cap)
|
.evict_to_size_cap_with_pins(cap, &pinned_blobs)
|
||||||
.await
|
.await
|
||||||
.context("evict_to_size_cap failed")?,
|
.context("evict_to_size_cap_with_pins failed")?,
|
||||||
)
|
)
|
||||||
} else {
|
} else {
|
||||||
None
|
None
|
||||||
@@ -291,6 +309,7 @@ async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
|
|||||||
println!("lru eviction (cap {} GiB):", evict_to_gb.unwrap_or(0));
|
println!("lru eviction (cap {} GiB):", evict_to_gb.unwrap_or(0));
|
||||||
println!(" chunks removed: {}", evict.chunks_removed);
|
println!(" chunks removed: {}", evict.chunks_removed);
|
||||||
println!(" bytes reclaimed: {}", evict.bytes_reclaimed);
|
println!(" bytes reclaimed: {}", evict.bytes_reclaimed);
|
||||||
|
println!(" pinned blobs: {}", pinned_blobs.len());
|
||||||
}
|
}
|
||||||
println!("total elapsed: {:?}", elapsed);
|
println!("total elapsed: {:?}", elapsed);
|
||||||
println!("────────────────────────────────────────────────────");
|
println!("────────────────────────────────────────────────────");
|
||||||
|
|||||||
Reference in New Issue
Block a user