a burst of writes shares one gc, and gc reads each generation once
Signed-off-by: dawn <90008@klbr.net>
1279 unmodified linesExpand all1280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151279 unmodified linesExpand all ); after(&first_id); assert!(!abandoned.exists(), "the current one collects");}
#[test]fn a_write_finding_a_gc_queued_leaves_the_collecting_to_it() { let env = setup_test_env(); assert!(knot_static(&env.config, ["generate"]).status.success()); let first = get_only_generation_dir(&env.output); let first_id = first.file_name().unwrap().to_str().unwrap().to_string(); assert!( knot_static(&env.config, ["activate", &first_id]) .status .success() ); let abandoned = env.output.join("generations").join(".building-abandoned"); std::fs::create_dir_all(&abandoned).unwrap(); let after = || { let ran = knot_static(&env.config, ["after-activation", &first_id, "--at", "1"]); assert!( ran.status.success(), "{}", String::from_utf8_lossy(&ran.stderr) ); };
std::fs::create_dir_all(env.output.join("runtime")).unwrap(); let queued = std::fs::File::create(env.output.join("runtime/gc.queued")).unwrap(); queued.lock().unwrap(); after(); assert!(abandoned.exists(), "the queued gc collects, not this one"); drop(queued); after(); assert!(!abandoned.exists());}454 unmodified linesExpand all45545645745845946046146246346446546646246346446746846946647047147247347410 unmodified lines48548648748848949048548648749149249349449549649749850 unmodified lines454 unmodified linesExpand all || publish::current_generation(&config.output_root).as_deref() != Some(generation); if superseded() { return Ok(()); } // a burst of writes shares one gc: the one waiting collects for everyone activated before it let gc_slot = crate::gc::queue(&config.output_root)?; if gc_slot.is_none() && config.activate.site_hook.is_none() { return Ok(()); } let generations = config.output_root.join("generations"); let hook = { let _output_lock = crate::output::lock(&config.output_root)?; if superseded() { return Ok(()); } let collect = gc_slot.is_some(); drop(gc_slot); let hook = match &config.activate.site_hook { Some(_) => { Some(_) if !superseded() => { let generation_dir = generations.join(generation); let manifest = GenerationManifest::load(&generation_dir)?; let previous = previous.and_then(|previous| { GenerationManifest::load(&generations.join(previous)).ok()10 unmodified lines .ok() } _ => None, }; if collect { let removed = let removed = crate::gc::collect(&config.output_root, config.activate.keep_generations)?; if !removed.is_empty() { eprintln!("gc: {removed}"); crate::gc::collect(&config.output_root, config.activate.keep_generations)?; if !removed.is_empty() { eprintln!("gc: {removed}"); } } hook }; if let (Some(hook), Some(path)) = (&config.activate.site_hook, hook) {50 unmodified lines1 unmodified line234566789105 unmodified lines161718192021222324252616 unmodified lines434445464748495051525354555657585 unmodified lines646566675768697071721 unmodified line747576776778798081828 unmodified lines919293948495969798992 unmodified lines1021031041059510610710810911011111210210310411311411511611711811912012112212312412512612712812913013113210710810911011111211311411511611711811912012112212312412512613313413513614 unmodified lines15115215315415515615715815916016116216316416516616716816917017117217317417517617717868 unmodified lines24724824925022525125225325425525683 unmodified lines340341342343317344345346347348349350324351352353354355356357358359360361362363364365366367368148 unmodified linesExpand all1 unmodified lineuse std::path::{Path, PathBuf};use std::sync::Arc;
use anyhow::Context as _;use serde::Deserialize;use serde::{Deserialize, Serialize};
use crate::config::ValidatedConfig;use crate::datastore::Datastore;use crate::datastore::keys::RepoKey;5 unmodified linespub(crate) const DRAIN_SECS: u64 = crate::nginx::WORKER_SHUTDOWN_SECS + 60;
const STORE_DIRS: [&str; 2] = ["releases", "cache/export"];const PINS_DIR: &str = "pins";const QUEUE_FILE: &str = "runtime/gc.queued";/// what gc reads of each generation's manifest, kept by generation id since a generation never changesconst LIVE_CACHE: &str = "cache/gc-live";
/// a build holds it until its activation, so another activation's gc leaves the candidate alone#[derive(Debug)]pub struct Pin {16 unmodified lines .with_context(|| format!("pin {}", path.display()))?; Ok(Pin { _file: file })}
/// at most one gc waits for the output lock; a write that finds the slot taken leaves its gc to/// that one, which collects against whatever is active once it has the lockpub(crate) fn queue(output_root: &Path) -> anyhow::Result<Option<std::fs::File>> { std::fs::create_dir_all(output_root.join("runtime")).context("create runtime dir")?; let file = crate::output::lock_file(output_root, QUEUE_FILE)?; Ok(file.try_lock().is_ok().then_some(file))}
/// a pin nobody holds is left over from a build that ended, and goesfn pinned(output_root: &Path, generation: &str) -> bool { let path = output_root.join(PINS_DIR).join(generation); let Ok(file) = std::fs::File::open(&path) else {5 unmodified lines let _ = std::fs::remove_file(&path); false}
#[derive(Deserialize)]#[derive(Deserialize, Serialize)]struct Live { #[serde(default)] repos: Vec<LiveRepo>, #[serde(default)]1 unmodified line #[serde(default)] profile: LiveProfile,}
#[derive(Deserialize, Default)]#[derive(Deserialize, Serialize, Default)]struct LiveProfile { #[serde(default)] atproto: BTreeMap<String, String>, #[serde(default)]8 unmodified lines seals: BTreeSet<String>, git_seals: BTreeSet<String>,}
#[derive(Deserialize)]#[derive(Deserialize, Serialize, PartialEq, Eq, PartialOrd, Ord)]struct LiveRepo { repo_did: String, #[serde(default)] release_id: Option<String>,2 unmodified lines #[serde(default)] object_format: Option<String>,}
#[derive(Deserialize, Default)]#[derive(Deserialize, Serialize, Default)]struct LiveStats { #[serde(default)] created_at_unix: u64,}
impl Live { fn load(generation_dir: &Path) -> Option<Self> { let bytes = std::fs::read(generation_dir.join("manifest.json")).ok()?; serde_json::from_slice(&bytes).ok() fn load(output_root: &Path, generation: &str) -> Option<Self> { let cached = output_root.join(LIVE_CACHE).join(generation); if let Ok(bytes) = std::fs::read(&cached) && let Ok(live) = serde_json::from_slice(&bytes) { return Some(live); } let manifest = output_root .join("generations") .join(generation) .join("manifest.json"); let live: Self = serde_json::from_slice(&std::fs::read(manifest).ok()?).ok()?; if let Ok(bytes) = serde_json::to_vec(&live) && std::fs::create_dir_all(output_root.join(LIVE_CACHE)).is_ok() { let _ = crate::cas::atomic_write(&cached, &bytes); } Some(live) }
fn entries(&self, output_root: &Path) -> impl Iterator<Item = PathBuf> { self.repos.iter().flat_map(move |repo| { let dir = crate::repolock::repo_dir(output_root, &repo_key(&repo.repo_did)); let release = repo .release_id .as_ref() .map(|id| dir.join("releases").join(id)); let store = RepoStore::open(&dir); let format = (repo.object_format.as_deref()) .and_then(|format| knot_types::ObjectFormat::from_capability(format)); let export = repo.git_export_hash.iter().flat_map(move |fingerprint| { let entry = store.entry_path(crate::git_export::EXPORT_CACHE_KIND, fingerprint); let filed = format .map(|format| crate::partitions::filed_ref_deltas(&store, &entry, format)); std::iter::once(entry).chain(filed.into_iter().flatten()) }); release.into_iter().chain(export) }) }
fn served(&self, into: &mut BTreeMap<String, Served>) { for (did, tip) in &self.profile.atproto { into.entry(repo_key(did)) .or_default()14 unmodified lines } }}
impl LiveRepo { fn entries(&self, output_root: &Path) -> Vec<PathBuf> { let dir = crate::repolock::repo_dir(output_root, &repo_key(&self.repo_did)); let release = self .release_id .as_ref() .map(|id| dir.join("releases").join(id)); let store = RepoStore::open(&dir); let format = (self.object_format.as_deref()) .and_then(|format| knot_types::ObjectFormat::from_capability(format)); let export = self.git_export_hash.iter().flat_map(|fingerprint| { let entry = store.entry_path(crate::git_export::EXPORT_CACHE_KIND, fingerprint); let filed = format.map(|format| crate::partitions::filed_ref_deltas(&store, &entry, format)); std::iter::once(entry).chain(filed.into_iter().flatten()) }); release.into_iter().chain(export).collect() }}
#[derive(Debug, Default, PartialEq, Eq)]pub(crate) struct Removed { pub generations: usize, pub store_entries: usize,68 unmodified lines if name.starts_with('.') { remove(&entry.path())?; removed.staging += 1; } else if entry.file_type()?.is_dir() { generations.push((name, Live::load(&entry.path()))); let live = Live::load(output_root, &name); generations.push((name, live)); } }
let Some(current) = crate::publish::current_generation(output_root) else {83 unmodified lines crate::publish::write_backend_record(output_root, &record)?; }
// an unreadable kept manifest could name any entry, and nothing is swept. let mut live = BTreeSet::new(); let mut repos_live = BTreeSet::new(); let mut served = BTreeMap::new(); for (name, view) in &generations { if kept.contains(name) { let Some(view) = view else { return Ok(removed); }; live.extend(view.entries(output_root)); repos_live.extend(&view.repos); view.served(&mut served); } } // most kept generations carry the same entries, so each is listed once let live: BTreeSet<PathBuf> = repos_live .into_iter() .flat_map(|repo| repo.entries(output_root)) .collect(); sweep(&output_root.join(LIVE_CACHE), |path| { path.file_name() .and_then(|name| name.to_str()) .is_some_and(|name| kept.contains(name)) })?; let chain_root = output_root.join("atproto"); let Ok(repos) = std::fs::read_dir(output_root.join("repos")) else { return Ok(removed); };148 unmodified linesExpand all409 unmodified linesExpand all410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440409 unmodified linesExpand all collect_at(root, 1, NOW, DRAIN_SECS).unwrap(); } assert!(!root.join(PINS_DIR).join("g1").exists());}
#[test]fn a_generation_is_read_once_and_its_cache_goes_with_it() { let root = knot(&["g1", "g2", "g3"]); let root = root.path(); activate(root, "g3", 3, &[("g1", 2), ("g2", 3)]); collect_at(root, 1, NOW, DRAIN_SECS).unwrap(); assert_eq!(generations(root), ["g2", "g3"]);
std::fs::write(root.join("generations/g2/manifest.json"), b"{ not json").unwrap(); let orphan = release(root, "r-orphan"); std::fs::create_dir_all(&orphan).unwrap(); collect_at(root, 1, NOW, DRAIN_SECS).unwrap(); assert!(!orphan.exists(), "g2 is read from what gc kept of it"); assert!(release(root, "r-g2").is_dir());
generation(root, "g4", 4); activate(root, "g4", 4, &[("g2", 3), ("g3", 4)]); collect_at(root, 1, NOW, DRAIN_SECS).unwrap(); assert_eq!(generations(root), ["g3", "g4"]); let mut cached: Vec<_> = std::fs::read_dir(root.join(LIVE_CACHE)) .unwrap() .map(|entry| entry.unwrap().file_name().into_string().unwrap()) .collect(); cached.sort(); assert_eq!(cached, ["g3", "g4"]);}