truth-mirror 0.13.3

//! Crash-safe transaction boundary for mutable `.truth` state.
//!
//! The state lock is deliberately separate from `watcher.lock`: watcher ownership
//! spans reviewer execution, while this guard only protects short filesystem
//! mutations. Lock records use the same pid-plus-start-token identity as watcher
//! ownership, and the OS releases the exclusive file lock when a process dies.

use std::{
    fs,
    io::{self, Seek, SeekFrom, Write},
    path::{Path, PathBuf},
    sync::atomic::{AtomicU64, Ordering},
    time::{SystemTime, UNIX_EPOCH},
};

use fs2::FileExt;
use serde::{Deserialize, Serialize};
use thiserror::Error;

use crate::{time::unix_now, watcher::ProcessIdentity};

/// File name of the short-lived shared-state transaction lock.
pub const STATE_LOCK_FILE: &str = "state.lock";
static TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(0);

#[derive(Debug, Error)]
pub enum StateError {
    #[error("state transaction IO failed: {0}")]
    Io(#[from] io::Error),
    #[error("state transaction lock JSON failed: {0}")]
    Json(#[from] serde_json::Error),
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
struct StateLock {
    identity: ProcessIdentity,
    created_at_unix: u64,
}

/// RAII guard for one short mutation of queue, run, or ledger state.
///
/// The durable record remains on disk for inspection, but the OS-level exclusive
/// lock is the racing gate. A process crash releases that lock automatically; the
/// next owner then overwrites the stale pid-plus-start-token identity.
#[derive(Debug)]
pub struct StateTransaction {
    lock_file: fs::File,
}

impl StateTransaction {
    pub fn acquire(root: impl Into<PathBuf>) -> Result<Self, StateError> {
        let root = root.into();
        create_dir_all_durable(&root)?;
        let path = root.join(STATE_LOCK_FILE);
        let mut lock_file = fs::OpenOptions::new()
            .create(true)
            .truncate(true)
            .read(true)
            .write(true)
            .open(&path)?;
        lock_file.lock_exclusive()?;

        let lock = StateLock {
            identity: ProcessIdentity::current(),
            created_at_unix: unix_now(),
        };
        let bytes = serde_json::to_vec_pretty(&lock)?;
        lock_file.set_len(0)?;
        lock_file.seek(SeekFrom::Start(0))?;
        lock_file.write_all(&bytes)?;
        lock_file.write_all(b"\n")?;
        lock_file.sync_all()?;
        sync_directory_best_effort(&root);
        Ok(Self { lock_file })
    }
}

impl Drop for StateTransaction {
    fn drop(&mut self) {
        let _ = FileExt::unlock(&self.lock_file);
    }
}

/// Replace `path` with `bytes` without exposing a partial destination file.
///
/// The temporary file is created beside the destination, fully synced, and then
/// atomically renamed over the destination. Directory syncing is best-effort
/// because some supported platforms do not allow directories to be opened.
pub fn atomic_replace(path: &Path, bytes: &[u8]) -> Result<(), StateError> {
    let parent = path.parent().unwrap_or_else(|| Path::new("."));
    create_dir_all_durable(parent)?;
    let (temp_path, mut temp) = create_temp_file(path, parent)?;
    let result = (|| -> Result<(), StateError> {
        temp.write_all(bytes)?;
        temp.sync_all()?;
        drop(temp);
        fs::rename(&temp_path, path)?;
        sync_directory_best_effort(parent);
        Ok(())
    })();
    if result.is_err() {
        let _ = fs::remove_file(&temp_path);
    }
    result
}

fn create_temp_file(path: &Path, parent: &Path) -> Result<(PathBuf, fs::File), StateError> {
    let file_name = path
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or("state");
    loop {
        let sequence = TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_or(0, |duration| duration.as_nanos());
        let temp_path = parent.join(format!(
            ".{file_name}.{}.{}.{}.tmp",
            std::process::id(),
            nanos,
            sequence
        ));
        match fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&temp_path)
        {
            Ok(file) => return Ok((temp_path, file)),
            Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {}
            Err(error) => return Err(error.into()),
        }
    }
}

fn create_dir_all_durable(path: &Path) -> Result<(), StateError> {
    let mut missing = Vec::new();
    let mut cursor = path;
    while !cursor.exists() {
        missing.push(cursor.to_path_buf());
        let Some(parent) = cursor.parent() else {
            break;
        };
        cursor = parent;
    }

    fs::create_dir_all(path)?;
    for directory in missing.iter().rev() {
        if let Some(parent) = directory.parent() {
            sync_directory_best_effort(parent);
        }
        sync_directory_best_effort(directory);
    }
    Ok(())
}

pub(crate) fn sync_directory_best_effort(path: &Path) {
    if let Ok(directory) = fs::File::open(path) {
        let _ = directory.sync_all();
    }
}

#[cfg(test)]
mod tests {
    use std::sync::{
        Arc, Barrier,
        atomic::{AtomicBool, Ordering},
        mpsc,
    };

    use super::{STATE_LOCK_FILE, StateLock, StateTransaction, atomic_replace};
    use crate::watcher::ProcessIdentity;

    #[test]
    fn state_lock_is_distinct_from_watcher_lock() {
        let temp = tempfile::tempdir().unwrap();
        std::fs::write(temp.path().join("watcher.lock"), b"occupied").unwrap();

        let transaction = StateTransaction::acquire(temp.path()).unwrap();

        assert!(temp.path().join(STATE_LOCK_FILE).exists());
        drop(transaction);
        assert!(temp.path().join("watcher.lock").exists());
    }

    #[test]
    fn state_transactions_are_exclusive_within_one_process() {
        let temp = tempfile::tempdir().unwrap();
        let first = StateTransaction::acquire(temp.path()).unwrap();
        let root = temp.path().to_path_buf();
        let acquired = Arc::new(AtomicBool::new(false));
        let acquired_in_thread = Arc::clone(&acquired);
        let (attempting_tx, attempting_rx) = mpsc::channel();
        let waiter = std::thread::spawn(move || {
            attempting_tx.send(()).unwrap();
            let second = StateTransaction::acquire(root).unwrap();
            acquired_in_thread.store(true, Ordering::Release);
            drop(second);
        });

        attempting_rx.recv().unwrap();
        std::thread::sleep(std::time::Duration::from_millis(50));
        assert!(!acquired.load(Ordering::Acquire));

        drop(first);
        waiter.join().unwrap();
        assert!(acquired.load(Ordering::Acquire));
    }

    #[test]
    fn stale_state_lock_identity_is_recovered() {
        let temp = tempfile::tempdir().unwrap();
        let stale = StateLock {
            identity: ProcessIdentity {
                pid: u32::MAX,
                start_token: "dead".to_owned(),
            },
            created_at_unix: 0,
        };
        std::fs::write(
            temp.path().join(STATE_LOCK_FILE),
            serde_json::to_vec(&stale).unwrap(),
        )
        .unwrap();

        let transaction = StateTransaction::acquire(temp.path()).unwrap();
        let recovered: StateLock =
            serde_json::from_slice(&std::fs::read(temp.path().join(STATE_LOCK_FILE)).unwrap())
                .unwrap();

        assert_eq!(recovered.identity.pid, std::process::id());
        drop(transaction);
    }

    #[test]
    fn atomic_replace_never_exposes_partial_bytes() {
        let temp = tempfile::tempdir().unwrap();
        let path = temp.path().join("nested").join("state.json");
        let first = vec![b'a'; 128 * 1024];
        let second = vec![b'b'; 128 * 1024];
        atomic_replace(&path, &first).unwrap();

        let barrier = Arc::new(Barrier::new(2));
        let stop = Arc::new(AtomicBool::new(false));
        let reader_path = path.clone();
        let reader_barrier = Arc::clone(&barrier);
        let reader_stop = Arc::clone(&stop);
        let reader = std::thread::spawn(move || {
            reader_barrier.wait();
            while !reader_stop.load(Ordering::Acquire) {
                let bytes = std::fs::read(&reader_path).unwrap();
                assert!(
                    bytes.iter().all(|byte| *byte == b'a')
                        || bytes.iter().all(|byte| *byte == b'b')
                );
                assert_eq!(bytes.len(), 128 * 1024);
            }
        });

        barrier.wait();
        for _ in 0..20 {
            atomic_replace(&path, &second).unwrap();
            atomic_replace(&path, &first).unwrap();
        }
        stop.store(true, Ordering::Release);
        reader.join().unwrap();

        let temp_files = std::fs::read_dir(path.parent().unwrap())
            .unwrap()
            .filter_map(Result::ok)
            .filter(|entry| entry.file_name().to_string_lossy().ends_with(".tmp"))
            .count();
        assert_eq!(temp_files, 0);
    }
}