a write's candidate survives another activation's gc until it activates
Signed-off-by: dawn <90008@klbr.net>
1001 unmodified linesExpand all1002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881001 unmodified linesExpand all before, "a second sync builds nothing" );}
#[test]fn a_write_waiting_to_activate_keeps_its_candidate_through_another_writes_gc() { let env = setup_test_env(); let mut config = std::fs::read_to_string(&env.config).unwrap(); config.push_str("keep_generations = 1\n"); std::fs::write(&env.config, config).unwrap(); let beta = env._root.path().join("source-beta"); create_test_repo(&beta, "beta"); let _ = knot_static( &env.config, [ "repo", "import", "--owner", OWNER, "--rkey", "beta", "--name", "beta", "--default-branch", "main", beta.to_str().unwrap(), ], ); let describe = |repo: &str, text: &str| { knot_static( &env.config, ["repo", "settings", repo, "--description", text], ) }; for text in ["one", "two"] { assert!(describe("sample", text).status.success()); }
let handshake = env._root.path().join("paused"); let first = common::knot_static_cmd( &env.config, MASTER_KEY_ENV, &MASTER_KEY, ["repo", "settings", "sample", "--description", "first"], ) .env("KNOT_FAILPOINT", "before_activation") .env("KNOT_FAILPOINT_FILE", &handshake) .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()) .spawn() .unwrap(); let deadline = std::time::Instant::now() + std::time::Duration::from_secs(60); while !handshake.exists() { assert!( std::time::Instant::now() < deadline, "the first write never built" ); std::thread::sleep(std::time::Duration::from_millis(20)); } // the gc keeps builds made in the second of the current activation std::thread::sleep(std::time::Duration::from_millis(1100)); let second = describe("beta", "second"); assert!( second.status.success(), "{}", String::from_utf8_lossy(&second.stderr) ); std::fs::remove_file(&handshake).unwrap(); let first = first.wait_with_output().unwrap(); assert!( first.status.success(), "{}", String::from_utf8_lossy(&first.stderr) ); let index = std::fs::read_to_string( env.output .join("generations") .join(active_generation(&env.output)) .join("www/index.html"), ) .unwrap(); assert!( index.contains("first") && index.contains("second"), "{index}" );}276 unmodified linesExpand all277278279280281282283284285110 unmodified linesExpand all276 unmodified linesExpand all ("marks.rs", ".write(true)", 1), // lock files, opened writable only for flock. ("output.rs", ".write(true)", 1), ("repolock.rs", ".write(true)", 1), ("gc.rs", ".write(true)", 1), ("mod.rs", ".write(true)", 1), // copied because the export can't hardlink the source pack. ("reach.rs", "let target = std::fs::File::create(&to)?;", 1), (110 unmodified linesExpand all6 unmodified lines78910111213141516171819202122232425262728293031323334353637381516171819394041424319 unmodified lines6 unmodified linesfn is_armed(name: &str) -> bool { armed().as_deref() == Some(name)}
/// tells the test which failpoint fired, through `KNOT_FAILPOINT_FILE`fn signal(name: &str) -> Option<String> { let file = std::env::var("KNOT_FAILPOINT_FILE") .ok() .filter(|file| !file.is_empty())?; let path = file.as_str(); let _ = std::fs::write(path, format!("{name}\n")); Some(file)}
/// armed, it waits until the test removes the file it signalled throughpub fn pause(name: &str) { if !is_armed(name) { return; } let Some(file) = signal(name) else { return; }; let deadline = std::time::Instant::now() + std::time::Duration::from_secs(60); while std::path::Path::new(&file).exists() && std::time::Instant::now() < deadline { std::thread::sleep(std::time::Duration::from_millis(20)); }}
pub fn check(name: &str) -> anyhow::Result<()> { if !is_armed(name) { return Ok(()); } if let Ok(path) = std::env::var("KNOT_FAILPOINT_FILE") && !path.is_empty() { let _ = std::fs::write(path, format!("{name}\n")); } signal(name); anyhow::bail!("failpoint {name} armed (KNOT_FAILPOINT)")}
#[cfg(test)]19 unmodified lines14 unmodified lines15161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960220 unmodified linesExpand all281282283284285286287288289290291292293294295178 unmodified linesExpand all14 unmodified lines/// the extra minute covers the reload landing after the activation record.pub(crate) const DRAIN_SECS: u64 = crate::nginx::WORKER_SHUTDOWN_SECS + 60;
const STORE_DIRS: [&str; 2] = ["releases", "cache/export"];const PINS_DIR: &str = "pins";
/// a build holds it until its activation, so another activation's gc leaves the candidate alone#[derive(Debug)]pub struct Pin { _file: std::fs::File,}
/// gc only runs under the output lock, so take it under that lock toopub(crate) fn pin(output_root: &Path, generation: &str) -> anyhow::Result<Pin> { let dir = output_root.join(PINS_DIR); std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?; let path = dir.join(generation); let file = std::fs::OpenOptions::new() .create(true) .truncate(false) .write(true) .open(&path) .with_context(|| format!("open pin {}", path.display()))?; // shared, two builds can make the same generation file.lock_shared() .with_context(|| format!("pin {}", path.display()))?; Ok(Pin { _file: 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 { return false; }; if file.try_lock().is_err() { return true; } let _ = std::fs::remove_file(&path); false}
#[derive(Deserialize)]struct Live { #[serde(default)] repos: Vec<LiveRepo>,220 unmodified linesExpand all unrecorded.sort_unstable_by(|a, b| b.cmp(a)); let missing = keep - targets.len(); targets.extend(unrecorded.into_iter().take(missing).map(|(_, name)| name)); kept.extend(targets.into_iter().map(str::to_string)); let on_the_way: Vec<String> = generations .iter() .map(|(name, _)| name) .filter(|name| !kept.contains(*name) && pinned(output_root, name)) .cloned() .collect(); kept.extend(on_the_way);
for (name, _) in &generations { if !kept.contains(name) { remove(&generations_root.join(name))?;178 unmodified linesExpand all50 unmodified lines5152535455555657585960616263626364656667686970717273747567686970644 unmodified linesExpand all71571671771871972072173073173273373473573673772272372450 unmodified lines matches!(self.outcome, RepoOutcome::Unchanged { .. }) }}
#[derive(Debug, Clone, PartialEq)]#[derive(Debug)]pub enum BuildOutcome { Unchanged { generation_id: String, }, Built { generation_id: String, publication: Publication, telemetry: BuildTelemetry, /// keep it until the generation is activated pin: crate::gc::Pin, },}
#[derive(Debug, Clone, PartialEq, Default, serde::Serialize, serde::Deserialize)]pub(crate) struct BuildTelemetry { planning_seconds: f64, git_export_seconds: f64, staging_seconds: f64, publication_seconds: f64, total_wall_seconds: f64,}
fn entry_release_sealed(output_root: &Path, entry: &RepoEntry) -> bool { let Some(release_id) = &entry.release_id else { return false; };644 unmodified linesExpand all total_wall_seconds, generation_dir.display() ); } let pin = crate::gc::pin(&output_root, &manifest.generation_id)?; Ok(BuildOutcome::Built { generation_id: manifest.generation_id, publication, telemetry: BuildTelemetry { planning_seconds, git_export_seconds, staging_seconds, publication_seconds, total_wall_seconds, }, pin, })}471 unmodified linesExpand all47247347447547647747847948011 unmodified lines49249349449549549649749849950050150250350450550649649749849950050150250350450550650750850850951051151251388 unmodified lines471 unmodified linesExpand all config: Arc<ValidatedConfig>, generation_id: &str, repo_hint: Option<&str>,) -> anyhow::Result<Vec<String>> { crate::failpoint::pause("before_activation"); let first = match crate::failpoint::check("merge_rebuild") { Ok(()) => crate::activate::activate(config.clone(), generation_id).await, Err(_) => Err(crate::publish::StalePlan::injected().into()), };11 unmodified lines Ok(()) => crate::generate::run(config.clone(), repo_hint).await, Err(error) => Err(error), }; match rebuilt { Ok(crate::generate::BuildOutcome::Built { generation_id, .. }) => { match crate::activate::activate(config.clone(), &generation_id).await { Ok(result) => { out!( "generation {} activated after merge", manifest::short_id(&result.generation) ); Ok(result.changed_repos) } Err(second) => { Err(second.context("activation failed after the merge rebuild")) } Ok(crate::generate::BuildOutcome::Built { generation_id, pin: _pin, .. }) => match crate::activate::activate(config.clone(), &generation_id).await { Ok(result) => { out!( "generation {} activated after merge", manifest::short_id(&result.generation) ); Ok(result.changed_repos) } } Err(second) => Err(second.context("activation failed after the merge rebuild")), }, Ok(crate::generate::BuildOutcome::Unchanged { generation_id }) => { out!( "generation {} already covers the merged state", manifest::short_id(&generation_id)88 unmodified lines198 unmodified linesExpand all19920020120220320420520320420520620720820921021131 unmodified lines198 unmodified linesExpand all repo_hint: Option<&str>,) -> anyhow::Result<Vec<String>> { match crate::generate::run(config.clone(), repo_hint).await? { crate::generate::BuildOutcome::Unchanged { .. } => Ok(Vec::new()), crate::generate::BuildOutcome::Built { generation_id, .. } => { crate::git_ssh::activate_with_merge(config.clone(), &generation_id, repo_hint).await } crate::generate::BuildOutcome::Built { generation_id, pin: _pin, .. } => crate::git_ssh::activate_with_merge(config.clone(), &generation_id, repo_hint).await, }}
/// deregister first. a crash between the two leaves an orphan directory.31 unmodified lines382 unmodified linesExpand all383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413382 unmodified linesExpand all assert_eq!(generations(&path).len(), 3); drop(held); assert_eq!(worker.join().unwrap().unwrap().generations, 1);}
#[test]fn a_pinned_candidate_and_its_release_outlast_another_activations_gc() { let root = knot(&["g1", "g2", "g3", "g4"]); let root = root.path(); activate(root, "g4", 5, &[("g3", 5)]); let pin = crate::gc::pin(root, "g1").unwrap(); let left_over = root.join(PINS_DIR).join("g2"); std::fs::write(&left_over, "").unwrap();
collect_at(root, 1, NOW, DRAIN_SECS).unwrap();
assert_eq!(generations(root), ["g1", "g3", "g4"]); assert!( release(root, "r-g1").is_dir(), "the candidate's release stays" ); assert!(!left_over.exists(), "a pin nobody holds goes");
drop(pin); let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); while generations(root).contains(&"g1".to_string()) { assert!(std::time::Instant::now() < deadline, "g1 stays pinned"); collect_at(root, 1, NOW, DRAIN_SECS).unwrap(); } assert!(!root.join(PINS_DIR).join("g1").exists());}