Phase 4b: TagStore expiry primitives (pin TTL groundwork) #45
@@ -335,9 +335,22 @@ impl ClusterServices {
|
||||
// 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 pinned = match &tag_store_for_gc {
|
||||
Some(ts) => {
|
||||
let raw = ts.pinned_blob_values().await.unwrap_or_default();
|
||||
// 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()
|
||||
|
||||
@@ -277,6 +277,20 @@ impl TagStore {
|
||||
/// is well under 100 KiB of memory.
|
||||
pub async fn pinned_blob_values(
|
||||
&self,
|
||||
) -> Result<std::collections::HashSet<[u8; 32]>> {
|
||||
// No `now` filter → treat as "no expiry gate" by passing u64::MAX,
|
||||
// so any tag whose expiry ≤ u64::MAX (i.e. any) still counts. All
|
||||
// expiry-aware callers should use `pinned_blob_values_at`.
|
||||
self.pinned_blob_values_at(u64::MAX).await
|
||||
}
|
||||
|
||||
/// Phase 4b (2026-07-13): expiry-aware pin gathering. A stamped tag
|
||||
/// is treated as pinning its value only if either (a) it has no
|
||||
/// sidecar expiry, or (b) `expires_at > now`. Legacy `tags/` entries
|
||||
/// have no expiry surface and always count.
|
||||
pub async fn pinned_blob_values_at(
|
||||
&self,
|
||||
now_unix: u64,
|
||||
) -> Result<std::collections::HashSet<[u8; 32]>> {
|
||||
let mut out = std::collections::HashSet::new();
|
||||
// Unstamped: same walk as `list`, but we skip TagEntry hex
|
||||
@@ -321,7 +335,23 @@ impl TagStore {
|
||||
Err(_) => continue,
|
||||
};
|
||||
if let Ok((_, stamped)) = decode_stamped_record(&bytes) {
|
||||
out.insert(stamped.value);
|
||||
// Check for `<path>.exp` sidecar. Absent → no
|
||||
// expiry. Present + not-yet-expired → still counts.
|
||||
// Present + expired → drop.
|
||||
let mut exp_path = entry.path();
|
||||
exp_path.set_extension("svtag.exp");
|
||||
let keep = match tokio::fs::read(&exp_path).await {
|
||||
Ok(bytes) if bytes.len() == 8 => {
|
||||
let expires_at = u64::from_le_bytes(
|
||||
bytes.as_slice().try_into().unwrap(),
|
||||
);
|
||||
expires_at > now_unix
|
||||
}
|
||||
_ => true,
|
||||
};
|
||||
if keep {
|
||||
out.insert(stamped.value);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -329,6 +359,95 @@ impl TagStore {
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// Phase 4b: attach an expiry to a stamped tag. `expires_at_unix`
|
||||
/// is absolute wall-clock seconds; `0` means "never expire" and
|
||||
/// clears any existing sidecar. Writing to a key that has no
|
||||
/// stamped tag on disk is not an error — the sidecar is written
|
||||
/// and will start filtering once the tag lands.
|
||||
pub async fn set_stamped_expiry(&self, key: &str, expires_at_unix: u64) -> Result<()> {
|
||||
validate_key(key)?;
|
||||
let path = self.stamped_expiry_path(key);
|
||||
if expires_at_unix == 0 {
|
||||
match tokio::fs::remove_file(&path).await {
|
||||
Ok(()) => {}
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
|
||||
Err(e) => return Err(anyhow::Error::from(e)),
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
if let Some(parent) = path.parent() {
|
||||
tokio::fs::create_dir_all(parent).await.with_context(|| {
|
||||
format!("creating stamped-tag bucket {}", parent.display())
|
||||
})?;
|
||||
}
|
||||
self.atomic_write(&path, &expires_at_unix.to_le_bytes()).await
|
||||
}
|
||||
|
||||
/// Phase 4b: read the expiry sidecar for a stamped tag, if any.
|
||||
pub async fn get_stamped_expiry(&self, key: &str) -> Result<Option<u64>> {
|
||||
validate_key(key)?;
|
||||
let path = self.stamped_expiry_path(key);
|
||||
match tokio::fs::read(&path).await {
|
||||
Ok(bytes) if bytes.len() == 8 => {
|
||||
Ok(Some(u64::from_le_bytes(bytes.as_slice().try_into().unwrap())))
|
||||
}
|
||||
Ok(_) => bail!("stamped-tag expiry sidecar at {} has wrong length", path.display()),
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
|
||||
Err(e) => Err(anyhow::Error::from(e)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Phase 4b: delete expired stamped tags and their sidecars.
|
||||
/// Returns count removed. A tag is expired iff its `.svtag.exp`
|
||||
/// sidecar contains an `expires_at ≤ now_unix`. Legacy `tags/`
|
||||
/// entries are never touched here — they have no expiry surface.
|
||||
pub async fn prune_expired_stamped_at(&self, now_unix: u64) -> Result<usize> {
|
||||
let mut removed = 0usize;
|
||||
let stamped_root = self.root.join("tags-v2");
|
||||
if !stamped_root.is_dir() {
|
||||
return Ok(0);
|
||||
}
|
||||
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 tag_path = entry.path();
|
||||
if tag_path.extension().and_then(|s| s.to_str()) != Some("svtag") {
|
||||
continue;
|
||||
}
|
||||
let mut exp_path = tag_path.clone();
|
||||
exp_path.set_extension("svtag.exp");
|
||||
let expires_at = match tokio::fs::read(&exp_path).await {
|
||||
Ok(bytes) if bytes.len() == 8 => {
|
||||
u64::from_le_bytes(bytes.as_slice().try_into().unwrap())
|
||||
}
|
||||
_ => continue,
|
||||
};
|
||||
if expires_at > now_unix {
|
||||
continue;
|
||||
}
|
||||
let _ = tokio::fs::remove_file(&tag_path).await;
|
||||
let _ = tokio::fs::remove_file(&exp_path).await;
|
||||
removed += 1;
|
||||
}
|
||||
}
|
||||
Ok(removed)
|
||||
}
|
||||
|
||||
fn stamped_expiry_path(&self, key: &str) -> PathBuf {
|
||||
let hash_hex = hex32(blake3::hash(key.as_bytes()).as_bytes());
|
||||
self.root
|
||||
.join("tags-v2")
|
||||
.join(&hash_hex[..2])
|
||||
.join(format!("{hash_hex}.svtag.exp"))
|
||||
}
|
||||
|
||||
fn tag_path(&self, key: &str) -> PathBuf {
|
||||
let hash_hex = hex32(blake3::hash(key.as_bytes()).as_bytes());
|
||||
self.root
|
||||
@@ -755,6 +874,52 @@ mod tests {
|
||||
assert_eq!(pins.len(), 2, "duplicate values dedupe");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn expiry_gates_pin_set_and_prune_removes_expired() {
|
||||
// Phase 4b: sidecar `.exp` files scope the pin protection
|
||||
// window. A stamped tag with expires_at ≤ now is neither
|
||||
// reported as a pinned value nor should it survive
|
||||
// `prune_expired_stamped_at(now)`.
|
||||
let (_tmp, store) = open();
|
||||
|
||||
let live = StampedTagValue { value: [0x01; 32], clock: 1, node: [0; 8] };
|
||||
let expired = StampedTagValue { value: [0x02; 32], clock: 1, node: [0; 8] };
|
||||
let no_ttl = StampedTagValue { value: [0x03; 32], clock: 1, node: [0; 8] };
|
||||
store.put_stamped("live", live).await.unwrap();
|
||||
store.put_stamped("expired", expired).await.unwrap();
|
||||
store.put_stamped("no-ttl", no_ttl).await.unwrap();
|
||||
|
||||
// now = 1000. live expires at 2000, expired at 500.
|
||||
store.set_stamped_expiry("live", 2000).await.unwrap();
|
||||
store.set_stamped_expiry("expired", 500).await.unwrap();
|
||||
// no-ttl deliberately has no sidecar.
|
||||
|
||||
// Round-trip the sidecar.
|
||||
assert_eq!(store.get_stamped_expiry("live").await.unwrap(), Some(2000));
|
||||
assert_eq!(store.get_stamped_expiry("no-ttl").await.unwrap(), None);
|
||||
|
||||
// Pin set at now=1000: only live + no-ttl.
|
||||
let pins = store.pinned_blob_values_at(1000).await.unwrap();
|
||||
assert!(pins.contains(&[0x01; 32]));
|
||||
assert!(pins.contains(&[0x03; 32]));
|
||||
assert!(!pins.contains(&[0x02; 32]), "expired pin must not count");
|
||||
assert_eq!(pins.len(), 2);
|
||||
|
||||
// Prune at now=1000: only "expired" goes.
|
||||
let removed = store.prune_expired_stamped_at(1000).await.unwrap();
|
||||
assert_eq!(removed, 1);
|
||||
assert_eq!(store.get_stamped("expired").await.unwrap(), None);
|
||||
assert!(store.get_stamped("live").await.unwrap().is_some());
|
||||
assert!(store.get_stamped("no-ttl").await.unwrap().is_some());
|
||||
|
||||
// Clear via expires_at=0 removes the sidecar.
|
||||
store.set_stamped_expiry("live", 0).await.unwrap();
|
||||
assert_eq!(store.get_stamped_expiry("live").await.unwrap(), None);
|
||||
// Now "live" pin has no expiry gate → always counts.
|
||||
let pins_after = store.pinned_blob_values_at(0).await.unwrap();
|
||||
assert!(pins_after.contains(&[0x01; 32]));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stamped_and_unstamped_stores_are_independent() {
|
||||
// put() writes to tags/, put_stamped() writes to tags-v2/.
|
||||
|
||||
+17
-4
@@ -272,17 +272,29 @@ async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
|
||||
// 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 now_unix = std::time::SystemTime::now()
|
||||
.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 ts = TagStore::open(tags_dir.clone())
|
||||
.with_context(|| format!("opening tag store at {}", tags_dir.display()))?;
|
||||
ts.pinned_blob_values()
|
||||
// Phase 4b: prune first so the set we hand to the evictor is
|
||||
// current as of `now_unix`.
|
||||
let pruned = ts
|
||||
.prune_expired_stamped_at(now_unix)
|
||||
.await
|
||||
.context("pruning expired pins")?;
|
||||
let pins = ts
|
||||
.pinned_blob_values_at(now_unix)
|
||||
.await
|
||||
.context("collecting pinned tag values")?
|
||||
.into_iter()
|
||||
.map(BlobId::from_bytes)
|
||||
.collect::<std::collections::HashSet<_>>()
|
||||
.collect::<std::collections::HashSet<_>>();
|
||||
(pins, pruned)
|
||||
} else {
|
||||
std::collections::HashSet::new()
|
||||
(std::collections::HashSet::new(), 0)
|
||||
};
|
||||
|
||||
// Phase 2 (optional): pin-aware LRU eviction to hit a size cap.
|
||||
@@ -310,6 +322,7 @@ async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
|
||||
println!(" chunks removed: {}", evict.chunks_removed);
|
||||
println!(" bytes reclaimed: {}", evict.bytes_reclaimed);
|
||||
println!(" pinned blobs: {}", pinned_blobs.len());
|
||||
println!(" expired pins pruned: {}", expired_pruned);
|
||||
}
|
||||
println!("total elapsed: {:?}", elapsed);
|
||||
println!("────────────────────────────────────────────────────");
|
||||
|
||||
Reference in New Issue
Block a user