use std::error::Error;
use std::fs;
use std::io::{BufRead, BufReader, Read, Write};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::{Arc, Mutex};
use super::bridge::block_on;
use super::{DurableStore, EphemeralRoot, EphemeralSweepCounts};
type TestResult = Result<(), Box<dyn Error>>;
#[test]
fn a_throwaway_store_is_owned_under_its_configured_root() -> TestResult {
let persistence = tempfile::tempdir()?;
let root_path = persistence.path().join("ephemeral");
fs::create_dir(&root_path)?;
let root = EphemeralRoot::open(persistence.path())?;
let store = root.open_store(1)?;
let store_path = store
.ephemeral_dir_path()
.ok_or("store has no owned directory")?
.to_path_buf();
assert!(
store_path.starts_with(&root_path),
"throwaway store escaped its configured root: {}",
store_path.display()
);
drop(root);
assert!(
EphemeralRoot::open(persistence.path()).is_err(),
"store must retain its root lock"
);
drop(store);
assert!(!store_path.exists());
assert!(root_path.is_dir());
assert!(root_path.join("root.lock").is_file());
assert_eq!(
EphemeralRoot::open(persistence.path())?.sweep_counts(),
EphemeralSweepCounts::default()
);
Ok(())
}
#[derive(Clone, Default)]
struct Log(Arc<Mutex<Vec<u8>>>);
impl Write for Log {
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
self.0
.lock()
.map_err(|error| std::io::Error::other(error.to_string()))?
.extend_from_slice(bytes);
Ok(bytes.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
fn captured_root(path: &Path) -> Result<(Arc<EphemeralRoot>, String), Box<dyn Error>> {
let log = Log::default();
let writer = log.clone();
let subscriber = tracing_subscriber::fmt()
.with_ansi(false)
.without_time()
.with_writer(move || writer.clone())
.finish();
let root = tracing::subscriber::with_default(subscriber, || EphemeralRoot::open(path))?;
let text = String::from_utf8(log.0.lock().map_err(|error| error.to_string())?.clone())?;
Ok((root, text))
}
#[test]
fn an_empty_sweep_always_logs_zero_and_keeps_its_answer() -> TestResult {
let persistence = tempfile::tempdir()?;
let (root, log) = captured_root(persistence.path())?;
assert_eq!(root.sweep_counts(), EphemeralSweepCounts::default());
assert!(log.contains("ephemeral root startup sweep"), "{log}");
for field in ["removed_count=0", "removed_bytes=0", "held_count=0"] {
assert!(log.contains(field), "{log}");
}
let store = root.open_store(1)?;
drop(store);
assert_eq!(root.sweep_counts(), EphemeralSweepCounts::default());
Ok(())
}
struct ChildOwner {
process: Child,
output: Option<BufReader<std::process::ChildStdout>>,
}
impl Drop for ChildOwner {
fn drop(&mut self) {
match self.process.try_wait() {
Ok(Some(_)) => return,
Ok(None) => {}
Err(error) => tracing::error!(%error, "test child status failed"),
}
if let Err(error) = self.process.kill() {
tracing::error!(%error, "test child kill failed");
}
if let Err(error) = self.process.wait() {
tracing::error!(%error, "test child reap failed");
}
}
}
fn child(path: &Path, mode: &str) -> Result<(ChildOwner, PathBuf), Box<dyn Error>> {
let process = Command::new(std::env::current_exe()?)
.args([
"--exact",
"durability::tests_ephemeral_root::root_child",
"--nocapture",
])
.env("LIMINAL_ROOT_PIN_PARENT", path)
.env("LIMINAL_ROOT_PIN_MODE", mode)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::inherit())
.spawn()?;
let mut owner = ChildOwner {
process,
output: None,
};
owner.output = Some(BufReader::new(
owner.process.stdout.take().ok_or("child stdout absent")?,
));
loop {
let mut line = String::new();
if owner
.output
.as_mut()
.ok_or("child reader absent")?
.read_line(&mut line)?
== 0
{
break;
}
if let Some(path) = line.trim_end().strip_prefix("root-pin-ready ") {
return Ok((owner, PathBuf::from(path)));
}
}
Err("root child exited before ready signal".into())
}
fn tell_child(child: &mut ChildOwner, byte: u8) -> TestResult {
child
.process
.stdin
.as_mut()
.ok_or("child stdin absent")?
.write_all(&[byte])?;
Ok(())
}
fn file_bytes(path: &Path) -> Result<u64, Box<dyn Error>> {
let mut total = 0_u64;
for entry in fs::read_dir(path)? {
let entry = entry?;
let metadata = entry.metadata()?;
total += if metadata.is_dir() {
file_bytes(&entry.path())?
} else {
metadata.len()
};
}
Ok(total)
}
#[test]
fn a_crashed_child_is_swept_once_with_exact_bytes_and_named_log() -> TestResult {
let persistence = tempfile::tempdir()?;
let (mut child, entry) = child(persistence.path(), "crash")?;
tell_child(&mut child, b'C')?;
assert_eq!(child.process.wait()?.code(), Some(73));
assert!(entry.exists(), "process exit must leave crash residue");
let expected_bytes = file_bytes(&entry)?;
assert!(expected_bytes > 0);
let (root, log) = captured_root(persistence.path())?;
assert_eq!(
root.sweep_counts(),
EphemeralSweepCounts {
removed_count: 1,
removed_bytes: expected_bytes,
held_count: 0
}
);
assert!(!entry.exists());
assert!(log.contains(&entry.display().to_string()), "{log}");
assert!(log.contains(&format!("bytes={expected_bytes}")), "{log}");
assert!(log.contains("removed_count=1"), "{log}");
Ok(())
}
#[test]
fn a_live_holder_is_preserved_and_named() -> TestResult {
let persistence = tempfile::tempdir()?;
fs::create_dir(persistence.path().join("ephemeral"))?;
let (mut child, entry) = child(persistence.path(), "held")?;
let (root, log) = captured_root(persistence.path())?;
assert_eq!(
root.sweep_counts(),
EphemeralSweepCounts {
removed_count: 0,
removed_bytes: 0,
held_count: 1
}
);
assert!(entry.is_dir());
assert!(log.contains(&entry.display().to_string()), "{log}");
assert!(log.contains("kept live ephemeral entry"), "{log}");
tell_child(&mut child, b'S')?;
assert!(child.process.wait()?.success());
drop(root);
assert_eq!(
EphemeralRoot::open(persistence.path())?
.sweep_counts()
.removed_count,
1
);
Ok(())
}
#[test]
fn a_second_root_owner_refuses_without_removing_any_entry() -> TestResult {
let persistence = tempfile::tempdir()?;
let root = EphemeralRoot::open(persistence.path())?;
let store = root.open_store(1)?;
let entry = store.ephemeral_dir_path().ok_or("store path absent")?;
let error = EphemeralRoot::open(persistence.path())
.err()
.ok_or("second owner did not refuse")?;
assert!(
error.to_string().contains("ephemeral root lock held"),
"{error}"
);
assert!(entry.is_dir());
assert_eq!(root.sweep_counts(), EphemeralSweepCounts::default());
Ok(())
}
#[test]
fn a_missing_operator_parent_is_never_created() -> TestResult {
let persistence = tempfile::tempdir()?;
let absent = persistence.path().join("missing");
assert!(EphemeralRoot::open(&absent).is_err());
assert!(!absent.exists());
Ok(())
}
#[test]
fn root_child() -> TestResult {
let Some(parent) = std::env::var_os("LIMINAL_ROOT_PIN_PARENT") else {
return Ok(());
};
let parent = PathBuf::from(parent);
let mode = std::env::var("LIMINAL_ROOT_PIN_MODE")?;
if mode == "held" {
let entry = parent.join("ephemeral/liminal-durability-live-holder");
fs::create_dir(&entry)?;
fs::write(entry.join("payload"), b"live")?;
let holder = super::ephemeral_root::lock_file(&entry.join("holder.lock"))?;
rustix::fs::flock(&holder, rustix::fs::FlockOperation::LockExclusive)?;
writeln!(std::io::stdout(), "root-pin-ready {}", entry.display())?;
std::io::stdout().flush()?;
std::io::stdin().read_exact(&mut [0_u8])?;
drop(holder);
return Ok(());
}
assert_eq!(mode, "crash");
let root = EphemeralRoot::open(&parent)?;
let store = root.open_store(1)?;
block_on(store.append("crash-pin", b"written payload".to_vec(), 0))??;
block_on(store.flush())??;
let entry = store
.ephemeral_dir_path()
.ok_or("child store path absent")?;
writeln!(std::io::stdout(), "root-pin-ready {}", entry.display())?;
std::io::stdout().flush()?;
std::io::stdin().read_exact(&mut [0_u8])?;
std::process::exit(73);
}
#[test]
fn unrooted_construction_is_an_explicit_test_support_seam() -> TestResult {
let source = include_str!("store.rs");
let prefix = source
.split("pub fn open_ephemeral(")
.next()
.ok_or("test constructor absent")?;
assert!(prefix.ends_with("#[cfg(any(test, feature = \"test-support\"))]\n"));
Ok(())
}