pharos_native/runtime/
state_store.rs

1//! Serialized state mutation for file-backed runtime state.
2
3use std::fs::{File, OpenOptions};
4use std::io::{ErrorKind, Write};
5use std::path::{Path, PathBuf};
6use std::thread;
7use std::time::{Duration, Instant};
8
9const LOCK_WAIT: Duration = Duration::from_secs(5);
10const LOCK_RETRY: Duration = Duration::from_millis(5);
11const STALE_LOCK_AGE: Duration = Duration::from_secs(120);
12
13/// Owns the exclusive lock for one state document mutation.
14///
15/// # Fields
16///
17/// - This type stores module-owned state for [`crate::runtime::state_store::StateMutationGuard`].
18///
19/// # Examples
20///
21/// ```rust,ignore
22/// let _ = std::mem::size_of::<pharos_native::runtime::state_store::StateMutationGuard>();
23/// ```
24///
25/// # Related
26///
27/// - [`crate::runtime::state_store`] for constructors, helpers, and neighboring runtime types.
28pub struct StateMutationGuard {
29    lock_path: PathBuf,
30    _file: File,
31}
32
33impl StateMutationGuard {
34    /// Acquires a cross-process lock covering the complete state read-modify-write operation.
35    ///
36    /// # Parameters
37    ///
38    /// - `state_path`: State file path whose lock companion will be acquired.
39    ///
40    /// # Returns
41    ///
42    /// Returns the acquired guard when the lock can be taken within the retry window.
43    ///
44    /// # Errors
45    ///
46    /// Returns an error when the lock path is invalid, unavailable, or times out.
47    ///
48    /// # Performance
49    ///
50    /// Performance is bounded by retry time and lock-file filesystem operations.
51    ///
52    /// # Examples
53    ///
54    /// ```rust,ignore
55    /// let _result = pharos_native::runtime::state_store::StateMutationGuard::acquire(std::path::Path::new("state.json"));
56    /// ```
57    ///
58    /// # Related
59    ///
60    /// - [`crate::runtime::state_store`] for adjacent helpers that share the same runtime surface.
61    pub fn acquire(state_path: &Path) -> Result<Self, &'static str> {
62        let file_name = state_path
63            .file_name()
64            .and_then(|value| value.to_str())
65            .filter(|value| !value.is_empty())
66            .ok_or("invalid state path")?;
67        let lock_path = state_path.with_file_name(format!(".{file_name}.lock"));
68        let started = Instant::now();
69        loop {
70            match OpenOptions::new()
71                .write(true)
72                .create_new(true)
73                .open(&lock_path)
74            {
75                Ok(mut file) => {
76                    let created_at = std::time::SystemTime::now()
77                        .duration_since(std::time::UNIX_EPOCH)
78                        .map(|value| value.as_secs())
79                        .unwrap_or(0);
80                    if writeln!(file, "{created_at}").is_err() || file.sync_all().is_err() {
81                        let _ = std::fs::remove_file(&lock_path);
82                        return Err("state lock initialization failed");
83                    }
84                    return Ok(Self {
85                        lock_path,
86                        _file: file,
87                    });
88                }
89                Err(error) if error.kind() == ErrorKind::AlreadyExists => {
90                    let now = std::time::SystemTime::now()
91                        .duration_since(std::time::UNIX_EPOCH)
92                        .map(|value| value.as_secs())
93                        .unwrap_or(0);
94                    let stale = std::fs::read_to_string(&lock_path)
95                        .ok()
96                        .and_then(|value| value.trim().parse::<u64>().ok())
97                        .is_some_and(|created_at| {
98                            now.saturating_sub(created_at) >= STALE_LOCK_AGE.as_secs()
99                        });
100                    if stale {
101                        let _ = std::fs::remove_file(&lock_path);
102                        continue;
103                    }
104                    if started.elapsed() >= LOCK_WAIT {
105                        return Err("state mutation lock timeout");
106                    }
107                    thread::sleep(LOCK_RETRY);
108                }
109                Err(_) => return Err("state mutation lock unavailable"),
110            }
111        }
112    }
113}
114
115impl Drop for StateMutationGuard {
116    /// Performs drop.
117    fn drop(&mut self) {
118        let _ = std::fs::remove_file(&self.lock_path);
119    }
120}
121
122#[cfg(test)]
123mod tests {
124    use super::StateMutationGuard;
125    use std::fs;
126    use std::sync::{Arc, Barrier};
127    use std::thread;
128
129    /// Confirms concurrent writers cannot enter the same state transaction together.
130    #[test]
131    fn serializes_concurrent_state_mutations() {
132        let root = std::env::temp_dir().join(format!("pharos-state-lock-{}", std::process::id()));
133        fs::create_dir_all(&root).expect("create test root");
134        let state_path = root.join("state.json");
135        fs::write(&state_path, "{}").expect("write state");
136        let entered = Arc::new(Barrier::new(2));
137        let worker_path = state_path.clone();
138        let worker_entered = Arc::clone(&entered);
139        let worker = thread::spawn(move || {
140            let _guard = StateMutationGuard::acquire(&worker_path).expect("worker lock");
141            worker_entered.wait();
142            thread::sleep(std::time::Duration::from_millis(50));
143        });
144        entered.wait();
145        let started = std::time::Instant::now();
146        let _guard = StateMutationGuard::acquire(&state_path).expect("second lock");
147        assert!(started.elapsed() >= std::time::Duration::from_millis(40));
148        worker.join().expect("join worker");
149        drop(_guard);
150        fs::remove_dir_all(root).expect("remove test root");
151    }
152
153    /// Confirms an abandoned lock record is recovered without mutating state.
154    #[test]
155    fn recovers_an_abandoned_state_lock() {
156        let root =
157            std::env::temp_dir().join(format!("pharos-state-stale-lock-{}", std::process::id()));
158        fs::create_dir_all(&root).expect("create test root");
159        let state_path = root.join("state.json");
160        fs::write(&state_path, "{}").expect("write state");
161        fs::write(root.join(".state.json.lock"), "0\n").expect("write abandoned lock");
162        let guard = StateMutationGuard::acquire(&state_path).expect("recover lock");
163        assert_eq!(fs::read_to_string(&state_path).expect("read state"), "{}");
164        drop(guard);
165        fs::remove_dir_all(root).expect("remove test root");
166    }
167}
pharos_native/runtime/state_store.rs

pharos_native/runtime/
state_store.rs
