truth_mirror/
state.rs

1//! Crash-safe transaction boundary for mutable `.truth` state.
2//!
3//! The state lock is deliberately separate from `watcher.lock`: watcher ownership
4//! spans reviewer execution, while this guard only protects short filesystem
5//! mutations. Lock records use the same pid-plus-start-token identity as watcher
6//! ownership, and the OS releases the exclusive file lock when a process dies.
7
8use std::{
9    fs,
10    io::{self, Seek, SeekFrom, Write},
11    path::{Path, PathBuf},
12    sync::atomic::{AtomicU64, Ordering},
13    time::{SystemTime, UNIX_EPOCH},
14};
15
16use fs2::FileExt;
17use serde::{Deserialize, Serialize};
18use thiserror::Error;
19
20use crate::{time::unix_now, watcher::ProcessIdentity};
21
22/// File name of the short-lived shared-state transaction lock.
23pub const STATE_LOCK_FILE: &str = "state.lock";
24static TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(0);
25
26#[derive(Debug, Error)]
27pub enum StateError {
28    #[error("state transaction IO failed: {0}")]
29    Io(#[from] io::Error),
30    #[error("state transaction lock JSON failed: {0}")]
31    Json(#[from] serde_json::Error),
32}
33
34#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
35struct StateLock {
36    identity: ProcessIdentity,
37    created_at_unix: u64,
38}
39
40/// RAII guard for one short mutation of queue, run, or ledger state.
41///
42/// The durable record remains on disk for inspection, but the OS-level exclusive
43/// lock is the racing gate. A process crash releases that lock automatically; the
44/// next owner then overwrites the stale pid-plus-start-token identity.
45#[derive(Debug)]
46pub struct StateTransaction {
47    lock_file: fs::File,
48}
49
50impl StateTransaction {
51    pub fn acquire(root: impl Into<PathBuf>) -> Result<Self, StateError> {
52        let root = root.into();
53        create_dir_all_durable(&root)?;
54        let path = root.join(STATE_LOCK_FILE);
55        let mut lock_file = fs::OpenOptions::new()
56            .create(true)
57            .truncate(true)
58            .read(true)
59            .write(true)
60            .open(&path)?;
61        lock_file.lock_exclusive()?;
62
63        let lock = StateLock {
64            identity: ProcessIdentity::current(),
65            created_at_unix: unix_now(),
66        };
67        let bytes = serde_json::to_vec_pretty(&lock)?;
68        lock_file.set_len(0)?;
69        lock_file.seek(SeekFrom::Start(0))?;
70        lock_file.write_all(&bytes)?;
71        lock_file.write_all(b"\n")?;
72        lock_file.sync_all()?;
73        sync_directory_best_effort(&root);
74        Ok(Self { lock_file })
75    }
76}
77
78impl Drop for StateTransaction {
79    fn drop(&mut self) {
80        let _ = FileExt::unlock(&self.lock_file);
81    }
82}
83
84/// Replace `path` with `bytes` without exposing a partial destination file.
85///
86/// The temporary file is created beside the destination, fully synced, and then
87/// atomically renamed over the destination. Directory syncing is best-effort
88/// because some supported platforms do not allow directories to be opened.
89pub fn atomic_replace(path: &Path, bytes: &[u8]) -> Result<(), StateError> {
90    let parent = path.parent().unwrap_or_else(|| Path::new("."));
91    create_dir_all_durable(parent)?;
92    let (temp_path, mut temp) = create_temp_file(path, parent)?;
93    let result = (|| -> Result<(), StateError> {
94        temp.write_all(bytes)?;
95        temp.sync_all()?;
96        drop(temp);
97        fs::rename(&temp_path, path)?;
98        sync_directory_best_effort(parent);
99        Ok(())
100    })();
101    if result.is_err() {
102        let _ = fs::remove_file(&temp_path);
103    }
104    result
105}
106
107fn create_temp_file(path: &Path, parent: &Path) -> Result<(PathBuf, fs::File), StateError> {
108    let file_name = path
109        .file_name()
110        .and_then(|name| name.to_str())
111        .unwrap_or("state");
112    loop {
113        let sequence = TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
114        let nanos = SystemTime::now()
115            .duration_since(UNIX_EPOCH)
116            .map_or(0, |duration| duration.as_nanos());
117        let temp_path = parent.join(format!(
118            ".{file_name}.{}.{}.{}.tmp",
119            std::process::id(),
120            nanos,
121            sequence
122        ));
123        match fs::OpenOptions::new()
124            .write(true)
125            .create_new(true)
126            .open(&temp_path)
127        {
128            Ok(file) => return Ok((temp_path, file)),
129            Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {}
130            Err(error) => return Err(error.into()),
131        }
132    }
133}
134
135fn create_dir_all_durable(path: &Path) -> Result<(), StateError> {
136    let mut missing = Vec::new();
137    let mut cursor = path;
138    while !cursor.exists() {
139        missing.push(cursor.to_path_buf());
140        let Some(parent) = cursor.parent() else {
141            break;
142        };
143        cursor = parent;
144    }
145
146    fs::create_dir_all(path)?;
147    for directory in missing.iter().rev() {
148        if let Some(parent) = directory.parent() {
149            sync_directory_best_effort(parent);
150        }
151        sync_directory_best_effort(directory);
152    }
153    Ok(())
154}
155
156pub(crate) fn sync_directory_best_effort(path: &Path) {
157    if let Ok(directory) = fs::File::open(path) {
158        let _ = directory.sync_all();
159    }
160}
161
162#[cfg(test)]
163mod tests {
164    use std::sync::{
165        Arc, Barrier,
166        atomic::{AtomicBool, Ordering},
167        mpsc,
168    };
169
170    use super::{STATE_LOCK_FILE, StateLock, StateTransaction, atomic_replace};
171    use crate::watcher::ProcessIdentity;
172
173    #[test]
174    fn state_lock_is_distinct_from_watcher_lock() {
175        let temp = tempfile::tempdir().unwrap();
176        std::fs::write(temp.path().join("watcher.lock"), b"occupied").unwrap();
177
178        let transaction = StateTransaction::acquire(temp.path()).unwrap();
179
180        assert!(temp.path().join(STATE_LOCK_FILE).exists());
181        drop(transaction);
182        assert!(temp.path().join("watcher.lock").exists());
183    }
184
185    #[test]
186    fn state_transactions_are_exclusive_within_one_process() {
187        let temp = tempfile::tempdir().unwrap();
188        let first = StateTransaction::acquire(temp.path()).unwrap();
189        let root = temp.path().to_path_buf();
190        let acquired = Arc::new(AtomicBool::new(false));
191        let acquired_in_thread = Arc::clone(&acquired);
192        let (attempting_tx, attempting_rx) = mpsc::channel();
193        let waiter = std::thread::spawn(move || {
194            attempting_tx.send(()).unwrap();
195            let second = StateTransaction::acquire(root).unwrap();
196            acquired_in_thread.store(true, Ordering::Release);
197            drop(second);
198        });
199
200        attempting_rx.recv().unwrap();
201        std::thread::sleep(std::time::Duration::from_millis(50));
202        assert!(!acquired.load(Ordering::Acquire));
203
204        drop(first);
205        waiter.join().unwrap();
206        assert!(acquired.load(Ordering::Acquire));
207    }
208
209    #[test]
210    fn stale_state_lock_identity_is_recovered() {
211        let temp = tempfile::tempdir().unwrap();
212        let stale = StateLock {
213            identity: ProcessIdentity {
214                pid: u32::MAX,
215                start_token: "dead".to_owned(),
216            },
217            created_at_unix: 0,
218        };
219        std::fs::write(
220            temp.path().join(STATE_LOCK_FILE),
221            serde_json::to_vec(&stale).unwrap(),
222        )
223        .unwrap();
224
225        let transaction = StateTransaction::acquire(temp.path()).unwrap();
226        let recovered: StateLock =
227            serde_json::from_slice(&std::fs::read(temp.path().join(STATE_LOCK_FILE)).unwrap())
228                .unwrap();
229
230        assert_eq!(recovered.identity.pid, std::process::id());
231        drop(transaction);
232    }
233
234    #[test]
235    fn atomic_replace_never_exposes_partial_bytes() {
236        let temp = tempfile::tempdir().unwrap();
237        let path = temp.path().join("nested").join("state.json");
238        let first = vec![b'a'; 128 * 1024];
239        let second = vec![b'b'; 128 * 1024];
240        atomic_replace(&path, &first).unwrap();
241
242        let barrier = Arc::new(Barrier::new(2));
243        let stop = Arc::new(AtomicBool::new(false));
244        let reader_path = path.clone();
245        let reader_barrier = Arc::clone(&barrier);
246        let reader_stop = Arc::clone(&stop);
247        let reader = std::thread::spawn(move || {
248            reader_barrier.wait();
249            while !reader_stop.load(Ordering::Acquire) {
250                let bytes = std::fs::read(&reader_path).unwrap();
251                assert!(
252                    bytes.iter().all(|byte| *byte == b'a')
253                        || bytes.iter().all(|byte| *byte == b'b')
254                );
255                assert_eq!(bytes.len(), 128 * 1024);
256            }
257        });
258
259        barrier.wait();
260        for _ in 0..20 {
261            atomic_replace(&path, &second).unwrap();
262            atomic_replace(&path, &first).unwrap();
263        }
264        stop.store(true, Ordering::Release);
265        reader.join().unwrap();
266
267        let temp_files = std::fs::read_dir(path.parent().unwrap())
268            .unwrap()
269            .filter_map(Result::ok)
270            .filter(|entry| entry.file_name().to_string_lossy().ends_with(".tmp"))
271            .count();
272        assert_eq!(temp_files, 0);
273    }
274}
truth_mirror/state.rs

truth_mirror/
state.rs
