use super::*;
use crate::test_support::{fixture, remove};
use std::path::PathBuf;
const QUICK: Cadence = Cadence { grace: Duration::from_millis(20), interval: Duration::from_millis(20) };
fn index_in(root: &Path) -> PathBuf {
return root.join("index");
}
fn marked(seq: u64) -> String {
return format!("{LOCK_MARKER}{seq}\n");
}
struct ReleaseOnNotice {
lock: PathBuf,
text: Vec<u8>,
released: bool,
}
impl ReleaseOnNotice {
fn holding(lock: &Path) -> ReleaseOnNotice {
return ReleaseOnNotice { lock: lock.to_path_buf(), text: Vec::new(), released: false };
}
fn written(&self) -> String {
return String::from_utf8_lossy(&self.text).into_owned();
}
}
impl Write for ReleaseOnNotice {
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
self.text.extend_from_slice(bytes);
if !self.released {
self.released = true;
std::fs::remove_file(&self.lock).expect("release");
}
return Ok(bytes.len());
}
fn flush(&mut self) -> std::io::Result<()> {
return Ok(());
}
}
fn release_when_opened(lock: &Path) -> std::thread::JoinHandle<()> {
use std::os::unix::ffi::OsStrExt;
let directory = std::ffi::CString::new(lock.parent().expect("a directory").as_os_str().as_bytes()).expect("a path");
let descriptor: i32 = unsafe { libc::inotify_init1(libc::IN_CLOEXEC) };
assert!(descriptor >= 0, "inotify is available");
let added: i32 = unsafe { libc::inotify_add_watch(descriptor, directory.as_ptr(), libc::IN_OPEN) };
assert!(added >= 0, "the directory is watched");
let target: PathBuf = lock.to_path_buf();
return std::thread::spawn(move || {
let name: Vec<u8> = target.file_name().expect("a name").as_bytes().to_vec();
loop {
let mut buffer: [u8; 4096] = [0; 4096];
let read: isize = unsafe { libc::read(descriptor, buffer.as_mut_ptr().cast(), buffer.len()) };
let count: usize = usize::try_from(read).expect("events were read");
if buffer[..count].windows(name.len()).any(|window| { return window == name.as_slice(); }) {
break;
}
}
unsafe { libc::close(descriptor) };
std::fs::remove_file(&target).expect("release");
});
}
#[test]
fn the_holder_is_told_by_the_mark() {
let root: PathBuf = fixture("index-wait-holder");
let lock: PathBuf = lock_path_of(index_in(root.as_path()).as_path());
assert_eq!(holder_of(lock.as_path()), Holder::Nobody);
std::fs::write(&lock, marked(7)).expect("lock");
assert_eq!(holder_of(lock.as_path()), Holder::Service);
std::fs::write(&lock, "").expect("lock");
assert_eq!(holder_of(lock.as_path()), Holder::Foreign, "an empty lock is a program that has not written its index yet");
std::fs::write(&lock, "DIRC\0\0\0\u{2}").expect("lock");
assert_eq!(holder_of(lock.as_path()), Holder::Foreign);
std::fs::write(&lock, &LOCK_MARKER.as_bytes()[..LOCK_MARKER.len() / 2]).expect("lock");
assert_eq!(holder_of(lock.as_path()), Holder::Foreign, "half a mark is no mark");
remove(root.as_path());
}
#[test]
fn a_mark_that_arrives_a_moment_late_is_still_the_services() {
let root: PathBuf = fixture("index-wait-late");
let lock: PathBuf = lock_path_of(index_in(root.as_path()).as_path());
std::fs::write(&lock, "").expect("lock");
let pauses: std::cell::Cell<u32> = std::cell::Cell::new(0);
let writing = || {
pauses.set(pauses.get() + 1);
std::fs::write(&lock, marked(1)).expect("mark");
};
assert_eq!(holder_after_pauses(lock.as_path(), &writing), Holder::Service);
assert_eq!(pauses.get(), 1, "the mark was found by the read that followed the first pause");
remove(root.as_path());
}
#[test]
fn a_lock_that_stays_empty_is_foreign_after_the_reads_are_used_up() {
let root: PathBuf = fixture("index-wait-empty");
let lock: PathBuf = lock_path_of(index_in(root.as_path()).as_path());
std::fs::write(&lock, "").expect("lock");
let pauses: std::cell::Cell<u32> = std::cell::Cell::new(0);
let counting = || { pauses.set(pauses.get() + 1); };
assert_eq!(holder_after_pauses(lock.as_path(), &counting), Holder::Foreign);
assert_eq!(pauses.get(), EMPTY_LOOKS - 1, "one pause between each two reads");
remove(root.as_path());
}
#[test]
fn nothing_is_waited_for_when_nobody_or_a_foreign_program_holds_the_index() {
let root: PathBuf = fixture("index-wait-none");
let index: PathBuf = index_in(root.as_path());
let mut out: Vec<u8> = Vec::new();
assert_eq!(wait_for_service_hold(index.as_path(), "add", QUICK, &mut out), Waited::NotHeld);
std::fs::write(lock_path_of(index.as_path()), b"not the service\n").expect("lock");
assert_eq!(wait_for_service_hold(index.as_path(), "add", QUICK, &mut out), Waited::NotHeld);
assert!(out.is_empty(), "{}", String::from_utf8_lossy(&out));
remove(root.as_path());
}
#[test]
fn the_services_hold_is_waited_for_until_it_is_released() {
let root: PathBuf = fixture("index-wait-release");
let index: PathBuf = index_in(root.as_path());
let lock: PathBuf = lock_path_of(index.as_path());
std::fs::write(&lock, marked(3)).expect("lock");
let mut out: ReleaseOnNotice = ReleaseOnNotice::holding(lock.as_path());
let outcome: Waited = wait_for_service_hold(index.as_path(), "rebase", QUICK, &mut out);
assert_eq!(outcome, Waited::Released);
let text: String = out.written();
assert!(text.contains("git rebase") && text.contains("commit service"), "{text}");
assert!(!text.contains("index.lock") && !text.contains("lock"), "no internal name reaches the person: {text}");
remove(root.as_path());
}
#[test]
fn a_hold_that_outlasts_the_ceiling_ends_the_wait_and_stays() {
let root: PathBuf = fixture("index-wait-ceiling");
let index: PathBuf = index_in(root.as_path());
let lock: PathBuf = lock_path_of(index.as_path());
let identity: String = crate::service_index_marker::own_identity().expect("identity");
std::fs::write(&lock, format!("{LOCK_MARKER}9 {} {identity}\n", std::process::id())).expect("lock");
let (finished, ended) = std::sync::mpsc::channel();
let waiter_index: PathBuf = index.clone();
std::thread::spawn(move || {
let mut out: Vec<u8> = Vec::new();
let outcome: Waited = wait_for_service_hold_within(waiter_index.as_path(), "add", Duration::from_millis(300), QUICK, &mut out);
let _sent = finished.send((outcome, String::from_utf8_lossy(&out).into_owned()));
});
let (outcome, text) = ended.recv_timeout(Duration::from_secs(10)).expect("the wait ended at its ceiling");
assert_eq!(outcome, Waited::GaveUp);
assert!(text.contains("still finishing a commit") && text.contains("git add runs now"), "{text}");
assert!(lock.exists(), "the hold is not the command's to remove");
remove(root.as_path());
}
#[test]
fn where_no_watch_can_be_set_the_services_hold_is_polled_for() {
let root: PathBuf = fixture("index-wait-no-watch-poll");
let index: PathBuf = index_in(root.as_path());
let lock: PathBuf = lock_path_of(index.as_path());
let identity: String = crate::service_index_marker::own_identity().expect("identity");
std::fs::write(&lock, format!("{LOCK_MARKER}9 {} {identity}\n", std::process::id())).expect("lock");
let remover_lock: PathBuf = lock.clone();
let letting_go = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(300));
std::fs::remove_file(&remover_lock).expect("the service lets go");
});
let no_watch = |_: &Path| -> Option<Watch> { return None; };
let mut out: Vec<u8> = Vec::new();
let outcome: Waited = wait_with_watches(index.as_path(), "add", SERVICE_HOLD_CEILING, QUICK, &mut out, &no_watch);
letting_go.join().expect("the service lets go");
assert_eq!(outcome, Waited::Released, "{}", String::from_utf8_lossy(&out));
remove(root.as_path());
}
#[test]
fn a_holder_the_mark_proves_to_run_is_waited_for_with_no_bound() {
let root: PathBuf = fixture("index-wait-no-bound");
let index: PathBuf = index_in(root.as_path());
let lock: PathBuf = lock_path_of(index.as_path());
let identity: String = crate::service_index_marker::own_identity().expect("identity");
std::fs::write(&lock, format!("{LOCK_MARKER}9 {} {identity}\n", std::process::id())).expect("lock");
let holding: Duration = QUICK.interval * 20;
let remover_lock: PathBuf = lock.clone();
let letting_go = std::thread::spawn(move || {
std::thread::sleep(holding);
std::fs::remove_file(&remover_lock).expect("the service lets go");
});
let mut out: Vec<u8> = Vec::new();
assert_eq!(wait_for_service_hold(index.as_path(), "add", QUICK, &mut out), Waited::Released);
letting_go.join().expect("the service lets go");
let text: String = String::from_utf8_lossy(&out).into_owned();
assert!(text.matches("waiting for the commit service").count() > 1, "the line repeats while the service holds: {text}");
assert!(!text.contains("giving up") && !text.contains("runs now"), "a service that runs is never given up on: {text}");
remove(root.as_path());
}
#[test]
fn a_hold_older_than_any_landing_is_not_waited_for() {
let root: PathBuf = fixture("index-wait-stale");
let index: PathBuf = index_in(root.as_path());
let lock: PathBuf = lock_path_of(index.as_path());
std::fs::write(&lock, marked(5)).expect("lock");
let long_ago: std::time::SystemTime = std::time::SystemTime::now() - UNASKABLE_HOLD_AGE - Duration::from_secs(60);
std::fs::File::options().write(true).open(&lock).expect("open").set_modified(long_ago).expect("age");
let mut out: Vec<u8> = Vec::new();
assert_eq!(wait_for_service_hold(index.as_path(), "add", QUICK, &mut out), Waited::Stale);
assert!(String::from_utf8_lossy(&out).contains("does not wait for it"), "{}", String::from_utf8_lossy(&out));
assert!(lock.exists());
remove(root.as_path());
}
#[test]
fn a_hold_released_within_the_grace_prints_nothing() {
let root: PathBuf = fixture("index-wait-grace");
let index: PathBuf = index_in(root.as_path());
let lock: PathBuf = lock_path_of(index.as_path());
std::fs::write(&lock, marked(4)).expect("lock");
let releaser = release_when_opened(lock.as_path());
let mut out: Vec<u8> = Vec::new();
let slow: Cadence = Cadence { grace: Duration::from_secs(30), interval: Duration::from_secs(30) };
assert_eq!(wait_for_service_hold(index.as_path(), "add", slow, &mut out), Waited::Released);
releaser.join().expect("releaser");
assert!(out.is_empty(), "{}", String::from_utf8_lossy(&out));
remove(root.as_path());
}
#[test]
fn a_watch_that_ended_is_set_again_and_the_release_that_follows_is_seen() {
let root: PathBuf = fixture("index-wait-watch-ended");
let index: PathBuf = index_in(root.as_path());
let lock: PathBuf = lock_path_of(index.as_path());
std::fs::write(&lock, marked(6)).expect("lock");
let decoy: PathBuf = root.join("decoy");
std::fs::create_dir(&decoy).expect("decoy directory");
let set: std::cell::Cell<u32> = std::cell::Cell::new(0);
let set_watch = |directory: &Path| -> Option<Watch> {
set.set(set.get() + 1);
if set.get() > 1 {
std::fs::remove_file(&lock).expect("release");
return Watch::on(directory);
}
let ending: Option<Watch> = Watch::on(decoy.as_path());
std::fs::remove_dir(&decoy).expect("the watched directory goes");
return ending;
};
let began: Instant = Instant::now();
let mut out: Vec<u8> = Vec::new();
let slow: Cadence = Cadence { grace: Duration::from_secs(30), interval: Duration::from_secs(30) };
let outcome: Waited = wait_with_watches(index.as_path(), "add", SERVICE_HOLD_CEILING, slow, &mut out, &set_watch);
assert_eq!(outcome, Waited::Released);
assert_eq!(set.get(), 2, "a new watch was set once the first had ended");
assert!(began.elapsed() < slow.interval, "the release woke the wait, not the slow poll: {:?}", began.elapsed());
remove(root.as_path());
}
#[test]
fn a_hold_whose_process_runs_is_not_stale_by_age() {
let root: PathBuf = fixture("index-wait-live-old");
let index: PathBuf = index_in(root.as_path());
let lock: PathBuf = lock_path_of(index.as_path());
let identity: String = crate::service_index_marker::own_identity().expect("identity");
std::fs::write(&lock, format!("{LOCK_MARKER}8 {} {identity}\n", std::process::id())).expect("lock");
let long_ago: std::time::SystemTime = std::time::SystemTime::now() - UNASKABLE_HOLD_AGE - Duration::from_secs(60);
std::fs::File::options().write(true).open(&lock).expect("open").set_modified(long_ago).expect("age");
let mut waiting: ReleaseOnNotice = ReleaseOnNotice::holding(lock.as_path());
assert_eq!(wait_for_service_hold(index.as_path(), "add", QUICK, &mut waiting), Waited::Released);
assert!(!waiting.written().contains("does not wait for it"), "{}", waiting.written());
assert!(waiting.written().contains("waiting for the commit service"), "the hold was waited for: {}", waiting.written());
let mut out: Vec<u8> = Vec::new();
let parts: Vec<&str> = identity.split('#').collect();
std::fs::write(&lock, format!("{LOCK_MARKER}8 {} {}#{}#{}\n", std::process::id(), parts[0], "0".repeat(32), parts[2])).expect("lock of another boot");
std::fs::File::options().write(true).open(&lock).expect("open").set_modified(long_ago).expect("age");
assert_eq!(wait_for_service_hold(index.as_path(), "add", QUICK, &mut out), Waited::Stale);
remove(root.as_path());
}
#[test]
fn the_wait_drops_an_ended_watch_before_it_sets_the_next() {
let root: PathBuf = fixture("index-wait-watch-quota");
let index: PathBuf = index_in(root.as_path());
let lock: PathBuf = lock_path_of(index.as_path());
std::fs::write(&lock, marked(6)).expect("lock");
let decoy: PathBuf = root.join("decoy");
std::fs::create_dir(&decoy).expect("decoy");
let calls: std::cell::Cell<u32> = std::cell::Cell::new(0);
let quota_of_one = |directory: &Path| -> Option<Watch> {
calls.set(calls.get() + 1);
if calls.get() == 1 {
let ending: Watch = Watch::on(decoy.as_path()).expect("watch");
std::fs::remove_dir(&decoy).expect("the watched directory goes");
return Some(ending);
}
if crate::service_index_watch::live_watches_on_this_thread() > 0 {
return None;
}
std::fs::remove_file(&lock).expect("release");
return Watch::on(directory);
};
let began: Instant = Instant::now();
let mut out: Vec<u8> = Vec::new();
let slow: Cadence = Cadence { grace: Duration::from_secs(30), interval: Duration::from_secs(30) };
let outcome: Waited = wait_with_watches(index.as_path(), "add", SERVICE_HOLD_CEILING, slow, &mut out, "a_of_one);
assert_eq!(outcome, Waited::Released);
assert_eq!(calls.get(), 2);
assert!(began.elapsed() < slow.interval, "{:?}", began.elapsed());
remove(root.as_path());
}
#[test]
fn a_mark_whose_holder_is_gone_is_not_waited_for() {
let root: PathBuf = fixture("index-wait-gone-holder");
let index: PathBuf = index_in(root.as_path());
let lock: PathBuf = lock_path_of(index.as_path());
std::fs::write(&lock, format!("{LOCK_MARKER}7 2147483646 no-such-birth\n")).expect("lock");
let mut out: Vec<u8> = Vec::new();
assert_eq!(wait_for_service_hold(index.as_path(), "add", QUICK, &mut out), Waited::Stale);
let said: String = String::from_utf8_lossy(out.as_slice()).into_owned();
assert!(said.contains("no longer runs"), "{said}");
assert!(said.contains("next commit"), "{said}");
assert!(lock.exists(), "the leftover is not the command's to remove");
remove(root.as_path());
}
#[test]
fn a_hold_of_another_pid_namespace_is_neither_gone_nor_proven() {
let root: PathBuf = fixture("index-wait-foreign-namespace");
let index: PathBuf = index_in(root.as_path());
let lock: PathBuf = lock_path_of(index.as_path());
let identity: String = crate::service_index_marker::own_identity().expect("identity");
let parts: Vec<&str> = identity.split('#').collect();
std::fs::write(&lock, format!("{LOCK_MARKER}9 2147483646 {}#{}#1\n", parts[0], parts[1])).expect("lock");
let mut waiting: ReleaseOnNotice = ReleaseOnNotice::holding(lock.as_path());
assert_eq!(wait_for_service_hold(index.as_path(), "add", QUICK, &mut waiting), Waited::Released);
let said: String = waiting.written();
assert!(!said.contains("no longer runs"), "{said}");
std::fs::write(&lock, format!("{LOCK_MARKER}9 2147483646 {}#{}#1\n", parts[0], parts[1])).expect("lock again");
let long_ago: std::time::SystemTime = std::time::SystemTime::now() - UNASKABLE_HOLD_AGE - Duration::from_secs(60);
std::fs::File::options().write(true).open(&lock).expect("open").set_modified(long_ago).expect("age");
let mut old: Vec<u8> = Vec::new();
assert_eq!(wait_for_service_hold(index.as_path(), "add", QUICK, &mut old), Waited::Stale);
assert!(String::from_utf8_lossy(old.as_slice()).contains("does not wait for it"), "{}", String::from_utf8_lossy(old.as_slice()));
remove(root.as_path());
}