mod common;
use common::{git, git_raw, repo_key};
use knot_runtime::{K256Signer, SeededEntropy};
use serde::Deserialize;
use std::collections::BTreeMap;
use std::fs::{File, OpenOptions};
use std::path::{Path, PathBuf};
use std::process::{Child, ExitStatus, Output, Stdio};
use std::time::{Duration, Instant};
const OWNER: &str = "did:web:owner.repolock.test";
const COLLABORATOR: &str = "did:web:collaborator.repolock.test";
const KNOT_HOST: &str = "knot.repolock.test";
const MASTER_KEY_ENV: &str = "KNOT_STATIC_LOCK_TEST_KEY";
const MASTER_KEY: [u8; 32] = [7; 32];
const ZERO_OID: &str = "0000000000000000000000000000000000000000";
#[derive(Deserialize, Debug)]
struct Manifest {
#[serde(default)]
repos: Vec<RepoEntry>,
}
#[derive(Deserialize, Debug, Clone)]
struct RepoEntry {
repo_did: String,
#[serde(default)]
head: Option<String>,
}
impl Manifest {
fn load(generation: &Path) -> Self {
let bytes = std::fs::read(generation.join("manifest.json")).expect("generation manifest");
serde_json::from_slice(&bytes).expect("parse generation manifest")
}
fn head_of(&self, repo_did: &str) -> Option<String> {
self.repos
.iter()
.find(|entry| entry.repo_did == repo_did)
.and_then(|entry| entry.head.clone())
}
}
struct Env {
_root: tempfile::TempDir,
config: PathBuf,
output: PathBuf,
repos: BTreeMap<String, (String, PathBuf)>,
}
fn knot_with_env(
config: &Path,
args: impl IntoIterator<Item = impl AsRef<std::ffi::OsStr>>,
env: &[(&str, &str)],
) -> Output {
common::knot_static_raw(config, MASTER_KEY_ENV, &MASTER_KEY, args, env)
}
fn knot(config: &Path, args: &[&str]) -> Output {
knot_with_env(config, args, &[])
}
fn knot_ok(config: &Path, args: &[&str]) -> Output {
common::knot_static(config, MASTER_KEY_ENV, &MASTER_KEY, args)
}
fn spawn_knot(env: &Env, args: &[&str], extra_env: &[(&str, &str)]) -> Child {
let mut cmd = common::knot_static_cmd(&env.config, MASTER_KEY_ENV, &MASTER_KEY, args);
cmd.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
for (key, value) in extra_env {
cmd.env(key, value);
}
cmd.spawn().expect("spawn knot-static")
}
fn commit_on(repo: &Path, parent: &str, message: &str) -> String {
let tree_spec = format!("{parent}^{{tree}}");
let tree = git(repo, &["rev-parse", tree_spec.as_str()]);
git(repo, &["commit-tree", &tree, "-p", parent, "-m", message])
}
fn lock_file(env: &Env, rkey: &str) -> File {
let (did, _) = &env.repos[rkey];
let path = env
.output
.join("repos")
.join(repo_key(did))
.join(".build.lock");
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).unwrap();
}
OpenOptions::new()
.create(true)
.read(true)
.write(true)
.open(&path)
.unwrap_or_else(|e| panic!("open {}: {e}", path.display()))
}
fn hold_exclusive(env: &Env, rkey: &str) -> File {
let file = lock_file(env, rkey);
file.lock().expect("exclusive repo lock");
file
}
fn hold_lifetime(env: &Env, rkey: &str) -> File {
let file = lock_file(env, rkey);
file.lock_shared().expect("lifetime repo lock");
file
}
/// a fresh open, since flock is per open file, not per thread.
fn exclusive_available(env: &Env, rkey: &str) -> bool {
match lock_file(env, rkey).try_lock() {
Ok(()) => true,
Err(std::fs::TryLockError::WouldBlock) => false,
Err(error) => panic!("try_lock: {error}"),
}
}
fn wait_bounded(child: &mut Child, budget: Duration) -> Option<ExitStatus> {
let deadline = Instant::now() + budget;
loop {
if let Some(status) = child.try_wait().expect("try_wait") {
return Some(status);
}
if Instant::now() >= deadline {
return None;
}
std::thread::sleep(Duration::from_millis(20));
}
}
fn finish(child: Child, label: &str) -> Output {
let output = child.wait_with_output().expect("wait_with_output");
assert!(
output.status.success(),
"{label} failed: {}{}",
String::from_utf8_lossy(&output.stderr),
String::from_utf8_lossy(&output.stdout)
);
output
}
fn make_source(source: &Path, seed: &str) {
std::fs::create_dir_all(source).unwrap();
git(source, &["init", "-q", "-b", "main"]);
std::fs::write(source.join("README.md"), format!("# {seed}\n")).unwrap();
git(source, &["add", "-A"]);
git(source, &["commit", "-q", "-m", "initial commit"]);
std::fs::write(source.join(format!("{seed}.rs")), "fn main() {}\n").unwrap();
git(source, &["add", "-A"]);
git(source, &["commit", "-q", "-m", "add code"]);
git(source, &["tag", "v1", "HEAD"]);
}
fn write_config(path: &Path, scan: &Path, output: &Path, root: &Path) {
std::fs::write(
path,
format!(
r#"[knot]
hostname = "{KNOT_HOST}"
owner = "{OWNER}"
[repo]
scan_path = "{}"
default_branch = "main"
object_format = "sha1"
[secrets]
sealed_key_file = "{}"
master_key_env = "KNOT_STATIC_LOCK_TEST_KEY"
[atproto]
plc_directory = "https://127.0.0.1:9"
[output]
root = "{}"
[activate]
nginx_binary = "true"
reload_command = "true"
"#,
scan.display(),
root.join("keys.sealed").display(),
output.display(),
),
)
.unwrap();
}
/// with `build`, a later `--repo` build carries the other repos instead of capturing them.
fn setup(build: bool) -> Env {
let root = tempfile::tempdir().expect("tempdir");
let scan = root.path().join("state");
let output = root.path().join("output");
let config = root.path().join("config.toml");
std::fs::create_dir_all(&scan).unwrap();
write_config(&config, &scan, &output, root.path());
knot_ok(&config, &["init"]);
let mut repos = BTreeMap::new();
for seed in ["alpha", "beta"] {
let source = root.path().join(format!("source-{seed}"));
make_source(&source, seed);
let _ = knot(
&config,
&[
"repo",
"import",
"--owner",
OWNER,
"--rkey",
seed,
"--name",
seed,
"--default-branch",
"main",
source.to_str().unwrap(),
],
);
let listed = knot_ok(&config, &["repo", "list"]);
let line = String::from_utf8(listed.stdout)
.unwrap()
.lines()
.find(|line| line.contains(seed))
.unwrap_or_else(|| panic!("repo {seed} must be listed"))
.to_string();
let fields: Vec<&str> = line.split_whitespace().collect();
assert!(fields.len() >= 6, "repo list row: {line}");
repos.insert(
seed.to_string(),
(fields[0].to_string(), PathBuf::from(fields[5])),
);
}
let env = Env {
_root: root,
config,
output,
repos,
};
if build {
knot_ok(&env.config, &["generate"]);
let generation = newest_generation(&env.output);
knot_ok(
&env.config,
&[
"activate",
generation.file_name().unwrap().to_str().unwrap(),
],
);
}
env
}
fn newest_generation(output: &Path) -> PathBuf {
let mut dirs: Vec<PathBuf> = std::fs::read_dir(output.join("generations"))
.expect("generations root")
.flatten()
.map(|entry| entry.path())
.filter(|path| {
path.is_dir() && !path.file_name().unwrap().to_str().unwrap().starts_with('.')
})
.collect();
dirs.sort_by_key(|path| std::fs::metadata(path).unwrap().modified().unwrap());
dirs.pop().expect("at least one generation")
}
fn generation_count(output: &Path) -> usize {
std::fs::read_dir(output.join("generations"))
.map(|entries| {
entries
.flatten()
.filter(|entry| {
entry.file_type().map(|t| t.is_dir()).unwrap_or(false)
&& !entry.file_name().to_string_lossy().starts_with('.')
})
.count()
})
.unwrap_or(0)
}
fn object_exists(repo: &Path, oid: &str) -> bool {
git_raw(repo, &["cat-file", "-e", oid]).status.success()
}
fn ref_exists(repo: &Path, name: &str) -> bool {
git_raw(repo, &["rev-parse", "--verify", "--quiet", name])
.status
.success()
}
/// no objects travel because the ref points at one the source already has.
/// git writes it because a hand made pack would need a real trailer checksum.
fn empty_pack(repo: &Path) -> Vec<u8> {
let mut child = common::git_command(repo)
.args(["pack-objects", "--stdout"])
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn pack-objects");
drop(child.stdin.take());
let output = child.wait_with_output().expect("pack-objects");
assert!(
output.status.success(),
"pack-objects failed: {}",
String::from_utf8_lossy(&output.stderr)
);
assert_eq!(output.stdout.len(), 32, "expected an empty pack");
output.stdout
}
fn pkt_line(payload: &str) -> Vec<u8> {
let mut out = format!("{:04x}", payload.len() + 4).into_bytes();
out.extend_from_slice(payload.as_bytes());
out
}
fn push_payload(repo: &Path, new_oid: &str, ref_name: &str) -> Vec<u8> {
let mut payload = pkt_line(&format!(
"{ZERO_OID} {new_oid} {ref_name}\0 report-status side-band-64k agent=git/2.0\n"
));
payload.extend_from_slice(b"0000");
payload.extend_from_slice(&empty_pack(repo));
payload
}
fn spawn_transport(
env: &Env,
original_command: &str,
stdin_payload: Option<Vec<u8>>,
extra_env: &[(&str, &str)],
) -> Child {
spawn_transport_as(
env,
&["--operator"],
original_command,
stdin_payload,
extra_env,
)
}
fn spawn_transport_as(
env: &Env,
pusher: &[&str],
original_command: &str,
stdin_payload: Option<Vec<u8>>,
extra_env: &[(&str, &str)],
) -> Child {
let mut cmd = common::knot_static_cmd(
&env.config,
MASTER_KEY_ENV,
&MASTER_KEY,
["git-ssh"].iter().chain(pusher),
);
cmd.env("SSH_ORIGINAL_COMMAND", original_command)
.stdin(if stdin_payload.is_some() {
Stdio::piped()
} else {
Stdio::null()
})
.stdout(Stdio::piped())
.stderr(Stdio::piped());
for (key, value) in extra_env {
cmd.env(key, value);
}
let mut child = cmd.spawn().expect("spawn git-ssh");
if let Some(payload) = stdin_payload {
use std::io::Write as _;
let mut stdin = child.stdin.take().expect("stdin");
stdin.write_all(&payload).expect("write transport payload");
drop(stdin);
}
child
}
fn spawn_receive_pack(env: &Env, rkey: &str, payload: Vec<u8>) -> Child {
spawn_transport(
env,
&format!("git-receive-pack '/{rkey}.git'"),
Some(payload),
&[],
)
}
fn wait_until(label: &str, budget: Duration, mut predicate: impl FnMut() -> bool) {
let deadline = Instant::now() + budget;
loop {
if predicate() {
return;
}
assert!(Instant::now() < deadline, "timed out waiting for {label}");
std::thread::sleep(Duration::from_millis(25));
}
}
#[test]
fn managed_mutation_excludes_capture_and_capture_stays_coherent() {
let env = setup(true);
let (did, repo) = env.repos["alpha"].clone();
let before = generation_count(&env.output);
let held = hold_exclusive(&env, "alpha");
let tip_before = git(&repo, &["rev-parse", "refs/heads/main"]);
let tip_after = commit_on(&repo, tip_before.as_str(), "post-mutation commit");
git(
&repo,
&["update-ref", "refs/heads/main", tip_after.as_str()],
);
assert_ne!(tip_before, tip_after);
let mut build = spawn_knot(&env, &["generate", "--repo", did.as_str()], &[]);
assert!(
wait_bounded(&mut build, Duration::from_millis(1500)).is_none(),
"a build must not finish while a managed mutation holds the exclusive lock"
);
assert_eq!(
generation_count(&env.output),
before,
"no generation may be produced while the mutation holds the lock"
);
drop(held);
let output = finish(build, "generate --repo during a mutation");
assert!(
String::from_utf8_lossy(&output.stderr).contains("generated"),
"{}",
String::from_utf8_lossy(&output.stderr)
);
let manifest = Manifest::load(&newest_generation(&env.output));
assert_eq!(
manifest.head_of(&did).as_deref(),
Some(tip_after.as_str()),
"the build must capture the ref target the mutation left behind"
);
}
#[test]
fn managed_receive_pack_waits_for_the_source_lifetime_holder() {
let env = setup(true);
let (_, repo) = env.repos["alpha"].clone();
let tip = git(&repo, &["rev-parse", "refs/heads/main"]);
let control = spawn_receive_pack(
&env,
"alpha",
push_payload(&repo, &tip, "refs/heads/control"),
);
wait_until("control ref", Duration::from_secs(30), || {
ref_exists(&repo, "refs/heads/control")
});
let mut control = control;
let _ = control.kill();
let _ = control.wait();
let held = hold_lifetime(&env, "alpha");
let mut blocked = spawn_receive_pack(
&env,
"alpha",
push_payload(&repo, &tip, "refs/heads/blocked"),
);
assert!(
wait_bounded(&mut blocked, Duration::from_millis(900)).is_none(),
"receive-pack must wait for the source lifetime holder"
);
assert!(
!ref_exists(&repo, "refs/heads/blocked"),
"the managed writer must not mutate refs while a lifetime holder is active"
);
drop(held);
wait_until("blocked ref", Duration::from_secs(30), || {
ref_exists(&repo, "refs/heads/blocked")
});
assert_eq!(git(&repo, &["rev-parse", "refs/heads/blocked"]), tip);
let _ = blocked.kill();
let _ = blocked.wait();
}
#[test]
fn an_ssh_push_cannot_write_cob_refs() {
let env = setup(true);
let (did, repo) = env.repos["alpha"].clone();
let tip = git(&repo, &["rev-parse", "refs/heads/main"]);
let signer = K256Signer::generate(&SeededEntropy::new(5));
common::cobs::Cobs {
repo: &knot_git::Repo::open(&repo).unwrap(),
did: &knot_types::RepoDid::new(did).unwrap(),
signer: &signer,
}
.add_collaborator(
&knot_types::AccountDid::new(COLLABORATOR).unwrap(),
&knot_types::AccountDid::new(OWNER).unwrap(),
10,
);
for pusher in [&["--operator"][..], &["--actor", COLLABORATOR]] {
let output = spawn_transport_as(
&env,
pusher,
"git-receive-pack '/alpha.git'",
Some(push_payload(&repo, &tip, "refs/cobs/forged")),
&[],
)
.wait_with_output()
.expect("run receive-pack");
assert!(!ref_exists(&repo, "refs/cobs/forged"), "{pusher:?}");
assert!(
String::from_utf8_lossy(&output.stdout).contains("hidden ref"),
"{pusher:?}: {}",
String::from_utf8_lossy(&output.stdout)
);
}
}
#[test]
fn managed_prune_waits_for_the_lifetime_holder_and_keeps_objects_readable() {
let env = setup(true);
let (did, repo) = env.repos["alpha"].clone();
let orphan_in = |repo: &Path, message: &str| {
let parent = git(repo, &["rev-parse", "refs/heads/main"]);
let oid = commit_on(repo, parent.as_str(), message);
git(repo, &["update-ref", "refs/heads/orphan", oid.as_str()]);
git(repo, &["update-ref", "-d", "refs/heads/orphan"]);
oid
};
let orphan = |message: &str| orphan_in(&repo, message);
let control_oid = orphan("control orphan");
assert!(object_exists(&repo, &control_oid));
let started = Instant::now();
knot_ok(
&env.config,
&["maintenance", "--repo", did.as_str(), "--command", "prune"],
);
assert!(started.elapsed() < Duration::from_secs(60));
assert!(
!object_exists(&repo, &control_oid),
"the control run must really prune the unreachable object"
);
let required_oid = orphan("required orphan");
let held = hold_lifetime(&env, "alpha");
let mut maintenance = spawn_knot(
&env,
&["maintenance", "--repo", did.as_str(), "--command", "prune"],
&[],
);
assert!(
wait_bounded(&mut maintenance, Duration::from_millis(1200)).is_none(),
"managed maintenance must wait for the source lifetime holder"
);
assert!(
object_exists(&repo, &required_oid),
"an object a lifetime holder still needs must stay readable while maintenance waits"
);
drop(held);
finish(maintenance, "maintenance prune after the holder released");
assert!(
!object_exists(&repo, &required_oid),
"once released, the same maintenance run must complete"
);
let (_, beta) = env.repos["beta"].clone();
let orphans = [
(&repo, orphan("alpha orphan")),
(&beta, orphan_in(&beta, "beta orphan")),
];
knot_ok(&env.config, &["maintenance", "--all", "--command", "prune"]);
for (repo, oid) in orphans {
assert!(!object_exists(repo, &oid), "{}", repo.display());
}
}
#[test]
fn maintenance_of_every_repo_goes_on_past_one_that_fails() {
use std::os::unix::fs::PermissionsExt as _;
let env = setup(true);
let (alpha_did, alpha) = env.repos["alpha"].clone();
let (_, beta) = env.repos["beta"].clone();
let packs = alpha.join("objects/pack");
std::fs::set_permissions(&packs, std::fs::Permissions::from_mode(0o500)).unwrap();
let output = knot(
&env.config,
&["maintenance", "--all", "--command", "repack"],
);
std::fs::set_permissions(&packs, std::fs::Permissions::from_mode(0o755)).unwrap();
let stderr = String::from_utf8_lossy(&output.stderr);
assert!(!output.status.success(), "{stderr}");
assert!(
stderr.contains(&format!("repack failed for {alpha_did}")) && stderr.contains("git "),
"{stderr}"
);
let bitmaps = std::fs::read_dir(beta.join("objects/pack"))
.unwrap()
.filter(|entry| {
let name = entry.as_ref().unwrap().file_name();
name.to_string_lossy().ends_with(".bitmap")
})
.count();
assert_eq!(bitmaps, 1, "{stderr}");
}
#[test]
fn targeted_generate_locks_only_the_named_repository() {
let env = setup(true);
let (alpha_did, _) = env.repos["alpha"].clone();
let held_beta = hold_exclusive(&env, "beta");
finish(
spawn_knot(&env, &["generate", "--repo", alpha_did.as_str()], &[]),
"generate --repo alpha while beta is locked",
);
drop(held_beta);
let held_alpha = hold_exclusive(&env, "alpha");
let mut build = spawn_knot(&env, &["generate", "--repo", alpha_did.as_str()], &[]);
assert!(
wait_bounded(&mut build, Duration::from_millis(1200)).is_none(),
"a targeted build must take the named repository's lock"
);
drop(held_alpha);
finish(build, "generate --repo alpha after alpha released");
}
#[test]
fn concurrent_full_builds_share_one_lock_order_and_do_not_deadlock() {
let env = setup(true);
let mut children: Vec<Child> = (0..3)
.map(|_| spawn_knot(&env, &["generate"], &[]))
.collect();
for (index, child) in children.iter_mut().enumerate() {
assert!(
wait_bounded(child, Duration::from_secs(180)).is_some(),
"concurrent build {index} did not finish: repository locks deadlocked"
);
}
for (index, child) in children.into_iter().enumerate() {
let label = format!("concurrent build {index}");
finish(child, label.as_str());
}
for rkey in ["alpha", "beta"] {
let label = format!("{rkey} lock release");
wait_until(label.as_str(), Duration::from_secs(10), || {
exclusive_available(&env, rkey)
});
}
}
#[test]
fn failed_route_phase_joins_the_export_worker_before_releasing_locks() {
// with no prior generation the export worker is still busy when routing fails.
let env = setup(false);
let marker = env.output.join("failpoint.marker");
let output = knot_with_env(
&env.config,
&["generate"],
&[
("KNOT_FAILPOINT", "route_emission"),
("KNOT_FAILPOINT_FILE", marker.to_str().unwrap()),
],
);
assert!(
!output.status.success(),
"the armed failpoint must fail the build"
);
let stderr = String::from_utf8_lossy(&output.stderr).to_string();
assert!(stderr.contains("failpoint route_emission"), "{stderr}");
assert_eq!(
std::fs::read_to_string(&marker).unwrap().trim(),
"route_emission",
"the failpoint barrier must have fired"
);
for (rkey, (did, _)) in &env.repos {
let cache = env
.output
.join("repos")
.join(repo_key(did))
.join("cache/export");
let markers: Vec<PathBuf> = std::fs::read_dir(&cache)
.unwrap_or_else(|error| {
panic!(
"{rkey}: export cache {}: {error}\nchild stderr:\n{stderr}",
cache.display()
)
})
.flatten()
.map(|entry| entry.path().join(".complete.json"))
.filter(|path| path.is_file())
.collect();
assert!(
!markers.is_empty(),
"{rkey}: the export worker must have materialized and published its entry before the \
build returned; child stderr:\n{stderr}"
);
let staging: Vec<String> = std::fs::read_dir(&cache)
.unwrap()
.flatten()
.map(|entry| entry.file_name().to_string_lossy().into_owned())
.filter(|name| name.starts_with(".tmp-"))
.collect();
assert!(staging.is_empty(), "{rkey}: orphan staging {staging:?}");
}
for rkey in env.repos.keys() {
let label = format!("{rkey} lock release");
wait_until(label.as_str(), Duration::from_secs(10), || {
exclusive_available(&env, rkey)
});
}
}