use super::commit::{git_text, remove_settled, wrapped_commit};
use super::service_commit::runtime_for;
use super::service_landing_lock::foreign_commit_until_released;
use super::service_landing_selection::base_repository;
use super::strace::{strace_available, strace_program};
use super::support::{Fixture, fixture, observe};
use super::wait_notice::{Watched, watch_notices};
use git_policy_cli::commit_timing::TIMING_FILE;
use git_policy_cli::service_client::ask_to_retire;
use git_policy_cli::service_instance::{instance_of, socket_for};
use git_policy_cli::service_protocol::RELEASE;
use std::os::unix::process::ExitStatusExt;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};
const SERVICE: &str = env!("CARGO_BIN_EXE_cli-git-service");
const PATIENCE: Duration = Duration::from_secs(60);
const SAFETY_IDLE_SECONDS: &str = "120";
const IDLE_SECONDS: u64 = 1;
const HOLD_SECONDS: u32 = 4;
const MESSAGE: &str = "timing control message that no log may repeat";
const OTHER_RELEASE: &str = "0.0.0-timing-control";
struct Held(Child);
impl Drop for Held {
fn drop(&mut self) {
let _killed = self.0.kill();
let _waited = self.0.wait();
}
}
struct Places {
socket: PathBuf,
log: PathBuf,
}
fn spawn_service(root: &Path, places: &Places, idle_seconds: &str, extra: &[&str]) -> Held {
return spawn_service_with_environment(root, places, idle_seconds, extra, &[]);
}
fn spawn_service_with_environment(root: &Path, places: &Places, idle_seconds: &str, extra: &[&str], environment: &[(&str, String)]) -> Held {
let log = std::fs::OpenOptions::new().create(true).append(true).open(&places.log).expect("service log");
return spawn_service_with_stderr(root, places, idle_seconds, extra, environment, Stdio::from(log));
}
fn spawn_service_with_stderr(root: &Path, places: &Places, idle_seconds: &str, extra: &[&str], environment: &[(&str, String)], stderr: Stdio) -> Held {
std::fs::create_dir_all(places.socket.parent().expect("parent")).expect("socket directory");
let child: Child = Command::new(SERVICE)
.args(["--journal"])
.arg(root.join("journal"))
.arg("--path")
.arg(&places.socket)
.args(["--idle-seconds", idle_seconds])
.args(extra)
.env_clear()
.env("PATH", "/usr/bin:/bin")
.env("HOME", root.join("home"))
.env("GIT_CONFIG_NOSYSTEM", "1")
.env("GIT_CONFIG_GLOBAL", "/nonexistent-global-config")
.env("CLI_GIT_LOCK_WAIT_SECONDS", "60")
.envs(environment.iter().map(|(name, value)| { return (*name, value.as_str()); }))
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(stderr)
.spawn()
.expect("the service starts");
let held: Held = Held(child);
let began: Instant = Instant::now();
while std::os::unix::net::UnixStream::connect(&places.socket).is_err() {
assert!(began.elapsed() < PATIENCE, "the service never listened");
std::thread::sleep(Duration::from_millis(10));
}
return held;
}
fn places_for(fixture: &Fixture, name: &str, repo: &Path) -> Places {
let runtime: PathBuf = runtime_for(name);
return Places { socket: socket_for(runtime.as_path(), repo.join(".git").as_path()), log: fixture.root.join(format!("{name}.log")) };
}
fn instance_argument(repo: &Path) -> String {
return instance_of(repo.join(".git").as_path());
}
fn exit_of(service: &mut Held, limit: Duration) -> std::process::ExitStatus {
let began: Instant = Instant::now();
loop {
if let Ok(Some(status)) = service.0.try_wait() {
return status;
}
assert!(began.elapsed() < limit, "the service did not exit within {limit:?}");
std::thread::sleep(Duration::from_millis(10));
}
}
fn lines_of(log: &Path, word: &str) -> Vec<String> {
let text: String = std::fs::read_to_string(log).expect("service log");
let prefix: String = format!("{}{word}", git_policy_cli::service_timing::LOG_PREFIX);
return text.lines().filter(|line| { return line.starts_with(prefix.as_str()); }).map(str::to_string).collect();
}
fn ms_of(line: &str, key: &str) -> u64 {
let marker: String = format!(" {key}=");
let from: usize = line.find(marker.as_str()).unwrap_or_else(|| { panic!("no {key} in {line}") }) + marker.len();
let digits: String = line[from..].chars().take_while(char::is_ascii_digit).collect();
return digits.parse::<u64>().unwrap_or_else(|_| { panic!("no number for {key} in {line}") });
}
fn records(repo: &Path) -> Vec<serde_json::Value> {
let text: String = std::fs::read_to_string(repo.join(".git").join(TIMING_FILE)).expect("the timing file");
return text.lines().map(|line| { return serde_json::from_str(line).expect("a JSON line"); }).collect();
}
fn number(record: &serde_json::Value, name: &str) -> u64 {
return record[name].as_u64().unwrap_or_else(|| { panic!("no number {name} in {record}") });
}
fn commit_a_command(fixture: &Fixture, repo: &Path, runtime: &Path) -> Command {
static ROUND: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
let round: usize = ROUND.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
std::fs::write(repo.join("a.txt"), format!("edit {round} by {}\n", std::process::id())).expect("a.txt");
let mut command: Command = wrapped_commit(fixture, repo, &["commit", "-m", MESSAGE, "--", "a.txt"]);
command.env("XDG_RUNTIME_DIR", runtime);
return command;
}
fn commit_a(fixture: &Fixture, repo: &Path, runtime: &Path) -> super::support::Observed {
return observe(&mut commit_a_command(fixture, repo, runtime), b"");
}
const DISJOINT: [&str; 8] = ["prepareMs", "buildMs", "provisionMs", "submitMs", "afterLandingMs", "gateMs", "handoffMs", "otherMs"];
#[test]
fn a_wrapped_commit_leaves_one_record_and_the_service_one_line_for_it() {
let fixture: Fixture = fixture("timing-record");
let repo: PathBuf = base_repository(&fixture);
let name: String = format!("timing-record-{}", std::process::id());
let places: Places = places_for(&fixture, name.as_str(), &repo);
let instance: String = instance_argument(&repo);
let _service: Held = spawn_service(fixture.root.as_path(), &places, SAFETY_IDLE_SECONDS, &["--instance", instance.as_str()]);
let identity: String = git_text(&fixture, &repo, &["var", "GIT_COMMITTER_IDENT"]);
let address: &str = identity.split_once('<').and_then(|(_, rest)| { return rest.split_once('>'); }).map(|(inside, _)| { return inside; }).expect("an address");
let branch: String = git_text(&fixture, &repo, &["symbolic-ref", "--short", "HEAD"]).trim().to_string();
let served = commit_a(&fixture, &repo, runtime_for(name.as_str()).as_path());
assert_eq!(served.code, Some(0), "{}", String::from_utf8_lossy(&served.stderr));
for (stream, bytes) in [("stdout", &served.stdout), ("stderr", &served.stderr)] {
let text: String = String::from_utf8_lossy(bytes).into_owned();
assert!(!text.contains(TIMING_FILE) && !text.contains("totalMs") && !text.contains("_ms="), "timing reached the terminal on {stream}: {text}");
}
let found: Vec<serde_json::Value> = records(&repo);
assert_eq!(found.len(), 1, "one record for one commit: {found:?}");
let record: &serde_json::Value = &found[0];
println!("TIMING RECORD: {record}");
assert_eq!(record["outcome"], "landed");
assert_eq!(record["exitCode"], 0);
assert_eq!(record["branch"], format!("refs/heads/{branch}"));
assert_eq!(record["release"], RELEASE);
assert_eq!(record["instance"], instance.as_str());
assert_eq!(record["answeredByRelease"], RELEASE);
assert_eq!(record["attempts"], 1);
assert_eq!(record["sends"], 1);
assert!(number(record, "connectTries") >= 1, "{record}");
let phases: u64 = DISJOINT.iter().map(|field| { return number(record, field); }).sum();
let slack: u64 = u64::try_from(DISJOINT.len()).expect("a small count");
assert!(phases <= number(record, "totalMs") && number(record, "totalMs") - phases < slack, "the phases and the remainder add up to the total: {record}");
assert!(number(record, "reachMs") + number(record, "answerWaitMs") <= number(record, "submitMs"), "{record}");
let started: Vec<String> = lines_of(&places.log, "started");
println!("SERVICE START: {}", started.join("\n"));
assert_eq!(started.len(), 1, "{started:?}");
assert!(started[0].contains(format!("release={RELEASE:?}").as_str()) && started[0].contains(format!("instance={instance}").as_str()), "{}", started[0]);
assert!(started[0].contains(format!("pid={}", _service.0.id()).as_str()), "{}", started[0]);
assert!(std::fs::read_to_string(&places.log).expect("log").contains("ready"), "the ready line stays");
let answered: Vec<String> = lines_of(&places.log, "answered");
println!("SERVICE ANSWER: {}", answered.join("\n"));
assert_eq!(answered.len(), 1, "one line for one request: {answered:?}");
let line: &str = answered[0].as_str();
assert!(line.contains("outcome=landed") && line.contains(format!("branch=\"refs/heads/{branch}\"").as_str()), "{line}");
assert!(line.contains(" size=1 position=1 "), "{line}");
let request_phases: u64 = ["read_ms", "queued_ms", "settle_ms", "landing_ms", "answer_ms"].iter().map(|key| { return ms_of(line, key); }).sum();
assert!(request_phases <= ms_of(line, "total_ms"), "the phases lie inside the request's total: {line}");
let landing_keys: [&str; 6] = ["lock_wait_ms", "reads_ms", "objects_ms", "journal_ms", "ref_ms", "sync_ms"];
let landing_phases: u64 = landing_keys.iter().map(|key| { return ms_of(line, key); }).sum();
let landing_total: u64 = ms_of(line, "landing_ms");
assert!(landing_phases <= landing_total && landing_total - landing_phases < u64::try_from(landing_keys.len()).expect("a small count"), "{line}");
assert!(!line.contains("lock_holder"), "nobody held the index: {line}");
let everything: String = std::fs::read_to_string(&places.log).expect("log") + &std::fs::read_to_string(repo.join(".git").join(TIMING_FILE)).expect("record");
assert!(!everything.contains(MESSAGE), "the commit's text is in a log");
assert!(!everything.contains(address), "the committer's address is in a log");
remove_settled(&fixture);
}
const ROUNDING_MS: u64 = 12;
fn wait_covers_the_hold(wait_ms: u64, age_ms: u64, hold_ms: u64) -> bool {
return wait_ms + age_ms + ROUNDING_MS >= hold_ms;
}
fn phase_lies_beside_the_wait(phase_ms: u64, wait_ms: u64, whole_ms: u64) -> bool {
return phase_ms + wait_ms <= whole_ms + ROUNDING_MS;
}
#[test]
fn a_slow_phase_and_a_late_wait_do_not_break_the_attribution() {
let hold: u64 = 4_000;
let wait: u64 = 2_100;
let age: u64 = 1_900;
assert!(wait_covers_the_hold(wait, age, hold), "the wait began {age} ms into a hold of {hold} ms");
assert!(phase_lies_beside_the_wait(2_681, wait, 2_681 + wait + 50), "a slow phase is not the wait");
}
#[test]
fn a_wait_that_lost_time_or_a_phase_that_overlaps_it_is_caught() {
assert!(!wait_covers_the_hold(500, 300, 4_000), "3.2 s of the hold went to no wait");
assert!(!phase_lies_beside_the_wait(3_000, 2_000, 4_000), "3 s and 2 s do not fit in 4 s");
}
const STALL_HOLDS: u64 = 2;
#[test]
fn a_commit_that_waits_for_a_foreign_index_holder_is_attributed_to_that_wait() {
attributes_the_wait_for_a_foreign_index_holder("timing-lock", Duration::ZERO);
}
#[test]
fn a_commit_that_reaches_a_foreign_index_holder_late_is_attributed_to_that_wait() {
attributes_the_wait_for_a_foreign_index_holder("timing-lock-late", Duration::from_secs(STALL_HOLDS * u64::from(HOLD_SECONDS)));
}
fn attributes_the_wait_for_a_foreign_index_holder(control: &str, stall: Duration) {
let fixture: Fixture = fixture(control);
let repo: PathBuf = base_repository(&fixture);
let name: String = format!("{control}-{}", std::process::id());
let places: Places = places_for(&fixture, name.as_str(), &repo);
let instance: String = instance_argument(&repo);
let _service: Held = spawn_service(fixture.root.as_path(), &places, SAFETY_IDLE_SECONDS, &["--instance", instance.as_str()]);
let (holder, release): (Child, PathBuf) = foreign_commit_until_released(&fixture, &repo, HOLD_SECONDS);
std::thread::sleep(stall);
let waiting_for_holder: String = format!("waiting for PID {}", holder.id());
let served: Watched = watch_notices(
&mut commit_a_command(&fixture, &repo, runtime_for(name.as_str()).as_path()),
waiting_for_holder.as_str(),
1,
|| { std::fs::write(&release, b"").expect("the foreign commit is told to let go"); },
);
let held = holder.wait_with_output().expect("the foreign commit ends");
assert!(held.status.success(), "{}", String::from_utf8_lossy(&held.stderr));
assert_eq!(served.code, Some(0), "{}", served.stderr);
let answered: Vec<String> = lines_of(&places.log, "answered");
assert_eq!(answered.len(), 1, "{answered:?}");
let line: &str = answered[0].as_str();
println!("SERVICE ANSWER (held index): {line}");
let lock_wait: u64 = ms_of(line, "lock_wait_ms");
let hold_ms: u64 = u64::from(HOLD_SECONDS) * 1_000;
assert!(line.contains("lock_holder=\"Process "), "the holder is named: {line}");
let found: Vec<serde_json::Value> = records(&repo);
let record: &serde_json::Value = found.last().expect("the wrapped commit's record");
println!("TIMING RECORD (held index): {record}");
let age_at_first_look: u64 = ms_of(line, "age_ms");
assert!(wait_covers_the_hold(lock_wait, age_at_first_look, hold_ms), "the wait for the lock is what was left of the hold ({hold_ms} ms) after {age_at_first_look} ms: {line}");
let landing_total: u64 = ms_of(line, "landing_ms");
for other in ["reads_ms", "objects_ms", "journal_ms", "ref_ms", "sync_ms"] {
assert!(phase_lies_beside_the_wait(ms_of(line, other), lock_wait, landing_total), "{other} is not what the time went to: {line}");
}
for other in ["read_ms", "queued_ms", "settle_ms", "answer_ms"] {
assert!(phase_lies_beside_the_wait(ms_of(line, other), lock_wait, ms_of(line, "total_ms")), "{other} is not what the time went to: {line}");
}
assert!(number(record, "answerWaitMs") >= lock_wait, "the wrapper waited at least as long as the service did for the lock: {record}");
for other in ["prepareMs", "buildMs", "provisionMs", "afterLandingMs", "gateMs", "handoffMs", "reachMs"] {
assert!(phase_lies_beside_the_wait(number(record, other), number(record, "answerWaitMs"), number(record, "totalMs")), "{other} is not what the time went to: {record}");
}
assert!(number(record, "waitingNotices") >= 1, "the service told the wrapper it was waiting: {record}");
assert_eq!(record["lockHolderKind"], "proven-alive", "the record says what kind of holder the wait was for: {record}");
let waited: Vec<String> = lines_of(&places.log, "waited for the index lock");
assert_eq!(waited.len(), 1, "{waited:?}");
println!("SERVICE WAIT (held index): {}", waited[0]);
assert!(waited[0].contains(" result=taken ") && ms_of(waited[0].as_str(), "lock_looks") >= 1, "{}", waited[0]);
assert!(waited[0].contains("lock_found=\"foreign no-marker"), "the holder is described: {}", waited[0]);
assert!(wait_covers_the_hold(ms_of(waited[0].as_str(), "lock_place_ms"), ms_of(waited[0].as_str(), "age_ms"), hold_ms), "the wait for the holder is timed apart from the flush: {}", waited[0]);
assert!(line.contains("lock_found=") && line.contains(" lock_stage_ms="), "the answer carries it too: {line}");
remove_settled(&fixture);
}
#[test]
fn every_service_line_states_the_time_of_its_event() {
let fixture: Fixture = fixture("timing-stamps");
let repo: PathBuf = base_repository(&fixture);
let name: String = format!("timing-stamps-{}", std::process::id());
let places: Places = places_for(&fixture, name.as_str(), &repo);
let instance: String = instance_argument(&repo);
let before: u128 = git_policy_cli::service_timing::wall_millis();
let _service: Held = spawn_service(fixture.root.as_path(), &places, SAFETY_IDLE_SECONDS, &["--instance", instance.as_str()]);
let served = commit_a(&fixture, &repo, runtime_for(name.as_str()).as_path());
assert_eq!(served.code, Some(0), "{}", String::from_utf8_lossy(&served.stderr));
let after: u128 = git_policy_cli::service_timing::wall_millis();
let text: String = std::fs::read_to_string(&places.log).expect("service log");
let lines: Vec<&str> = text.lines().filter(|line| { return line.starts_with(git_policy_cli::service_timing::LOG_PREFIX); }).collect();
assert!(lines.len() >= 3, "started, ready, answered: {text}");
for line in lines {
let stamp: u128 = line.rsplit_once(" at_ms=").unwrap_or_else(|| { panic!("no time at the end of {line}") }).1.parse::<u128>().unwrap_or_else(|_| { panic!("no number in {line}") });
assert!(stamp >= before && stamp <= after, "{stamp} is not between {before} and {after}: {line}");
}
remove_settled(&fixture);
}
#[test]
fn a_commit_that_meets_another_release_records_the_update_and_both_services_log_it() {
let fixture: Fixture = fixture("timing-update");
let repo: PathBuf = base_repository(&fixture);
let name: String = format!("timing-update-{}", std::process::id());
let places: Places = places_for(&fixture, name.as_str(), &repo);
let current_log: PathBuf = fixture.root.join("current.log");
let instance: String = instance_argument(&repo);
let retiring: Held = spawn_service(fixture.root.as_path(), &places, SAFETY_IDLE_SECONDS, &["--instance", instance.as_str(), "--release", OTHER_RELEASE]);
let successor = {
let socket: PathBuf = places.socket.clone();
let root: PathBuf = fixture.root.clone();
let argument: String = instance.clone();
let log: PathBuf = current_log.clone();
std::thread::spawn(move || {
let mut old: Held = retiring;
let status = exit_of(&mut old, PATIENCE);
let next: Held = spawn_service(root.as_path(), &Places { socket, log }, SAFETY_IDLE_SECONDS, &["--instance", argument.as_str()]);
return (status, next);
})
};
let served = commit_a(&fixture, &repo, runtime_for(name.as_str()).as_path());
assert_eq!(served.code, Some(0), "{}", String::from_utf8_lossy(&served.stderr));
let (old_status, _current): (std::process::ExitStatus, Held) = successor.join().expect("the successor thread");
assert!(old_status.success(), "the old service exited by itself: {old_status:?}");
let found: Vec<serde_json::Value> = records(&repo);
let record: &serde_json::Value = &found[0];
println!("TIMING RECORD (other release): {record}");
assert!(number(record, "updateRounds") >= 1, "{record}");
assert_eq!(record["releasesSeen"], serde_json::Value::from(vec![OTHER_RELEASE, RELEASE]));
assert_eq!(record["answeredByRelease"], RELEASE);
assert!(number(record, "connectTries") > number(record, "updateRounds"), "{record}");
assert!(number(record, "startWaitMs") <= number(record, "reachMs"), "{record}");
let retired: Vec<String> = lines_of(&places.log, "retire requested");
println!("OLD SERVICE: {}", std::fs::read_to_string(&places.log).expect("log"));
let rounds: usize = usize::try_from(number(record, "updateRounds")).expect("a small count");
assert!(!retired.is_empty() && retired.len() <= rounds, "{retired:?} against {rounds} rounds");
assert!(retired.iter().all(|line| { return line.contains(format!("{OTHER_RELEASE:?}").as_str()); }), "{retired:?}");
let exits: Vec<String> = lines_of(&places.log, "exiting");
assert_eq!(exits.len(), 1, "{exits:?}");
assert!(exits[0].contains("reason=retire"), "{}", exits[0]);
assert_eq!(lines_of(¤t_log, "answered").len(), 1, "the current release landed it");
assert_eq!(lines_of(&places.log, "answered").len(), 0, "the old release landed nothing");
remove_settled(&fixture);
}
fn traced_for_files(inner: &Command, trace: &Path) -> Command {
let mut outer: Command = Command::new(strace_program().expect("strace is installed"));
outer
.arg("--follow-forks")
.arg("--quiet")
.arg("--quiet")
.arg("--string-limit=4096")
.arg("--no-abbrev")
.arg(format!("--output={}", trace.display()))
.arg("--trace=openat,open,creat,link,linkat,rename,renameat,renameat2,unlink,unlinkat")
.arg(inner.get_program())
.args(inner.get_args());
outer.env_clear();
for (name, setting) in inner.get_envs() {
if let Some(text) = setting {
outer.env(name, text);
}
}
if let Some(directory) = inner.get_current_dir() {
outer.current_dir(directory);
}
return outer;
}
#[test]
fn a_client_that_updates_the_service_never_takes_the_real_index_lock_and_the_new_service_finds_it_free() {
if !strace_available() {
eprintln!("skipped: strace is not installed, so no file call was traced");
return;
}
let fixture: Fixture = fixture("timing-upgrade-lock");
let repo: PathBuf = base_repository(&fixture);
let name: String = format!("timing-upgrade-lock-{}", std::process::id());
let places: Places = places_for(&fixture, name.as_str(), &repo);
let current_log: PathBuf = fixture.root.join("current.log");
let instance: String = instance_argument(&repo);
let retiring: Held = spawn_service(fixture.root.as_path(), &places, SAFETY_IDLE_SECONDS, &["--instance", instance.as_str(), "--release", OTHER_RELEASE]);
let successor = {
let socket: PathBuf = places.socket.clone();
let root: PathBuf = fixture.root.clone();
let argument: String = instance.clone();
let log: PathBuf = current_log.clone();
std::thread::spawn(move || {
let mut old: Held = retiring;
let status = exit_of(&mut old, PATIENCE);
let next: Held = spawn_service(root.as_path(), &Places { socket, log }, SAFETY_IDLE_SECONDS, &["--instance", argument.as_str()]);
return (status, next);
})
};
std::fs::write(repo.join("a.txt"), format!("edit by {}\n", std::process::id())).expect("a.txt");
let mut inner: Command = wrapped_commit(&fixture, &repo, &["commit", "-m", MESSAGE, "--", "a.txt"]);
inner.env("XDG_RUNTIME_DIR", runtime_for(name.as_str()));
let trace: PathBuf = fixture.root.join("client.trace");
let served = observe(&mut traced_for_files(&inner, trace.as_path()), b"");
assert_eq!(served.code, Some(0), "{}", String::from_utf8_lossy(&served.stderr));
let (_old_status, _current): (std::process::ExitStatus, Held) = successor.join().expect("the successor thread");
let text: String = std::fs::read_to_string(&trace).expect("the trace");
let real_lock: String = format!("{}/.git/index.lock\"", repo.display());
let touching: Vec<&str> = text.lines().filter(|line| { return line.contains(real_lock.as_str()); }).collect();
let control_trace: PathBuf = fixture.root.join("control.trace");
let mut taker: Command = Command::new("/bin/sh");
taker.arg("-c").arg("echo held > \"$0/.git/index.lock\" && rm \"$0/.git/index.lock\"").arg(&repo);
let control = traced_for_files(&taker, control_trace.as_path()).output().expect("the control runs");
assert!(control.status.success(), "{}", String::from_utf8_lossy(&control.stderr));
let control_text: String = std::fs::read_to_string(&control_trace).expect("the control's trace");
assert!(control_text.lines().any(|line| { return line.contains(real_lock.as_str()); }), "the filter sees a program that takes the lock: {control_text}");
assert!(touching.is_empty(), "the client touched the real index lock: {touching:?}");
let answered: Vec<String> = lines_of(¤t_log, "answered");
assert_eq!(answered.len(), 1, "the current release landed it");
println!("SERVICE ANSWER (upgrade): {}", answered[0]);
assert!(!answered[0].contains(" lock_looks="), "the new service found the lock held: {}", answered[0]);
assert!(lines_of(¤t_log, "waited for the index lock").len() <= 1);
remove_settled(&fixture);
}
fn marker_files(repo: &Path) -> Vec<String> {
let mut names: Vec<String> = std::fs::read_dir(repo.join(".git"))
.expect("the git directory")
.flatten()
.map(|entry| { return entry.file_name().to_string_lossy().into_owned(); })
.filter(|name| { return name.starts_with("index.lock.cli-git-marker"); })
.collect();
names.sort();
return names;
}
#[test]
fn the_second_commit_of_a_service_finds_its_marker_made_ahead_and_nothing_is_left_when_it_retires() {
let fixture: Fixture = fixture("timing-prestage");
let repo: PathBuf = base_repository(&fixture);
let name: String = format!("timing-prestage-{}", std::process::id());
let places: Places = places_for(&fixture, name.as_str(), &repo);
let instance: String = instance_argument(&repo);
let mut service: Held = spawn_service(fixture.root.as_path(), &places, SAFETY_IDLE_SECONDS, &["--instance", instance.as_str()]);
let runtime: PathBuf = runtime_for(name.as_str());
for round in 0..3 {
let served = commit_a(&fixture, &repo, runtime.as_path());
assert_eq!(served.code, Some(0), "round {round}: {}", String::from_utf8_lossy(&served.stderr));
let began: Instant = Instant::now();
while marker_files(&repo).len() != 1 {
assert!(began.elapsed() < Duration::from_secs(10), "round {round}: {:?}", marker_files(&repo));
std::thread::sleep(Duration::from_millis(10));
}
}
let answered: Vec<String> = lines_of(&places.log, "answered");
assert_eq!(answered.len(), 3, "{answered:?}");
println!("SERVICE ANSWERS (prestage): {answered:#?}");
assert!(answered[0].contains(" lock_stage_kind=cold "), "the first commit has nothing made ahead: {}", answered[0]);
assert!(answered[1].contains(" lock_stage_kind=pre ") && answered[2].contains(" lock_stage_kind=pre "), "{answered:?}");
ask_to_retire(places.socket.as_path());
let status = exit_of(&mut service, PATIENCE);
assert!(status.success(), "{status:?}");
assert!(marker_files(&repo).is_empty(), "the marker made ahead is removed when the service ends: {:?}", marker_files(&repo));
remove_settled(&fixture);
}
#[test]
fn a_service_says_it_exits_idle_with_its_uptime() {
let fixture: Fixture = fixture("timing-idle");
let name: String = format!("timing-idle-{}", std::process::id());
let places: Places = Places { socket: runtime_for(name.as_str()).join("cli-git/idle.sock"), log: fixture.root.join("idle.log") };
let mut service: Held = spawn_service(fixture.root.as_path(), &places, IDLE_SECONDS.to_string().as_str(), &[]);
let status = exit_of(&mut service, PATIENCE);
assert!(status.success(), "{status:?}");
let exits: Vec<String> = lines_of(&places.log, "exiting");
println!("SERVICE EXIT (idle): {}", exits.join("\n"));
assert_eq!(exits.len(), 1, "{exits:?}");
assert!(exits[0].contains("reason=idle"), "{}", exits[0]);
assert!(ms_of(exits[0].as_str(), "uptime_ms") >= IDLE_SECONDS * 1_000, "{}", exits[0]);
remove_settled(&fixture);
}
#[test]
fn a_service_says_it_retires_when_asked() {
let fixture: Fixture = fixture("timing-retire");
let name: String = format!("timing-retire-{}", std::process::id());
let places: Places = Places { socket: runtime_for(name.as_str()).join("cli-git/retire.sock"), log: fixture.root.join("retire.log") };
let mut service: Held = spawn_service(fixture.root.as_path(), &places, SAFETY_IDLE_SECONDS, &[]);
ask_to_retire(places.socket.as_path());
let status = exit_of(&mut service, PATIENCE);
assert!(status.success(), "{status:?}");
println!("SERVICE RETIRE: {}", std::fs::read_to_string(&places.log).expect("log"));
assert_eq!(lines_of(&places.log, "retire requested").len(), 1);
let exits: Vec<String> = lines_of(&places.log, "exiting");
assert_eq!(exits.len(), 1, "{exits:?}");
assert!(exits[0].contains("reason=retire"), "{}", exits[0]);
remove_settled(&fixture);
}
#[test]
fn a_service_stopped_by_a_signal_says_so_and_still_dies_of_it() {
let fixture: Fixture = fixture("timing-signal");
let name: String = format!("timing-signal-{}", std::process::id());
let places: Places = Places { socket: runtime_for(name.as_str()).join("cli-git/signal.sock"), log: fixture.root.join("signal.log") };
let mut service: Held = spawn_service(fixture.root.as_path(), &places, SAFETY_IDLE_SECONDS, &[]);
let pid: libc::pid_t = libc::pid_t::try_from(service.0.id()).expect("a process number");
let sent: libc::c_int = unsafe { libc::kill(pid, libc::SIGTERM) };
assert_eq!(sent, 0);
let status = exit_of(&mut service, PATIENCE);
assert_eq!(status.signal(), Some(libc::SIGTERM), "{status:?}");
let exits: Vec<String> = lines_of(&places.log, "exiting");
println!("SERVICE EXIT (signal): {}", exits.join("\n"));
assert_eq!(exits.len(), 1, "{exits:?}");
assert!(exits[0].contains(format!("reason=signal-{}", libc::SIGTERM).as_str()), "{}", exits[0]);
remove_settled(&fixture);
}
fn native_journal(places: &Places, name: &str) -> (std::os::unix::net::UnixDatagram, Vec<(&'static str, String)>) {
use std::os::linux::net::SocketAddrExt;
use std::os::unix::fs::MetadataExt;
let abstract_name: String = format!("cli-git-test-journal-binary-{}-{name}", std::process::id());
let address = std::os::unix::net::SocketAddr::from_abstract_name(abstract_name.as_bytes()).expect("an abstract address");
let receiver = std::os::unix::net::UnixDatagram::bind_addr(&address).expect("bind the journal stand-in");
receiver.set_read_timeout(Some(Duration::from_millis(200))).expect("a read timeout");
std::fs::OpenOptions::new().create(true).append(true).open(&places.log).expect("the log");
let log = std::fs::metadata(&places.log).expect("log metadata");
let environment: Vec<(&'static str, String)> = vec![
(git_policy_cli::journald_native::STREAM_VARIABLE, format!("{}:{}", log.dev(), log.ino())),
(git_policy_cli::journald_native::TEST_SOCKET_VARIABLE, format!("@{abstract_name}")),
];
return (receiver, environment);
}
fn entries_received(receiver: &std::os::unix::net::UnixDatagram, until: impl Fn(&[std::collections::BTreeMap<String, String>]) -> bool) -> Vec<std::collections::BTreeMap<String, String>> {
let began: Instant = Instant::now();
let mut found: Vec<std::collections::BTreeMap<String, String>> = Vec::new();
let mut buffer: Vec<u8> = vec![0; 65_536];
while !until(found.as_slice()) && began.elapsed() < PATIENCE {
if let Ok(length) = receiver.recv(buffer.as_mut_slice()) {
let text: String = String::from_utf8_lossy(&buffer[..length]).into_owned();
found.push(text.lines().filter_map(|line| { return line.split_once('=').map(|(name, value)| { return (name.to_string(), value.to_string()); }); }).collect());
}
}
return found;
}
#[test]
fn a_service_on_the_journal_stream_sends_native_entries_that_join_the_clients_record() {
let fixture: Fixture = fixture("timing-native");
let repo: PathBuf = base_repository(&fixture);
let name: String = format!("timing-native-{}", std::process::id());
let places: Places = places_for(&fixture, name.as_str(), &repo);
let instance: String = instance_argument(&repo);
let (receiver, environment) = native_journal(&places, "send");
let before: u128 = git_policy_cli::service_timing::wall_millis();
let _service: Held = spawn_service_with_environment(fixture.root.as_path(), &places, SAFETY_IDLE_SECONDS, &["--instance", instance.as_str()], environment.as_slice());
let served = commit_a(&fixture, &repo, runtime_for(name.as_str()).as_path());
assert_eq!(served.code, Some(0), "{}", String::from_utf8_lossy(&served.stderr));
let after: u128 = git_policy_cli::service_timing::wall_millis();
let entries = entries_received(&receiver, |found| { return found.iter().any(|entry| { return entry.get("MESSAGE").is_some_and(|message| { return message.starts_with("answered "); }); }); });
let record: serde_json::Value = records(&repo).remove(0);
let answered = entries.iter().find(|entry| { return entry.get("MESSAGE").is_some_and(|message| { return message.starts_with("answered "); }); }).expect("an answered entry");
assert_eq!(Some(answered["CLI_GIT_REQUEST_ID"].as_str()), record["requestId"].as_str(), "the key that joins the client's record to the service's entry");
assert!(record["requestId"].as_str().is_some_and(|id| { return !id.is_empty(); }), "{record}");
assert_eq!(answered["PRIORITY"], "6");
assert_eq!(answered["SYSLOG_IDENTIFIER"], "cli-git-service");
assert_eq!(answered["CLI_GIT_OUTCOME"], "landed");
assert_eq!(answered["CLI_GIT_RELEASE"], RELEASE);
assert_eq!(answered["CLI_GIT_INSTANCE"], instance.as_str());
assert!(answered["MESSAGE"].contains(format!("request=\"{}\"", answered["CLI_GIT_REQUEST_ID"]).as_str()), "{answered:?}");
let stamp: u128 = answered["CLI_GIT_AT_MS"].parse().expect("a number");
assert!(stamp >= before && stamp <= after, "{stamp} is not between {before} and {after}");
let words: Vec<&str> = entries.iter().filter_map(|entry| { return entry.get("MESSAGE").and_then(|message| { return message.split(' ').next(); }); }).collect();
assert!(words.contains(&"started") && words.contains(&"ready"), "the life of the service is in the journal: {words:?}");
assert_eq!(std::fs::read_to_string(&places.log).expect("log"), "", "nothing is also written to standard error");
remove_settled(&fixture);
}
#[test]
fn a_service_stopped_by_a_signal_sends_its_exit_on_the_native_socket() {
let fixture: Fixture = fixture("timing-native-signal");
let name: String = format!("timing-native-signal-{}", std::process::id());
let places: Places = Places { socket: runtime_for(name.as_str()).join("cli-git/native-signal.sock"), log: fixture.root.join("native-signal.log") };
let (receiver, environment) = native_journal(&places, "signal");
let mut service: Held = spawn_service_with_environment(fixture.root.as_path(), &places, SAFETY_IDLE_SECONDS, &[], environment.as_slice());
let pid: libc::pid_t = libc::pid_t::try_from(service.0.id()).expect("a process number");
let before: u128 = git_policy_cli::service_timing::wall_millis();
assert_eq!(unsafe { libc::kill(pid, libc::SIGTERM) }, 0);
let status = exit_of(&mut service, PATIENCE);
assert_eq!(status.signal(), Some(libc::SIGTERM), "{status:?}");
let entries = entries_received(&receiver, |found| { return found.iter().any(|entry| { return entry.get("MESSAGE").is_some_and(|message| { return message.starts_with("exiting "); }); }); });
let exit = entries.iter().find(|entry| { return entry.get("MESSAGE").is_some_and(|message| { return message.starts_with("exiting "); }); }).expect("an exiting entry");
assert!(exit["MESSAGE"].starts_with(format!("exiting reason=signal-{} ", libc::SIGTERM).as_str()), "{exit:?}");
assert_eq!(exit["PRIORITY"], "6");
let stamp: u128 = exit["CLI_GIT_AT_MS"].parse().expect("a number");
assert!(stamp >= before && stamp <= git_policy_cli::service_timing::wall_millis(), "{stamp} against {before}");
assert!(!std::fs::read_to_string(&places.log).expect("log").contains("exiting"), "the exit is in the journal once");
remove_settled(&fixture);
}
#[test]
fn a_journal_that_is_gone_leaves_the_lines_on_standard_error_with_their_priority() {
let fixture: Fixture = fixture("timing-native-gone");
let name: String = format!("timing-native-gone-{}", std::process::id());
let places: Places = Places { socket: runtime_for(name.as_str()).join("cli-git/gone.sock"), log: fixture.root.join("gone.log") };
let (receiver, mut environment) = native_journal(&places, "gone");
drop(receiver);
environment.retain(|(variable, _)| { return *variable != git_policy_cli::journald_native::TEST_SOCKET_VARIABLE; });
environment.push((git_policy_cli::journald_native::TEST_SOCKET_VARIABLE, format!("@cli-git-test-journal-nobody-{}", std::process::id())));
let _service: Held = spawn_service_with_environment(fixture.root.as_path(), &places, SAFETY_IDLE_SECONDS, &[], environment.as_slice());
let text: String = std::fs::read_to_string(&places.log).expect("log");
assert!(text.lines().any(|line| { return line.starts_with("<6>cli-git-service: ready"); }), "{text}");
assert!(text.lines().all(|line| { return line.starts_with("<6>") && line.contains(" at_ms="); }), "{text}");
remove_settled(&fixture);
}
struct FullablePipe {
service_end: std::os::fd::OwnedFd,
control_end: std::os::fd::OwnedFd,
_reader: std::os::fd::OwnedFd,
}
fn fullable_pipe() -> FullablePipe {
use std::os::fd::{FromRawFd, OwnedFd};
let mut ends: [libc::c_int; 2] = [0; 2];
assert_eq!(unsafe { libc::pipe2(ends.as_mut_ptr(), libc::O_CLOEXEC) }, 0);
let (reader, writer) = unsafe { (OwnedFd::from_raw_fd(ends[0]), OwnedFd::from_raw_fd(ends[1])) };
let control_end: OwnedFd = writer.try_clone().expect("a second descriptor of the write end");
return FullablePipe { service_end: writer, control_end, _reader: reader };
}
impl FullablePipe {
fn fill(&self) {
use std::os::fd::AsRawFd;
let page: [u8; 4096] = [b'x'; 4096];
loop {
let mut request: libc::pollfd = libc::pollfd { fd: self.control_end.as_raw_fd(), events: libc::POLLOUT, revents: 0 };
let ready: libc::c_int = unsafe { libc::poll(&raw mut request, 1, 0) };
if ready <= 0 || request.revents & libc::POLLOUT == 0 {
return;
}
assert!(unsafe { libc::write(self.control_end.as_raw_fd(), page.as_ptr().cast(), page.len()) } > 0);
}
}
}
fn stopped_with_a_full_pipe(name: &str, on_stream: bool) {
use std::os::fd::AsRawFd;
let fixture: Fixture = fixture(name);
let unique: String = format!("{name}-{}", std::process::id());
let places: Places = Places { socket: runtime_for(unique.as_str()).join("cli-git/full-pipe.sock"), log: fixture.root.join("unused.log") };
let pipe: FullablePipe = fullable_pipe();
let mut environment: Vec<(&'static str, String)> = Vec::new();
if on_stream {
let mut status: libc::stat = unsafe { std::mem::zeroed() };
assert_eq!(unsafe { libc::fstat(pipe.control_end.as_raw_fd(), &raw mut status) }, 0);
environment.push((git_policy_cli::journald_native::STREAM_VARIABLE, format!("{}:{}", status.st_dev, status.st_ino)));
environment.push((git_policy_cli::journald_native::TEST_SOCKET_VARIABLE, format!("@cli-git-test-journal-nobody-{unique}")));
}
let service_end: std::os::fd::OwnedFd = pipe.service_end.try_clone().expect("a descriptor for the service");
let mut service: Held = spawn_service_with_stderr(fixture.root.as_path(), &places, SAFETY_IDLE_SECONDS, &[], environment.as_slice(), Stdio::from(service_end));
pipe.fill();
let pid: libc::pid_t = libc::pid_t::try_from(service.0.id()).expect("a process number");
assert_eq!(unsafe { libc::kill(pid, libc::SIGTERM) }, 0);
let status = exit_of(&mut service, FULL_PIPE_BOUND);
assert_eq!(status.signal(), Some(libc::SIGTERM), "{status:?}");
remove_settled(&fixture);
}
const FULL_PIPE_BOUND: Duration = Duration::from_secs(10);
#[test]
fn a_service_stopped_while_standard_error_is_a_full_pipe_still_exits() {
stopped_with_a_full_pipe("signal-full-pipe", false);
}
#[test]
fn a_service_on_a_journal_stream_that_is_a_full_pipe_still_exits() {
stopped_with_a_full_pipe("signal-full-stream", true);
}