use super::commit::{git_text, remove_settled};
use super::fault::{COMMIT_ARGUMENTS, Effect, Injection, When, staged_repository, verify_recovery_with};
use super::fault_limits::check_limited_run;
use super::fault_run::{end_the_pressure, in_batches, run_unsettled, run_watched, spawn_watched, unbounded_wrapper};
use super::strace::{strace_available, strace_program};
use super::support::{Fixture, ServiceLaunch, fixture, hermetic, service_runtime, wrapper_first_path};
use std::collections::HashMap;
use std::ffi::OsString;
use std::path::{Path, PathBuf};
use std::process::{Command, Output};
use std::sync::atomic::{AtomicUsize, Ordering};
const POSITIONS_PER_SWEEP: usize = 48;
const SLOW_FLUSH_MICROSECONDS: u64 = 100_000;
const STORAGE_CALLS: [&str; 6] = ["write", "pwrite64", "fsync", "fdatasync", "rename", "renameat2"];
const KILL_CALLS: [&str; 12] =
["write", "pwrite64", "fsync", "fdatasync", "rename", "renameat2", "unlink", "unlinkat", "mkdir", "mkdirat", "clone3", "vfork"];
const WRAPPER_KILL_CALLS: [&str; 10] = ["write", "pwrite64", "fsync", "rename", "renameat2", "unlink", "unlinkat", "mkdirat", "clone3", "vfork"];
fn tracer(trace: &Path, syscall: &str, fault: Option<&Injection>, follow: bool) -> Vec<OsString> {
let mut words: Vec<OsString> = vec![strace_program().expect("strace is installed").into_os_string()];
if follow {
words.push(OsString::from("--follow-forks"));
}
words.extend([OsString::from("--quiet"), OsString::from("--quiet")]);
words.push(OsString::from(format!("--output={}", trace.display())));
words.push(OsString::from(format!("--trace={syscall}")));
if let Some(injection) = fault {
words.push(OsString::from(format!("--inject={}", injection.specification())));
}
return words;
}
fn busiest_count(trace: &str, syscall: &str) -> usize {
let mut per_process: HashMap<&str, usize> = HashMap::new();
for line in trace.lines() {
let (process, call): (&str, &str) = match line.split_once(' ') {
Some((first, rest)) if first.bytes().all(|byte| { return byte.is_ascii_digit(); }) && !first.is_empty() => (first, rest.trim_start()),
_ => ("", line),
};
if call.starts_with(syscall) && call[syscall.len()..].starts_with('(') {
*per_process.entry(process).or_insert(0) += 1;
}
}
return per_process.values().copied().max().unwrap_or(0);
}
fn positions(count: usize, budget: usize) -> Vec<usize> {
if count <= budget {
return (1..=count).collect();
}
let mut spread: Vec<usize> = (0..budget).map(|step| { return 1 + (count - 1) * step / (budget - 1); }).collect();
spread.dedup();
return spread;
}
fn sweep_points(counts: &[(&'static str, usize)], budget: usize) -> Vec<(&'static str, usize)> {
let total: usize = counts.iter().map(|(_, count)| { return *count; }).sum();
let mut points: Vec<(&'static str, usize)> = Vec::new();
for (syscall, count) in counts {
if *count == 0 {
continue;
}
let share: usize = (budget * *count / total.max(1)).max(2);
points.extend(positions(*count, share).into_iter().map(|position| { return (*syscall, position); }));
}
return points;
}
type Recovery = fn(&Fixture, &Path, &str, &str) -> Result<(), String>;
fn recovers_any_way(fixture: &Fixture, repo: &Path, label: &str, said: &str) -> Result<(), String> {
return verify_recovery_with(fixture, repo, label, said, run_unsettled);
}
fn recovers_after_service_death(fixture: &Fixture, repo: &Path, label: &str, said: &str) -> Result<(), String> {
let status = run_unsettled(fixture, repo, &["status", "--porcelain"]);
if status.code != Some(0) {
return Err(format!("{label}: the next command must succeed, exit {:?}: {}\nfirst run said: {said}", status.code, String::from_utf8_lossy(status.stderr.as_slice())));
}
let commits: String = git_text(fixture, repo, &["rev-list", "--count", "HEAD"]);
let shown: String = String::from_utf8_lossy(status.stdout.as_slice()).into_owned();
match commits.trim() {
"1" if shown == "A a.txt\n" => {
let landed = run_unsettled(fixture, repo, &["commit", "--quiet", "--message=x", "--", "a.txt"]);
if landed.code != Some(0) {
return Err(format!("{label}: the commit after the kill must land, exit {:?}: {}\nfirst run said: {said}", landed.code, String::from_utf8_lossy(landed.stderr.as_slice())));
}
}
"2" if shown.is_empty() => {}
other => return Err(format!("{label}: {other} commits with status {shown:?}; first run said: {said}")),
}
return verify_recovery_with(fixture, repo, label, said, run_unsettled);
}
struct Found {
reached: AtomicUsize,
landed: AtomicUsize,
}
fn took_effect(trace: &str) -> bool {
return trace.contains("(INJECTED)") || trace.contains("(DELAYED)") || trace.contains("+++ killed by SIGKILL +++");
}
fn service_case(label: &str, fault: &Injection, found: &Found, recover: Recovery) -> Result<(), String> {
let fixture: Fixture = fixture(label);
let trace: PathBuf = fixture.root.join("service-trace.txt");
*fixture.service_launch.lock().expect("the service launch") =
ServiceLaunch { prefix: tracer(trace.as_path(), fault.syscall, Some(fault), fault.follow_forks), may_fail_to_start: true, ..ServiceLaunch::default() };
let repo: PathBuf = staged_repository(&fixture, "repo");
let output: Output = run_watched(&fixture, &mut unbounded_wrapper(&fixture, repo.as_path(), &COMMIT_ARGUMENTS));
let said: String = String::from_utf8_lossy(&[output.stdout.as_slice(), output.stderr.as_slice()].concat()).into_owned();
end_the_pressure(&fixture);
let traced: String = std::fs::read_to_string(&trace).unwrap_or_default();
if took_effect(traced.as_str()) {
found.reached.fetch_add(1, Ordering::SeqCst);
}
if output.status.code() == Some(0) {
found.landed.fetch_add(1, Ordering::SeqCst);
}
let verdict: Result<(), String> = check_limited_run(label, &output, false)
.and_then(|()| { return recover(&fixture, repo.as_path(), label, said.as_str()); });
remove_settled(&fixture);
return verdict;
}
fn service_call_count(syscall: &'static str, follow: bool) -> usize {
let fixture: Fixture = fixture("count-service");
let trace: PathBuf = fixture.root.join("service-trace.txt");
*fixture.service_launch.lock().expect("the service launch") =
ServiceLaunch { prefix: tracer(trace.as_path(), syscall, None, follow), ..ServiceLaunch::default() };
let repo: PathBuf = staged_repository(&fixture, "repo");
let output: Output = run_watched(&fixture, &mut unbounded_wrapper(&fixture, repo.as_path(), &COMMIT_ARGUMENTS));
assert_eq!(output.status.code(), Some(0), "the counting commit lands: {}", String::from_utf8_lossy(output.stderr.as_slice()));
end_the_pressure(&fixture);
let traced: Result<String, std::io::Error> = std::fs::read_to_string(&trace);
let count: usize = busiest_count(traced.as_deref().unwrap_or_default(), syscall);
assert!(
count > 0 || syscall != "write",
"the counting commit's trace holds no write at all: {:?}",
traced.as_ref().map(|text| { return text.chars().take(400).collect::<String>(); })
);
remove_settled(&fixture);
return count;
}
fn sweep_service(label: &str, calls: &[&'static str], effect: Effect, when: fn(usize) -> When, follow: bool, recover: Recovery) {
if !strace_available() {
eprintln!("skipped: strace is not installed, so no call was faulted");
return;
}
let counts: Vec<(&'static str, usize)> = calls.iter().map(|syscall| { return (*syscall, service_call_count(syscall, follow)); }).collect();
let points: Vec<(&'static str, usize)> = sweep_points(counts.as_slice(), POSITIONS_PER_SWEEP);
assert!(!points.is_empty(), "the commit made none of the calls {calls:?}: {counts:?}");
let found: Found = Found { reached: AtomicUsize::new(0), landed: AtomicUsize::new(0) };
let failures: Vec<String> = in_batches(points.as_slice(), &|(syscall, position): &(&'static str, usize)| {
let fault: Injection = Injection { syscall, effect, when: when(*position), follow_forks: follow };
return service_case(format!("{label}-{syscall}-{position}").as_str(), &fault, &found, recover);
});
assert!(failures.is_empty(), "{} of {} faults left the repository unrecovered (call counts {counts:?}):\n{}", failures.len(), points.len(), failures.join("\n"));
assert!(found.reached.load(Ordering::SeqCst) > 0, "no run reached its fault, so the sweep tested nothing (call counts {counts:?})");
}
#[test]
fn a_disk_that_fills_during_a_commit_leaves_a_recoverable_repository() {
sweep_service("full-disk", &STORAGE_CALLS, Effect::Fail("ENOSPC"), When::FromNth, true, recovers_any_way);
}
#[test]
fn storage_that_fails_during_a_commit_leaves_a_recoverable_repository() {
sweep_service("io-error", &STORAGE_CALLS, Effect::Fail("EIO"), When::FromNth, true, recovers_any_way);
}
#[test]
fn a_service_killed_at_any_call_leaves_a_recoverable_repository() {
sweep_service("service-kill", &KILL_CALLS, Effect::Kill, When::Nth, false, recovers_after_service_death);
}
#[test]
fn slow_flushes_delay_a_commit_and_change_nothing_else() {
if !strace_available() {
eprintln!("skipped: strace is not installed, so no call was delayed");
return;
}
let fixture: Fixture = fixture("slow-flush");
let trace: PathBuf = fixture.root.join("service-trace.txt");
let fault: Injection = Injection { syscall: "fsync,fdatasync", effect: Effect::SlowReturn(SLOW_FLUSH_MICROSECONDS), when: When::FromNth(1), follow_forks: true };
*fixture.service_launch.lock().expect("the service launch") =
ServiceLaunch { prefix: tracer(trace.as_path(), fault.syscall, Some(&fault), true), ..ServiceLaunch::default() };
let repo: PathBuf = staged_repository(&fixture, "repo");
let output: Output = run_watched(&fixture, &mut unbounded_wrapper(&fixture, repo.as_path(), &COMMIT_ARGUMENTS));
let said: String = String::from_utf8_lossy(&[output.stdout.as_slice(), output.stderr.as_slice()].concat()).into_owned();
assert_eq!(output.status.code(), Some(0), "a slow flush only delays the commit: {said}");
end_the_pressure(&fixture);
let traced: String = std::fs::read_to_string(&trace).unwrap_or_default();
assert!(took_effect(traced.as_str()), "no flush was delayed, so nothing was tested: {traced}");
assert_eq!(verify_recovery_with(&fixture, repo.as_path(), "slow-flush", said.as_str(), run_unsettled), Ok(()));
remove_settled(&fixture);
}
fn traced_wrapper(fixture: &Fixture, directory: &Path, arguments: &[&str], trace: &Path, syscall: &str, fault: Option<&Injection>) -> Command {
let words: Vec<OsString> = tracer(trace, syscall, fault, false);
let mut command: Command = Command::new(&words[0]);
command.args(&words[1..]).arg(fixture.root.join("bin/git")).args(arguments);
hermetic(&mut command, fixture, wrapper_first_path(fixture).as_os_str());
if let Some(runtime) = service_runtime(fixture) {
command.env("XDG_RUNTIME_DIR", runtime);
}
command.current_dir(directory);
return command;
}
struct Flow {
name: &'static str,
setup: fn(&Fixture) -> PathBuf,
arguments: &'static [&'static str],
recovered: fn(&Fixture, &Path, &Output) -> Result<(), String>,
}
fn flow_call_count(flow: &Flow, syscall: &'static str) -> usize {
let fixture: Fixture = fixture(format!("count-{}", flow.name).as_str());
let repo: PathBuf = (flow.setup)(&fixture);
let trace: PathBuf = fixture.root.join("wrapper-trace.txt");
let output: Output = run_watched(&fixture, &mut traced_wrapper(&fixture, repo.as_path(), flow.arguments, trace.as_path(), syscall, None));
assert_eq!(output.status.code(), Some(0), "the counting run of {} ends well: {}", flow.name, String::from_utf8_lossy(output.stderr.as_slice()));
let count: usize = busiest_count(std::fs::read_to_string(&trace).unwrap_or_default().as_str(), syscall);
remove_settled(&fixture);
return count;
}
fn sweep_wrapper_kill(flow: &Flow) {
if !strace_available() {
eprintln!("skipped: strace is not installed, so no process was killed");
return;
}
let counts: Vec<(&'static str, usize)> = WRAPPER_KILL_CALLS.iter().map(|syscall| { return (*syscall, flow_call_count(flow, syscall)); }).collect();
let points: Vec<(&'static str, usize)> = sweep_points(counts.as_slice(), POSITIONS_PER_SWEEP);
assert!(!points.is_empty(), "the wrapper made none of the calls {WRAPPER_KILL_CALLS:?} in {}: {counts:?}", flow.name);
let killed: AtomicUsize = AtomicUsize::new(0);
let failures: Vec<String> = in_batches(points.as_slice(), &|(syscall, position): &(&'static str, usize)| {
let label: String = format!("kill-{}-{syscall}-{position}", flow.name);
let fixture: Fixture = fixture(label.as_str());
let repo: PathBuf = (flow.setup)(&fixture);
let trace: PathBuf = fixture.root.join("wrapper-trace.txt");
let fault: Injection = Injection { syscall, effect: Effect::Kill, when: When::Nth(*position), follow_forks: false };
let mut command: Command = traced_wrapper(&fixture, repo.as_path(), flow.arguments, trace.as_path(), syscall, Some(&fault));
let output: Output = spawn_watched(&fixture, &mut command).finish_after_stragglers();
if took_effect(std::fs::read_to_string(&trace).unwrap_or_default().as_str()) {
killed.fetch_add(1, Ordering::SeqCst);
}
let verdict: Result<(), String> = (flow.recovered)(&fixture, repo.as_path(), &output).map_err(|message| { return format!("{label}: {message}"); });
remove_settled(&fixture);
return verdict;
});
assert!(failures.is_empty(), "{} of {} kills left the repository unrecovered (call counts {counts:?}):\n{}", failures.len(), points.len(), failures.join("\n"));
assert!(killed.load(Ordering::SeqCst) > 0, "no run was killed, so the sweep tested nothing (call counts {counts:?})");
}
fn merge_setup(fixture: &Fixture) -> PathBuf {
let repo: PathBuf = super::merge_isolation::diverged(fixture, "repo");
super::merge_isolation::stage_unrelated_work(fixture, repo.as_path());
return repo;
}
fn merge_recovered(fixture: &Fixture, repo: &Path, killed: &Output) -> Result<(), String> {
let status = run_unsettled(fixture, repo, &["status", "--porcelain=v1"]);
if status.code != Some(0) {
return Err(format!("the next command must succeed, exit {:?}: {} (the killed run: {:?})", status.code, String::from_utf8_lossy(status.stderr.as_slice()), killed.status));
}
let merges: String = super::commit::git_text(fixture, repo, &["rev-list", "--count", "--merges", "HEAD"]);
let shown: String = String::from_utf8_lossy(status.stdout.as_slice()).into_owned();
let staged_lines: Vec<&str> = shown.lines().collect();
let other_work_intact: bool = staged_lines.len() == 3
&& staged_lines.iter().any(|line| { return line.starts_with("M u.txt"); })
&& staged_lines.iter().any(|line| { return line.starts_with("A new.txt"); })
&& staged_lines.iter().any(|line| { return line.starts_with("D v.txt"); });
if merges.trim() == "1" && !other_work_intact {
return Err(format!("the merge landed but the other session's staged work is not what it was: {shown:?}"));
}
if merges.trim() == "0" && shown.lines().count() < 3 {
return Err(format!("no merge landed but staged work is missing: {shown:?}"));
}
if repo.join(".git/index.lock").exists() {
return Err(String::from("the index lock stays"));
}
let records: Vec<PathBuf> = super::merge_isolation::records_in(repo.join(".git/cli-git-merges").as_path())
.into_iter()
.filter(|path| { return path.extension().is_none_or(|extension| { return extension != "partial"; }); })
.collect();
if !records.is_empty() {
return Err(format!("the merge's records stay: {records:?}"));
}
let checked = Command::new("/usr/bin/git")
.current_dir(repo)
.args(["fsck", "--strict", "--no-dangling"])
.env_clear()
.env("GIT_CONFIG_NOSYSTEM", "1")
.env("GIT_CONFIG_GLOBAL", "/nonexistent-global-config")
.output()
.expect("git fsck starts");
if !checked.status.success() || !checked.stderr.is_empty() {
return Err(format!("git fsck: {}", String::from_utf8_lossy(&checked.stderr)));
}
let again = run_unsettled(fixture, repo, &["merge", "--no-edit", "feature"]);
if again.code != Some(0) {
return Err(format!("the merge asked for again must end well, exit {:?}: {}", again.code, String::from_utf8_lossy(again.stderr.as_slice())));
}
return Ok(());
}
#[test]
fn a_merge_killed_at_any_call_leaves_a_recoverable_checkout() {
sweep_wrapper_kill(&Flow { name: "merge", setup: merge_setup, arguments: &["merge", "--no-edit", "feature"], recovered: merge_recovered });
}
fn copy_setup(fixture: &Fixture) -> PathBuf {
return super::worktree_copy::repository_with_rules(fixture, true);
}
fn copy_recovered(fixture: &Fixture, repo: &Path, killed: &Output) -> Result<(), String> {
let status = run_unsettled(fixture, repo, &["status", "--porcelain=v1"]);
if status.code != Some(0) {
return Err(format!("the next command must succeed, exit {:?}: {} (the killed run: {:?})", status.code, String::from_utf8_lossy(status.stderr.as_slice()), killed.status));
}
let worktree: PathBuf = fixture.root.join("one");
if worktree.join(".git").exists() {
let inside = run_unsettled(fixture, worktree.as_path(), &["rev-parse", "--is-inside-work-tree"]);
if inside.code != Some(0) {
return Err(format!("the worktree Git made must answer, exit {:?}: {}", inside.code, String::from_utf8_lossy(inside.stderr.as_slice())));
}
} else {
let _pruned = run_unsettled(fixture, repo, &["worktree", "prune"]);
let _removed = std::fs::remove_dir_all(&worktree);
let again = run_unsettled(fixture, repo, &["worktree", "add", "--quiet", "-b", "one-again", "../one"]);
if again.code != Some(0) {
let words: String = String::from_utf8_lossy(again.stderr.as_slice()).into_owned();
if !words.contains("already exists") {
return Err(format!("the worktree add asked for again must end well, exit {:?}: {words}", again.code));
}
}
}
let staging: Vec<String> = std::fs::read_dir(&fixture.root)
.map(|entries| { return entries.flatten().map(|entry| { return entry.file_name().to_string_lossy().into_owned(); }).filter(|name| { return name.starts_with(".cli-git-worktree-copy-"); }).collect(); })
.unwrap_or_default();
if !staging.is_empty() {
return Err(format!("a staging directory stays after the next commands: {staging:?}"));
}
return Ok(());
}
#[test]
fn a_worktree_copy_killed_at_any_call_leaves_a_recoverable_repository() {
sweep_wrapper_kill(&Flow {
name: "copy",
setup: copy_setup,
arguments: &["worktree", "add", "--quiet", "-b", "one", "../one"],
recovered: copy_recovered,
});
}