haematite 0.7.0

//! FENCE-CHAIN r8 R3 — the self-re-exec crash harness the fence pins drive.
//!
//! This reuses the PROTOCOL of the landed native OS-child READY harness
//! (`tests/crash_gates.rs` and `branch/crash_support.rs`): a writer child performs
//! a durable operation, runs the R1a CUT (rewinding every directory-entry mutation
//! whose fence did not land), reports READY, and blocks; the parent kills it
//! (`kill -9`, no destructors, no follow-up flush); a fresh recovery child then
//! cold-reopens and asserts the post-READY oracle.
//!
//! It is, deliberately, a SECOND coordinator IMPLEMENTATION of that one protocol —
//! not a shared coordinator. That duplication is forced, not incidental: the CUT
//! is `cfg(test)` machinery that exists ONLY in the unit-test binary, so
//! `tests/crash_gates.rs`'s coordinator — an integration test that links the
//! library WITHOUT `cfg(test)` — structurally cannot reach the fence-chain journal
//! to drive the CUT. Two implementations, one protocol, is therefore the correct
//! shape, and the ruling-3 lineage constraint ("extend the landed harness, no
//! parallel OS-kill coordinators") is honoured at the PROTOCOL level rather than
//! the code level. Recorded here as the tear-seat sign-off on ruling 3 (Waffles,
//! 2026-07-22): one OS-kill protocol, and the only reason it has two drivers is the
//! `cfg(test)`-visibility boundary of the journal.

use std::env;
use std::error::Error;
use std::io::{BufRead, BufReader, Read, Write};
use std::path::{Path, PathBuf};
use std::process::{Child, ChildStdout, Command, Stdio};
use std::sync::mpsc::{self, RecvTimeoutError};
use std::thread;
use std::time::Duration;

pub type TestResult = Result<(), Box<dyn Error>>;

const ROLE_ENV: &str = "HAEMATITE_FENCE_PIN_ROLE";
const DIR_ENV: &str = "HAEMATITE_FENCE_PIN_DIR";
const READY_LINE: &str = "HAEMATITE_FENCE_PIN_READY";
const DEADLINE: Duration = Duration::from_secs(30);
const MAX_OUTPUT: u64 = 64 * 1024;

/// One durable-operation writer + its cold-recovery oracle.
pub struct PinCase {
    /// The exact unit-test path the child re-execs (`--exact <path>`).
    pub test_path: &'static str,
    /// Perform the durable operation, then run the CUT. Runs in the writer child.
    pub write: fn(&Path) -> TestResult,
    /// Cold-reopen and assert the post-READY oracle. Runs in the recovery child.
    pub recover: fn(&Path) -> TestResult,
}

fn harness_error(message: impl Into<String>) -> Box<dyn Error> {
    message.into().into()
}

/// Dispatch by role: parent orchestrates; writer performs+cuts+reports; recovery
/// asserts the oracle.
pub fn run_pin(case: &PinCase) -> TestResult {
    match env::var(ROLE_ENV).ok().as_deref() {
        None => run_parent(case),
        Some("writer") => run_writer(case),
        Some("recovery") => (case.recover)(&child_dir()?),
        Some(other) => Err(harness_error(format!("unknown fence-pin role {other:?}"))),
    }
}

fn child_dir() -> Result<PathBuf, Box<dyn Error>> {
    env::var_os(DIR_ENV)
        .map(PathBuf::from)
        .ok_or_else(|| harness_error("missing fence-pin child directory"))
}

fn run_writer(case: &PinCase) -> TestResult {
    let dir = child_dir()?;
    // Perform the durable op and run the CUT BEFORE reporting READY, so the
    // parent's kill cannot interrupt the deterministic disk rewind.
    (case.write)(&dir)?;
    let mut stdout = std::io::stdout().lock();
    writeln!(stdout, "{READY_LINE}")?;
    stdout.flush()?;
    drop(stdout);
    loop {
        thread::park();
    }
}

fn run_parent(case: &PinCase) -> TestResult {
    let temp = tempfile::tempdir()?;
    let dir = temp.path().join("world");
    std::fs::create_dir(&dir)?;

    let mut writer = spawn_child(case.test_path, "writer", &dir)?;
    let readiness = receive_readiness(&mut writer);
    // Kill the writer regardless of readiness outcome (no destructors run).
    let killed = kill_and_wait(writer);
    readiness?;
    let status = killed?;
    if status.success() {
        return Err(harness_error(
            "writer exited normally instead of being killed",
        ));
    }

    run_recovery_child(case.test_path, &dir)
}

fn spawn_child(test_path: &str, role: &str, dir: &Path) -> Result<Child, std::io::Error> {
    Command::new(env::current_exe()?)
        .arg("--exact")
        .arg(test_path)
        .arg("--nocapture")
        .arg("--test-threads=1")
        .env(ROLE_ENV, role)
        .env(DIR_ENV, dir)
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::inherit())
        .spawn()
}

fn receive_readiness(child: &mut Child) -> TestResult {
    let stdout = child
        .stdout
        .take()
        .ok_or_else(|| harness_error("writer readiness pipe is missing"))?;
    let (sender, receiver) = mpsc::channel();
    let _reader = thread::spawn(move || {
        let _ = sender.send(read_readiness(stdout));
    });
    match receiver.recv_timeout(DEADLINE) {
        Ok(Ok(())) => Ok(()),
        Ok(Err(message)) => Err(harness_error(message)),
        Err(RecvTimeoutError::Timeout) => Err(harness_error("writer readiness deadline expired")),
        Err(RecvTimeoutError::Disconnected) => Err(harness_error("writer readiness disconnected")),
    }
}

fn read_readiness(stdout: ChildStdout) -> Result<(), String> {
    let mut reader = BufReader::new(stdout);
    let mut seen = String::new();
    loop {
        let mut line = String::new();
        let bytes = reader
            .read_line(&mut line)
            .map_err(|error| format!("readiness read failed: {error}"))?;
        if bytes == 0 {
            return Err(format!(
                "writer reached EOF before readiness; output={seen:?}"
            ));
        }
        if line.contains(READY_LINE) {
            return Ok(());
        }
        seen.push_str(&line);
        if seen.len() as u64 > MAX_OUTPUT {
            return Err("writer output exceeded the bounded limit".to_owned());
        }
    }
}

fn kill_and_wait(mut child: Child) -> Result<std::process::ExitStatus, Box<dyn Error>> {
    Child::kill(&mut child).map_err(|error| harness_error(format!("kill failed: {error}")))?;
    let (sender, receiver) = mpsc::channel();
    let _waiter = thread::spawn(move || {
        let _ = sender.send(child.wait());
    });
    match receiver.recv_timeout(DEADLINE) {
        Ok(Ok(status)) => Ok(status),
        Ok(Err(error)) => Err(harness_error(format!("child wait failed: {error}"))),
        Err(_) => Err(harness_error("child wait deadline expired")),
    }
}

fn run_recovery_child(test_path: &str, dir: &Path) -> TestResult {
    let mut child = spawn_child(test_path, "recovery", dir)?;
    let stdout = child
        .stdout
        .take()
        .ok_or_else(|| harness_error("recovery output pipe is missing"))?;
    let (sender, receiver) = mpsc::channel();
    let _reader = thread::spawn(move || {
        let mut output = Vec::new();
        let _ = stdout.take(MAX_OUTPUT + 1).read_to_end(&mut output);
        let _ = sender.send(output);
    });
    let output = receiver
        .recv_timeout(DEADLINE)
        .map_err(|_| harness_error("recovery output deadline expired"))?;
    let status = {
        let (sender, receiver) = mpsc::channel();
        let _waiter = thread::spawn(move || {
            let _ = sender.send(child.wait());
        });
        receiver
            .recv_timeout(DEADLINE)
            .map_err(|_| harness_error("recovery wait deadline expired"))?
            .map_err(|error| harness_error(format!("recovery wait failed: {error}")))?
    };
    if !status.success() {
        return Err(harness_error(format!(
            "recovery child failed with {status}; output:\n{}",
            String::from_utf8_lossy(&output)
        )));
    }
    Ok(())
}