use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use anyhow::Result;
use tracing::{info, warn};
use crate::reconciler::static_asset::WORKLOAD_KIND;
use crate::ServiceConfig;
#[derive(Debug, Default)]
pub struct DeriveCacheLiveHashes {
pub fetch: BTreeSet<String>,
pub transform: BTreeSet<String>,
}
#[derive(Debug, Clone)]
pub struct DerivePruneCandidate {
pub path: PathBuf,
pub cache_kind: &'static str,
pub hash: String,
pub size: u64,
}
pub fn collect_live_derive_hashes<'a>(
workspace_root: &Path,
services: impl IntoIterator<Item = &'a ServiceConfig>,
) -> Result<DeriveCacheLiveHashes> {
let mut live = DeriveCacheLiveHashes::default();
for svc in services {
for component in &svc.components {
if component.kind != WORKLOAD_KIND {
continue;
}
let workload_path = workspace_root.join(&component.path).join("workload.toml");
let src = match std::fs::read_to_string(&workload_path) {
Ok(s) => s,
Err(e) => {
warn!(
path = %workload_path.display(),
error = %e,
"derive-cache prune: skipping unreadable workload",
);
continue;
}
};
let envelope: workload_spec::Workload = match toml::from_str(&src) {
Ok(w) => w,
Err(e) => {
warn!(
path = %workload_path.display(),
error = %e,
"derive-cache prune: skipping unparseable workload",
);
continue;
}
};
let workload_spec::Workload::StaticAsset(workload) = envelope else {
continue;
};
for entry in &workload.assets {
let Some(derive) = &entry.derive else {
continue;
};
live.fetch.insert(derive.fetch.blake3.0.clone());
if derive.transform.is_some() {
live.transform.insert(entry.blake3.0.clone());
}
}
}
}
Ok(live)
}
pub fn compute_derive_cache_candidates(
derive_cache_root: &Path,
live: &DeriveCacheLiveHashes,
) -> Result<Vec<DerivePruneCandidate>> {
let mut candidates = Vec::new();
for (subdir, live_set) in [("fetch", &live.fetch), ("transform", &live.transform)] {
let dir = derive_cache_root.join(subdir);
let entries = match std::fs::read_dir(&dir) {
Ok(e) => e,
Err(_) => continue, };
for entry in entries.flatten() {
let path = entry.path();
let ext = path.extension().and_then(|e| e.to_str());
if ext != Some("bin") {
continue; }
let Some(hash) = path
.file_stem()
.and_then(|s| s.to_str())
.map(|s| s.to_string())
else {
continue;
};
if live_set.contains(hash.as_str()) {
continue;
}
let size = entry.metadata().map(|m| m.len()).unwrap_or(0);
candidates.push(DerivePruneCandidate {
path,
cache_kind: if subdir == "fetch" {
"fetch"
} else {
"transform"
},
hash,
size,
});
}
}
candidates.sort_by(|a, b| a.path.cmp(&b.path));
Ok(candidates)
}
pub fn execute_derive_cache_prune(candidates: &[DerivePruneCandidate]) -> (usize, u64, usize) {
let mut deleted = 0usize;
let mut bytes = 0u64;
let mut errors = 0usize;
for c in candidates {
match std::fs::remove_file(&c.path) {
Ok(()) => {
info!(path = %c.path.display(), "pruned derive cache entry");
deleted += 1;
bytes += c.size;
}
Err(e) => {
warn!(path = %c.path.display(), error = %e, "failed to prune derive cache entry");
errors += 1;
}
}
}
(deleted, bytes, errors)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::{ServiceComponent, ServiceConfig};
use tempfile::TempDir;
fn svc(name: &str, components: Vec<ServiceComponent>) -> ServiceConfig {
ServiceConfig {
schema_version: 1,
name: name.into(),
address: crate::config::ServiceAddress::front_door(String::new()),
description: None,
components,
db: crate::DbCatalog::default(),
}
}
fn static_asset_component(id: &str, path: &str) -> ServiceComponent {
ServiceComponent {
mount: None,
id: id.into(),
kind: "static-asset".into(),
path: path.into(),
role: String::new(),
publishes: None,
wave: 0,
git: None,
deploy: Default::default(),
}
}
fn write_workload(dir: &Path, toml: &str) {
if let Some(p) = dir.parent() {
std::fs::create_dir_all(p).unwrap();
}
std::fs::create_dir_all(dir).unwrap();
std::fs::write(dir.join("workload.toml"), toml).unwrap();
}
const FETCH_HASH: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
const OUTPUT_HASH: &str = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
fn workload_with_fetch_only(fetch_hash: &str, output_hash: &str) -> String {
format!(
r#"kind = "static-asset"
schema_version = "V1"
[[asset]]
filename = "model.bin"
blake3 = "{output_hash}"
[asset.derive.fetch]
url = "https://example.com/model.bin"
blake3 = "{fetch_hash}"
license = "mit"
"#
)
}
fn workload_with_fetch_and_transform(fetch_hash: &str, output_hash: &str) -> String {
format!(
r#"kind = "static-asset"
schema_version = "V1"
[[asset]]
filename = "model.bin"
blake3 = "{output_hash}"
[asset.derive.fetch]
url = "https://example.com/model.bin"
blake3 = "{fetch_hash}"
license = "mit"
[asset.derive.transform]
recipe = "quantize"
"#
)
}
#[test]
fn collect_live_hashes_fetch_only() {
let tmp = TempDir::new().unwrap();
let workload_dir = tmp.path().join("app/web");
write_workload(
&workload_dir,
&workload_with_fetch_only(FETCH_HASH, OUTPUT_HASH),
);
let svc = svc("my-svc", vec![static_asset_component("site", "app/web")]);
let live = collect_live_derive_hashes(tmp.path(), [&svc]).unwrap();
assert!(live.fetch.contains(FETCH_HASH), "fetch hash must be live");
assert!(
live.transform.is_empty(),
"fetch-only asset has no transform cache entry"
);
}
#[test]
fn collect_live_hashes_fetch_and_transform() {
let tmp = TempDir::new().unwrap();
let workload_dir = tmp.path().join("app/web");
write_workload(
&workload_dir,
&workload_with_fetch_and_transform(FETCH_HASH, OUTPUT_HASH),
);
let svc = svc("my-svc", vec![static_asset_component("site", "app/web")]);
let live = collect_live_derive_hashes(tmp.path(), [&svc]).unwrap();
assert!(live.fetch.contains(FETCH_HASH));
assert!(
live.transform.contains(OUTPUT_HASH),
"transform cache entry keyed by output blake3"
);
}
#[test]
fn collect_live_hashes_missing_workload_skips_silently() {
let tmp = TempDir::new().unwrap();
let svc = svc("my-svc", vec![static_asset_component("site", "app/web")]);
let live = collect_live_derive_hashes(tmp.path(), [&svc]).unwrap();
assert!(live.fetch.is_empty());
assert!(live.transform.is_empty());
}
#[test]
fn collect_live_hashes_ignores_non_static_asset_components() {
let tmp = TempDir::new().unwrap();
let svc = svc(
"my-svc",
vec![ServiceComponent {
mount: None,
id: "api".into(),
kind: "container".into(),
path: "app/api".into(),
role: String::new(),
publishes: None,
wave: 0,
git: None,
deploy: Default::default(),
}],
);
let live = collect_live_derive_hashes(tmp.path(), [&svc]).unwrap();
assert!(live.fetch.is_empty());
}
#[test]
fn candidates_exclude_live_hashes() {
let tmp = TempDir::new().unwrap();
let fetch_dir = tmp.path().join("fetch");
std::fs::create_dir_all(&fetch_dir).unwrap();
std::fs::write(fetch_dir.join(format!("{FETCH_HASH}.bin")), b"live").unwrap();
let orphan = "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc";
std::fs::write(fetch_dir.join(format!("{orphan}.bin")), b"orphan").unwrap();
std::fs::write(fetch_dir.join(format!("{orphan}.partial")), b"partial").unwrap();
let mut live = DeriveCacheLiveHashes::default();
live.fetch.insert(FETCH_HASH.to_string());
let candidates = compute_derive_cache_candidates(tmp.path(), &live).unwrap();
assert_eq!(candidates.len(), 1, "exactly one orphan");
assert_eq!(candidates[0].hash, orphan);
assert_eq!(candidates[0].cache_kind, "fetch");
}
#[test]
fn execute_prune_deletes_candidates() {
let tmp = TempDir::new().unwrap();
let fetch_dir = tmp.path().join("fetch");
std::fs::create_dir_all(&fetch_dir).unwrap();
let file = fetch_dir.join("dead.bin");
std::fs::write(&file, b"orphan bytes").unwrap();
assert!(file.exists());
let candidate = DerivePruneCandidate {
path: file.clone(),
cache_kind: "fetch",
hash: "dead".into(),
size: 12,
};
let (deleted, bytes, errors) = execute_derive_cache_prune(&[candidate]);
assert_eq!(deleted, 1);
assert_eq!(bytes, 12);
assert_eq!(errors, 0);
assert!(!file.exists(), "file must be deleted");
}
}