use super::*;
use crate::publish::{BackendRecord, Superseded};
const DID: &str = "did:plc:a";
const NOW: u64 = 1_000_000;
fn repo_dir(root: &Path) -> PathBuf {
crate::repolock::repo_dir(root, &repo_key(DID))
}
fn release(root: &Path, id: &str) -> PathBuf {
repo_dir(root).join("releases").join(id)
}
fn export_entry(root: &Path, fingerprint: &str) -> PathBuf {
RepoStore::open(&repo_dir(root)).entry_path(crate::git_export::EXPORT_CACHE_KIND, fingerprint)
}
fn generation(root: &Path, id: &str, created: u64) {
let dir = root.join("generations").join(id);
std::fs::create_dir_all(&dir).unwrap();
let manifest = serde_json::json!({
"generation_id": id,
"repos": [{
"repo_did": DID,
"release_id": format!("r-{id}"),
"git_export_hash": format!("e-{id}"),
"object_format": "sha1",
"a_field_gc_never_reads": true,
}],
"stats": { "created_at_unix": created },
});
std::fs::write(dir.join("manifest.json"), manifest.to_string()).unwrap();
for store_entry in [
release(root, &format!("r-{id}")),
export_entry(root, &format!("e-{id}")),
] {
std::fs::create_dir_all(&store_entry).unwrap();
std::fs::write(store_entry.join("payload"), id).unwrap();
}
}
fn activate(root: &Path, id: &str, activated_at: u64, history: &[(&str, u64)]) {
let link = root.join("current");
let _ = std::fs::remove_file(&link);
std::os::unix::fs::symlink(Path::new("generations").join(id), link).unwrap();
crate::publish::write_backend_record(
root,
&BackendRecord {
revision: 1,
generation: id.to_string(),
rollback: false,
reconciled: false,
activated_at_unix: activated_at,
telemetry: None,
superseded: history
.iter()
.map(|(generation, at_unix)| Superseded {
generation: generation.to_string(),
at_unix: *at_unix,
})
.collect(),
},
)
.unwrap();
}
fn history(root: &Path) -> Vec<String> {
crate::publish::load_backend_record(root)
.unwrap()
.unwrap()
.superseded
.into_iter()
.map(|entry| entry.generation)
.collect()
}
fn generations(root: &Path) -> Vec<String> {
let mut names: Vec<String> = std::fs::read_dir(root.join("generations"))
.unwrap()
.map(|entry| entry.unwrap().file_name().into_string().unwrap())
.collect();
names.sort();
names
}
fn knot(ids: &[&str]) -> tempfile::TempDir {
let root = tempfile::tempdir().unwrap();
for (created, id) in ids.iter().enumerate() {
generation(root.path(), id, created as u64 + 1);
}
root
}
#[test]
fn ref_delta_answers_stay_while_a_kept_export_holds_their_pack() {
let root = knot(&["g1"]);
let root = root.path();
activate(root, "g1", 1, &[]);
let entry = export_entry(root, "e-g1");
crate::bench_fixture::repo(&entry, 2, 8, 0);
let idx = std::fs::read_dir(entry.join("objects/pack"))
.unwrap()
.map(|entry| entry.unwrap().path())
.find(|path| path.extension().is_some_and(|ext| ext == "idx"))
.unwrap();
let stem = idx.file_stem().unwrap().to_str().unwrap();
std::fs::write(
entry.join(crate::partitions::PARTITIONS),
format!("public {stem}\n"),
)
.unwrap();
let kept = gix::odb::pack::index::File::at(&idx, gix::hash::Kind::Sha1)
.unwrap()
.pack_checksum()
.to_string();
let store = RepoStore::open(&repo_dir(root));
let kind = crate::partitions::REF_DELTAS_KIND;
for key in [kept.as_str(), "gone"] {
store.publish_file(kind, key, &false).unwrap();
}
let in_flight = store.kind_dir(kind).join(".tmp-in-flight");
std::fs::write(&in_flight, "").unwrap();
let removed = collect_at(root, 1, NOW, DRAIN_SECS).unwrap();
assert_eq!(removed.store_entries, 1);
assert!(store.file_path(kind, &kept).exists());
assert!(!store.file_path(kind, "gone").exists());
assert!(in_flight.exists());
}
#[test]
fn keeps_current_pending_and_rollback_targets_and_what_they_name() {
let root = knot(&["g1", "g2", "g3", "g4", "g5"]);
let root = root.path();
activate(root, "g3", 3, &[("g1", 2), ("g2", 3)]);
std::fs::create_dir_all(release(root, "orphan")).unwrap();
std::fs::create_dir_all(release(root, ".building-dead")).unwrap();
std::fs::create_dir_all(root.join("generations/.building-dead")).unwrap();
for runtime in ["runtime/g1/proxy_temp", "runtime/g3/proxy_temp"] {
std::fs::create_dir_all(root.join(runtime)).unwrap();
}
std::fs::write(root.join("runtime/nginx.pid"), "1\n").unwrap();
let removed = collect_at(root, 1, NOW, DRAIN_SECS).unwrap();
assert_eq!(generations(root), ["g2", "g3", "g4", "g5"]);
assert_eq!(
removed,
Removed {
generations: 1,
store_entries: 3,
staging: 2,
..Removed::default()
}
);
for id in ["g2", "g3", "g4", "g5"] {
assert!(release(root, &format!("r-{id}")).is_dir(), "{id}'s release");
assert!(
export_entry(root, &format!("e-{id}")).is_dir(),
"{id}'s delta seed"
);
}
assert!(!release(root, "r-g1").exists());
assert!(!export_entry(root, "e-g1").exists());
assert!(!release(root, "orphan").exists());
assert!(
!root.join("runtime/g1").exists(),
"g1's runtime goes with it"
);
assert!(root.join("runtime/g3").is_dir());
assert!(root.join("runtime/nginx.pid").is_file());
assert_eq!(history(root), ["g2"], "the history forgets what gc removed");
assert!(
collect_at(root, 1, NOW, DRAIN_SECS).unwrap().is_empty(),
"a second pass is a no-op"
);
}
/// several builds share a second here, and only the activation order can pick the rollback.
#[test]
fn rollback_targets_follow_activation_order_not_build_time() {
let root = tempfile::tempdir().unwrap();
let root = root.path();
for (id, created) in [
("a", 1),
("b", 2),
("c", 2),
("d", 2),
("pending", 2),
("never", 1),
] {
generation(root, id, created);
}
activate(root, "c", 2, &[("a", 1), ("d", 2), ("b", 2)]);
collect_at(root, 1, NOW, DRAIN_SECS).unwrap();
assert_eq!(generations(root), ["b", "c", "pending"]);
assert_eq!(history(root), ["b"]);
}
#[test]
fn build_time_fills_rollback_slots_no_activation_recorded() {
let root = knot(&["g1", "g2", "g3", "g4"]);
let root = root.path();
activate(root, "g4", 4, &[("g2", 4)]);
collect_at(root, 2, NOW, DRAIN_SECS).unwrap();
assert_eq!(generations(root), ["g2", "g3", "g4"]);
}
#[test]
fn a_generation_nginx_may_still_drain_survives_until_the_window_ends() {
let root = knot(&["g1", "g2", "g3"]);
let root = root.path();
activate(
root,
"g3",
NOW,
&[("g1", NOW - DRAIN_SECS + 1), ("g2", NOW)],
);
collect_at(root, 1, NOW, DRAIN_SECS).unwrap();
assert_eq!(generations(root), ["g1", "g2", "g3"]);
assert!(release(root, "r-g1").is_dir());
activate(root, "g3", NOW, &[("g1", NOW - DRAIN_SECS), ("g2", NOW)]);
collect_at(root, 1, NOW, DRAIN_SECS).unwrap();
assert_eq!(generations(root), ["g2", "g3"]);
}
#[test]
fn force_skips_the_drain_window_and_nothing_else() {
let root = knot(&["g1", "g2", "g3", "g4", "g5"]);
let root = root.path();
let recent = [("g1", NOW - 5), ("g2", NOW - 4), ("g3", NOW)];
activate(root, "g4", 4, &recent);
collect_at(root, 1, NOW, DRAIN_SECS).unwrap();
assert_eq!(
generations(root),
["g1", "g2", "g3", "g4", "g5"],
"inside the window nothing goes"
);
let removed = collect_at(root, 1, NOW, 0).unwrap();
assert_eq!(removed.generations, 2);
assert_eq!(
generations(root),
["g3", "g4", "g5"],
"current, pending and the newest rollback target stay"
);
assert!(release(root, "r-g3").is_dir());
assert!(!release(root, "r-g1").exists());
assert_eq!(history(root), ["g3"]);
}
#[test]
fn an_unreadable_manifest_keeps_its_generation_and_every_store_entry() {
let root = knot(&["g1", "g2", "g3"]);
let root = root.path();
let odd = root.join("generations/odd");
std::fs::create_dir_all(&odd).unwrap();
std::fs::write(odd.join("manifest.json"), b"{ not json").unwrap();
activate(root, "g3", 3, &[("g1", 2), ("g2", 3)]);
collect_at(root, 1, NOW, DRAIN_SECS).unwrap();
assert_eq!(generations(root), ["g2", "g3", "odd"]);
assert!(
release(root, "r-g1").is_dir(),
"odd may name it, so no store entry goes"
);
}
#[test]
fn a_kept_seal_that_does_not_read_keeps_the_served_store() {
let root = knot(&["g1", "g2"]);
let root = root.path();
let manifest = root.join("generations/g2/manifest.json");
let mut view: serde_json::Value =
serde_json::from_slice(&std::fs::read(&manifest).unwrap()).unwrap();
view["profile"] = serde_json::json!({ "atproto_entries": { DID: "1-gone" } });
std::fs::write(&manifest, view.to_string()).unwrap();
let pooled = repo_dir(root).join("served/atproto/records/a.json");
std::fs::create_dir_all(pooled.parent().unwrap()).unwrap();
std::fs::write(&pooled, "{}").unwrap();
activate(root, "g2", 2, &[("g1", 2)]);
collect_at(root, 1, NOW, DRAIN_SECS).unwrap();
assert!(pooled.is_file());
}
/// the newest file is the wrong answer after a concurrent build or a clock step.
#[test]
fn cob_states_keep_the_repo_state_the_last_build_recorded() {
let root = knot(&["g1"]);
let root = root.path();
activate(root, "g1", 1, &[]);
let store = RepoStore::open(&repo_dir(root));
let state = |cobs: &[&str]| serde_json::json!({ "records": [], "versions": {}, "cobs": cobs });
store
.publish_file("atproto-cob", "1-gone", &"gone")
.unwrap();
store
.publish_file("atproto-cob", "1-kept", &"kept")
.unwrap();
store
.publish_file("atproto-cobs", "1-kept-state", &state(&["1-kept"]))
.unwrap();
crate::atproto::record_cob_state(
&Datastore::at(root),
&RepoKey::parse_did(DID).unwrap(),
&crate::datastore::keys::CobStateKey::new("1-kept-state").unwrap(),
)
.unwrap();
store
.publish_file("atproto-cobs", "1-newer", &state(&["1-gone"]))
.unwrap();
let in_flight = store.kind_dir("atproto-cob").join(".1-new.json.tmp.1.0");
std::fs::write(&in_flight, "").unwrap();
let removed = collect_at(root, 1, NOW, DRAIN_SECS).unwrap();
assert_eq!(removed.store_entries, 2);
assert!(!store.file_path("atproto-cobs", "1-newer").exists());
assert!(!store.file_path("atproto-cob", "1-gone").exists());
assert!(store.file_path("atproto-cobs", "1-kept-state").is_file());
assert!(store.file_path("atproto-cob", "1-kept").is_file());
assert!(in_flight.is_file());
}
#[test]
fn cob_states_stay_until_a_build_records_which_one_is_current() {
let root = knot(&["g1"]);
let root = root.path();
activate(root, "g1", 1, &[]);
let store = RepoStore::open(&repo_dir(root));
store.publish_file("atproto-cob", "1-cob", &"cob").unwrap();
store
.publish_file("atproto-cobs", "1-state", &serde_json::json!({}))
.unwrap();
let removed = collect_at(root, 1, NOW, DRAIN_SECS).unwrap();
assert_eq!(removed.store_entries, 0);
assert!(store.file_path("atproto-cobs", "1-state").is_file());
assert!(store.file_path("atproto-cob", "1-cob").is_file());
}
#[test]
fn nothing_goes_before_the_first_activation_or_with_a_tombstone() {
let root = knot(&["g1", "g2", "g3", "g4"]);
let root = root.path();
collect_at(root, 1, NOW, DRAIN_SECS).unwrap();
assert_eq!(generations(root).len(), 4);
activate(root, "g4", 4, &[]);
crate::tombstone::write(root, "g4").unwrap();
assert!(collect_at(root, 1, NOW, DRAIN_SECS).is_err());
assert_eq!(generations(root).len(), 4);
}
#[test]
fn gc_waits_for_the_output_lock() {
let root = knot(&["g1", "g2", "g3"]);
let path = root.path().to_path_buf();
activate(&path, "g3", 3, &[("g1", 2), ("g2", 3)]);
let held = crate::output::lock(&path).unwrap();
let worker = std::thread::spawn({
let path = path.clone();
move || collect_locking(&path, 1, DRAIN_SECS)
});
std::thread::sleep(std::time::Duration::from_millis(200));
assert!(
!worker.is_finished(),
"gc ran while a build or activation held the lock"
);
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());
}
#[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"]);
}
#[test]
fn a_retired_generation_goes_once_nothing_older_than_its_retirement_reads() {
let root = knot(&["g1", "g2", "g3"]);
let root = root.path();
activate(root, "g3", NOW, &[("g1", NOW - 10), ("g2", NOW)]);
let kept = |reader| {
collect_with(root, 1, NOW, DRAIN_SECS, reader).unwrap();
generations(root)
};
assert_eq!(
kept(None),
["g1", "g2", "g3"],
"unknown readers wait out the window"
);
assert_eq!(
kept(Some(NOW - 20)),
["g1", "g2", "g3"],
"an old worker still runs"
);
assert_eq!(
kept(Some(NOW - 10)),
["g1", "g2", "g3"],
"one started the second g1 retired"
);
assert_eq!(kept(Some(NOW - 9)), ["g2", "g3"]);
}
#[test]
fn the_oldest_reader_is_a_worker_or_a_request_started_first() {
let root = tempfile::tempdir().unwrap();
let (proc, output) = (root.path().join("proc"), root.path().join("out"));
std::fs::create_dir_all(output.join("runtime")).unwrap();
std::fs::create_dir_all(&proc).unwrap();
std::fs::write(proc.join("stat"), "cpu 1 2 3\nbtime 1000\n").unwrap();
let process = |pid: u32, name: &str, parent: u32, started_ticks: u64, args: &[&str]| {
let dir = proc.join(pid.to_string());
std::fs::create_dir_all(&dir).unwrap();
let zeros = "0 ".repeat(17);
let stat = format!("{pid} ({name}) S {parent} {zeros}{started_ticks} 0 0\n");
std::fs::write(dir.join("stat"), stat).unwrap();
let command: Vec<u8> = args
.iter()
.flat_map(|arg| [arg.as_bytes(), b"\0"].concat())
.collect();
std::fs::write(dir.join("cmdline"), command).unwrap();
};
assert_eq!(oldest_reader_in(&proc, &output, 100), None, "no nginx pid");
std::fs::write(output.join("runtime/nginx.pid"), "10\n").unwrap();
process(1, "systemd", 0, 0, &["init"]);
process(10, "nginx", 1, 100, &["nginx: master process"]);
process(20, "fcgiwrap", 1, 200, &["fcgiwrap"]);
process(21, "fcgiwrap", 20, 200, &["fcgiwrap"]);
// services up since boot read no generation
process(
30,
"knot-static-fir",
1,
300,
&["knot-static-firehose", "/out/firehose"],
);
process(
31,
"knot-static",
1,
300,
&["knot-static", "maintenance", "--all"],
);
process(32, "git", 31, 400, &["git", "repack"]);
assert_eq!(
oldest_reader_in(&proc, &output, 100),
Some(u64::MAX),
"nothing reads"
);
process(11, "nginx", 10, 5000, &["nginx: worker process"]);
process(40, "knot-static", 21, 3000, &["knot-static"]);
process(41, "git", 40, 9000, &["git", "upload-pack"]);
assert_eq!(
oldest_reader_in(&proc, &output, 100),
Some(1030),
"a cgi request"
);
process(50, "sshd", 1, 100, &["sshd"]);
process(
51,
"knot-static",
50,
2500,
&["knot-static", "--config", "c", "git-ssh"],
);
assert_eq!(
oldest_reader_in(&proc, &output, 100),
Some(1025),
"a git-ssh session"
);
process(
12,
"nginx",
10,
2000,
&["nginx: worker process is shutting down"],
);
assert_eq!(
oldest_reader_in(&proc, &output, 100),
Some(1020),
"an old worker"
);
}
#[test]
fn wait_for_reload_waits_until_new_workers_replace_every_old_one() {
use std::time::{Duration, Instant};
fn write_proc(proc: &Path, pid: u32, parent: u32, cmdline: &str) {
let dir = proc.join(pid.to_string());
std::fs::create_dir_all(&dir).unwrap();
let zeros = "0 ".repeat(17);
std::fs::write(
dir.join("stat"),
format!("{pid} (nginx) S {parent} {zeros}100 0 0\n"),
)
.unwrap();
std::fs::write(dir.join("cmdline"), format!("{cmdline}\0")).unwrap();
}
let root = tempfile::tempdir().unwrap();
let (proc, output) = (root.path().join("proc"), root.path().join("out"));
std::fs::create_dir_all(output.join("runtime")).unwrap();
std::fs::create_dir_all(&proc).unwrap();
assert!(active_workers_in(&proc, &output).is_empty());
assert!(wait_for_reload_in(
&proc,
&output,
&BTreeSet::new(),
Duration::from_millis(1)
));
std::fs::write(output.join("runtime/nginx.pid"), "10\n").unwrap();
write_proc(&proc, 10, 1, "nginx: master process");
write_proc(&proc, 11, 10, "nginx: worker process");
write_proc(&proc, 12, 10, "nginx: worker process");
write_proc(&proc, 19, 10, "nginx: worker process is shutting down");
let before = active_workers_in(&proc, &output);
assert_eq!(before, BTreeSet::from([11, 12]));
assert!(
!wait_for_reload_in(&proc, &output, &before, Duration::from_millis(25)),
"old workers still active"
);
let proc_bg = proc.clone();
let swap = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(30));
// new workers fork before old ones process SIGQUIT
write_proc(&proc_bg, 13, 10, "nginx: worker process");
std::thread::sleep(Duration::from_millis(40));
std::fs::remove_dir_all(proc_bg.join("11")).unwrap();
write_proc(&proc_bg, 12, 10, "nginx: worker process is shutting down");
});
let t0 = Instant::now();
assert!(wait_for_reload_in(
&proc,
&output,
&before,
Duration::from_secs(2)
));
let elapsed = t0.elapsed();
swap.join().unwrap();
assert!(
elapsed >= Duration::from_millis(60),
"returned while an old worker was still active: {elapsed:?}"
);
assert_eq!(active_workers_in(&proc, &output), BTreeSet::from([13]));
}