Phase 7d follow-on: snapshots pin blobs against LRU eviction #58

Merged
osobh merged 1 commits from phase-7d-snapshot-pins into main 2026-07-14 16:30:36 +00:00
3 changed files with 83 additions and 3 deletions
Showing only changes of commit eebc62d87b - Show all commits
+21 -1
View File
@@ -302,6 +302,14 @@ impl ClusterServices {
(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 {
@@ -339,7 +347,7 @@ impl ClusterServices {
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let pinned = match &tag_store_for_gc {
let mut pinned = match &tag_store_for_gc {
Some(ts) => {
// Phase 4b: prune expired pins first so
// the pin set reflects the wall-clock
@@ -357,6 +365,17 @@ impl ClusterServices {
}
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
@@ -366,6 +385,7 @@ impl ClusterServices {
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
+48
View File
@@ -183,6 +183,28 @@ impl SnapshotStore {
}
}
/// Phase 7d follow-on: union of every `blob_id` referenced by ANY
/// snapshot. Feeds pin-aware eviction — every blob captured by a
/// live snapshot survives the LRU cap. Semantically "snapshots
/// act as immortal retention pins until the operator deletes
/// them".
///
/// Cheap: one file read + JSON parse per snapshot. A fleet with
/// 100 snapshots × 10k blobs each is ~1 MB of JSON I/O.
pub async fn pinned_blob_ids(
&self,
) -> Result<std::collections::HashSet<BlobId>> {
let mut out = std::collections::HashSet::new();
for summary in self.list().await? {
if let Some(m) = self.get(&summary.name).await? {
for id in m.blob_ids {
out.insert(id);
}
}
}
Ok(out)
}
fn snapshot_path(&self, name: &str) -> PathBuf {
// Names are validated: no slashes, printable only. Safe to
// use directly as a filename fragment. We still normalize
@@ -347,6 +369,32 @@ mod tests {
assert!(validate_name("v2026.07.14-pre-release").is_ok());
}
#[tokio::test]
async fn pinned_blob_ids_unions_all_snapshots() {
// Two snapshots, some overlap. Union must dedupe.
let (_tmp, blob, snap) = open();
let a = blob.put_bytes(b"pin-alpha").await.unwrap();
let b = blob.put_bytes(b"pin-beta").await.unwrap();
let c = blob.put_bytes(b"pin-gamma").await.unwrap();
// Snapshot 1: captures a, b, c
snap.create("s1", &blob, 1).await.unwrap();
// Snapshot 2 (later): same content, still captures a, b, c
snap.create("s2", &blob, 2).await.unwrap();
let pins = snap.pinned_blob_ids().await.unwrap();
assert_eq!(pins.len(), 3);
assert!(pins.contains(&a));
assert!(pins.contains(&b));
assert!(pins.contains(&c));
}
#[tokio::test]
async fn pinned_blob_ids_empty_when_no_snapshots() {
let (_tmp, blob, snap) = open();
blob.put_bytes(b"blob-with-no-snapshot").await.unwrap();
assert!(snap.pinned_blob_ids().await.unwrap().is_empty());
}
#[tokio::test]
async fn round_trip_preserves_blob_ids_sorted() {
// list_blob_ids order is filesystem-dependent. Snapshot
+14 -2
View File
@@ -341,7 +341,7 @@ async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let (pinned_blobs, expired_pruned) = if tags_dir.is_dir() {
let (mut pinned_blobs, expired_pruned) = if tags_dir.is_dir() {
let ts = TagStore::open(tags_dir.clone())
.with_context(|| format!("opening tag store at {}", tags_dir.display()))?;
// Phase 4b: prune first so the set we hand to the evictor is
@@ -361,6 +361,18 @@ async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
} else {
(std::collections::HashSet::new(), 0)
};
// Phase 7d follow-on: snapshot references act as immortal pins.
// Any blob captured by ANY snapshot survives the LRU cap so
// operators can guarantee retention windows via snapshots alone,
// without hand-managing per-blob tags.
let snapshot_store = cluster::snapshot::SnapshotStore::open(root.clone())
.context("opening snapshot store")?;
let snapshot_pins = snapshot_store
.pinned_blob_ids()
.await
.context("collecting snapshot pins")?;
let snapshot_pin_count = snapshot_pins.len();
pinned_blobs.extend(snapshot_pins);
// Phase 2 (optional): pin-aware LRU eviction to hit a size cap.
let evict = if let Some(gb) = evict_to_gb {
@@ -386,7 +398,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!(" chunks removed: {}", evict.chunks_removed);
println!(" bytes reclaimed: {}", evict.bytes_reclaimed);
println!(" pinned blobs: {}", pinned_blobs.len());
println!(" pinned blobs: {} ({} from snapshots)", pinned_blobs.len(), snapshot_pin_count);
println!(" expired pins pruned: {}", expired_pruned);
}
println!("total elapsed: {:?}", elapsed);