a3s_box_runtime/
box_state.rs

1//! Durable local state store for box execution records.
2
3use std::path::{Path, PathBuf};
4
5use crate::file_lock::FileLock;
6use crate::store_io::quarantine_label;
7use crate::BoxRecord;
8use a3s_box_core::{ExecutionId, OperationId};
9
10/// Durable collection of local box execution records.
11///
12/// All mutating operations use the sibling `boxes.json.lock` advisory lock and
13/// a durable temporary-file rename. Callers must keep transaction closures
14/// synchronous and must not acquire the same store lock recursively.
15#[derive(Debug)]
16pub struct BoxStateStore {
17    path: PathBuf,
18    records: Vec<BoxRecord>,
19}
20
21impl BoxStateStore {
22    /// Build an in-memory store for `path` from existing records.
23    pub fn from_records(path: impl Into<PathBuf>, records: Vec<BoxRecord>) -> Self {
24        Self {
25            path: path.into(),
26            records,
27        }
28    }
29
30    /// Load state strictly, returning invalid JSON or schema data as an error.
31    ///
32    /// A missing state file is represented by an empty store and its parent
33    /// directory is created for subsequent writes.
34    pub fn load(path: &Path) -> std::io::Result<Self> {
35        Self::load_unlocked(path, CorruptionPolicy::ReturnError, true)
36    }
37
38    /// Load state and preserve an invalid file as a timestamped sibling.
39    ///
40    /// This compatibility path keeps the CLI available for manual recovery.
41    /// New runtime services should prefer [`Self::load`] and fail closed.
42    pub fn load_or_quarantine(path: &Path) -> std::io::Result<Self> {
43        Self::load_unlocked(path, CorruptionPolicy::Quarantine, true)
44    }
45
46    /// Load a side-effect-free snapshot.
47    ///
48    /// This never creates directories, quarantines invalid data, acquires a
49    /// lock, reconciles process state, or writes the file back.
50    pub fn load_readonly(path: impl Into<PathBuf>) -> std::io::Result<Self> {
51        let path = path.into();
52        Self::load_unlocked(&path, CorruptionPolicy::ReturnError, false)
53    }
54
55    /// Save this snapshot under the cross-process state lock.
56    pub fn save(&self) -> std::io::Result<()> {
57        let _lock = FileLock::acquire(&self.path)?;
58        self.write_unlocked()
59    }
60
61    /// Apply a strict atomic read-modify-write transaction.
62    ///
63    /// The closure runs while the cross-process lock is held. If it returns an
64    /// error, no write is performed.
65    pub fn modify<R>(
66        path: &Path,
67        f: impl FnOnce(&mut Self) -> std::io::Result<R>,
68    ) -> std::io::Result<R> {
69        Self::modify_with_policy(path, CorruptionPolicy::ReturnError, f)
70    }
71
72    /// Apply a strict atomic transaction with a caller-defined error type.
73    ///
74    /// This is equivalent to [`Self::modify`] but lets a domain repository
75    /// return typed conflicts while still converting state I/O errors.
76    pub fn transact<R, E>(path: &Path, f: impl FnOnce(&mut Self) -> Result<R, E>) -> Result<R, E>
77    where
78        E: From<std::io::Error>,
79    {
80        Self::modify_with_policy(path, CorruptionPolicy::ReturnError, f)
81    }
82
83    /// Apply an atomic read-modify-write transaction that quarantines invalid
84    /// existing state before starting from an empty collection.
85    ///
86    /// This exists for CLI behavior compatibility. Runtime services should use
87    /// [`Self::modify`] so corrupt durable state fails closed.
88    pub fn modify_or_quarantine<R, E>(
89        path: &Path,
90        f: impl FnOnce(&mut Self) -> Result<R, E>,
91    ) -> Result<R, E>
92    where
93        E: From<std::io::Error>,
94    {
95        Self::modify_with_policy(path, CorruptionPolicy::Quarantine, f)
96    }
97
98    fn modify_with_policy<R, E>(
99        path: &Path,
100        policy: CorruptionPolicy,
101        f: impl FnOnce(&mut Self) -> Result<R, E>,
102    ) -> Result<R, E>
103    where
104        E: From<std::io::Error>,
105    {
106        let _lock = FileLock::acquire(path).map_err(E::from)?;
107        let mut store = Self::load_unlocked(path, policy, true).map_err(E::from)?;
108        let output = f(&mut store)?;
109        store.write_unlocked().map_err(E::from)?;
110        Ok(output)
111    }
112
113    fn load_unlocked(
114        path: &Path,
115        corruption_policy: CorruptionPolicy,
116        create_parent: bool,
117    ) -> std::io::Result<Self> {
118        if !path.exists() {
119            if create_parent {
120                if let Some(parent) = path.parent() {
121                    std::fs::create_dir_all(parent)?;
122                }
123            }
124            return Ok(Self::from_records(path.to_path_buf(), Vec::new()));
125        }
126
127        let data = std::fs::read_to_string(path)?;
128        let parsed = serde_json::from_str::<Vec<BoxRecord>>(&data)
129            .map_err(|error| error.to_string())
130            .and_then(|records| {
131                validate_managed_records(&records)?;
132                Ok(records)
133            });
134        match parsed {
135            Ok(records) => Ok(Self::from_records(path.to_path_buf(), records)),
136            Err(error) if corruption_policy == CorruptionPolicy::ReturnError => {
137                Err(std::io::Error::new(std::io::ErrorKind::InvalidData, error))
138            }
139            Err(error) => {
140                let preserved = quarantine_label(path);
141                eprintln!(
142                    "a3s-box: WARNING: state file {} is corrupt ({error}); preserved a \
143                     copy at {preserved} and started from empty state. Running boxes are \
144                     no longer tracked; repair and restore the preserved records, then \
145                     reconcile state. Otherwise remove leaked executions manually.",
146                    path.display(),
147                );
148                Ok(Self::from_records(path.to_path_buf(), Vec::new()))
149            }
150        }
151    }
152
153    fn write_unlocked(&self) -> std::io::Result<()> {
154        validate_managed_records(&self.records)
155            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
156        if let Some(parent) = self.path.parent() {
157            std::fs::create_dir_all(parent)?;
158        }
159        let data = serde_json::to_vec_pretty(&self.records).map_err(std::io::Error::other)?;
160        let temporary_path = self.path.with_extension("json.tmp");
161        a3s_box_core::fs_atomic::write_durable(&temporary_path, &self.path, &data)
162    }
163
164    /// Path of the durable state file.
165    pub fn path(&self) -> &Path {
166        &self.path
167    }
168
169    /// All execution records in persistence order.
170    pub fn records(&self) -> &[BoxRecord] {
171        &self.records
172    }
173
174    /// Mutable execution records for a synchronous transaction.
175    pub fn records_mut(&mut self) -> &mut Vec<BoxRecord> {
176        &mut self.records
177    }
178
179    /// Find a record by exact execution ID.
180    pub fn find_by_id(&self, id: &str) -> Option<&BoxRecord> {
181        self.records.iter().find(|record| record.id == id)
182    }
183
184    /// Find a mutable record by exact execution ID.
185    pub fn find_by_id_mut(&mut self, id: &str) -> Option<&mut BoxRecord> {
186        self.records.iter_mut().find(|record| record.id == id)
187    }
188
189    /// Remove a record by exact execution ID.
190    pub fn remove_by_id(&mut self, id: &str) -> bool {
191        let previous_len = self.records.len();
192        self.records.retain(|record| record.id != id);
193        self.records.len() < previous_len
194    }
195
196    /// Find a record by exact user-visible name.
197    pub fn find_by_name(&self, name: &str) -> Option<&BoxRecord> {
198        self.records.iter().find(|record| record.name == name)
199    }
200
201    /// Find a managed execution by its idempotent creation operation.
202    pub fn find_by_operation_id(&self, operation_id: &OperationId) -> Option<&BoxRecord> {
203        self.records.iter().find(|record| {
204            record
205                .managed_execution
206                .as_ref()
207                .is_some_and(|metadata| &metadata.operation_id == operation_id)
208        })
209    }
210
211    /// Find a mutable managed execution by its idempotent creation operation.
212    pub fn find_by_operation_id_mut(
213        &mut self,
214        operation_id: &OperationId,
215    ) -> Option<&mut BoxRecord> {
216        self.records.iter_mut().find(|record| {
217            record
218                .managed_execution
219                .as_ref()
220                .is_some_and(|metadata| &metadata.operation_id == operation_id)
221        })
222    }
223
224    /// Find records matching a full-ID or short-ID prefix.
225    pub fn find_by_id_prefix(&self, prefix: &str) -> Vec<&BoxRecord> {
226        self.records
227            .iter()
228            .filter(|record| record.id.starts_with(prefix) || record.short_id.starts_with(prefix))
229            .collect()
230    }
231
232    /// List all records or only records in the running state.
233    pub fn list(&self, all: bool) -> Vec<&BoxRecord> {
234        self.records
235            .iter()
236            .filter(|record| all || record.status == "running")
237            .collect()
238    }
239}
240
241fn validate_managed_records(records: &[BoxRecord]) -> Result<(), String> {
242    let mut operation_ids = std::collections::HashSet::new();
243    for record in records {
244        let Some(metadata) = &record.managed_execution else {
245            continue;
246        };
247        let execution_id = ExecutionId::new(record.id.clone())
248            .map_err(|error| format!("invalid managed execution {}: {error}", record.id))?;
249        metadata
250            .validate()
251            .map_err(|error| format!("invalid managed execution {}: {error}", record.id))?;
252        if let Some(binding) = &metadata.oci_runtime {
253            binding
254                .validate_for(&execution_id)
255                .map_err(|error| format!("invalid managed execution {}: {error}", record.id))?;
256        }
257        record
258            .managed_state()
259            .map_err(|error| format!("invalid managed execution {}: {error}", record.id))?;
260        if !operation_ids.insert(metadata.operation_id.clone()) {
261            return Err(format!(
262                "duplicate managed operation ID: {}",
263                metadata.operation_id
264            ));
265        }
266    }
267    Ok(())
268}
269
270#[derive(Clone, Copy, PartialEq, Eq)]
271enum CorruptionPolicy {
272    ReturnError,
273    Quarantine,
274}
275
276#[cfg(test)]
277mod tests {
278    use super::*;
279
280    fn record(id: &str) -> BoxRecord {
281        serde_json::from_value(serde_json::json!({
282            "id": id,
283            "short_id": BoxRecord::make_short_id(id),
284            "name": format!("box-{id}"),
285            "image": "alpine:latest",
286            "status": "created",
287            "pid": null,
288            "cpus": 1,
289            "memory_mb": 128,
290            "volumes": [],
291            "env": {},
292            "cmd": ["sh"],
293            "box_dir": format!("/tmp/{id}"),
294            "console_log": format!("/tmp/{id}/console.log"),
295            "created_at": "2026-07-14T12:00:00Z",
296            "started_at": null,
297            "auto_remove": false
298        }))
299        .unwrap()
300    }
301
302    fn managed_record(id: &str, operation_id: OperationId) -> BoxRecord {
303        let mut record = record(id);
304        let config = a3s_box_core::BoxConfig {
305            image: "alpine:latest".to_string(),
306            isolation: a3s_box_core::ExecutionIsolation::Sandbox,
307            ..Default::default()
308        };
309        record.managed_execution = Some(
310            crate::ManagedExecutionMetadata::new(
311                operation_id,
312                a3s_box_core::ExecutionGeneration::INITIAL,
313                a3s_box_core::CreateExecutionRequest {
314                    external_sandbox_id: format!("sandbox-{id}"),
315                    config,
316                    labels: Default::default(),
317                    policy: Default::default(),
318                    rootfs_snapshot_id: None,
319                },
320            )
321            .unwrap(),
322        );
323        record
324    }
325
326    #[test]
327    fn missing_state_is_empty_and_creates_parent() {
328        let directory = tempfile::tempdir().unwrap();
329        let path = directory.path().join("nested").join("boxes.json");
330
331        let store = BoxStateStore::load(&path).unwrap();
332
333        assert!(store.records().is_empty());
334        assert!(path.parent().unwrap().exists());
335        assert!(!path.exists());
336    }
337
338    #[test]
339    fn strict_load_reports_corruption_without_moving_file() {
340        let directory = tempfile::tempdir().unwrap();
341        let path = directory.path().join("boxes.json");
342        std::fs::write(&path, "invalid json").unwrap();
343
344        let error = BoxStateStore::load(&path).unwrap_err();
345
346        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
347        assert_eq!(std::fs::read_to_string(&path).unwrap(), "invalid json");
348    }
349
350    #[test]
351    fn compatibility_load_quarantines_corruption() {
352        let directory = tempfile::tempdir().unwrap();
353        let path = directory.path().join("boxes.json");
354        std::fs::write(&path, "invalid json").unwrap();
355
356        let store = BoxStateStore::load_or_quarantine(&path).unwrap();
357
358        assert!(store.records().is_empty());
359        assert!(!path.exists());
360        let backups: Vec<_> = std::fs::read_dir(directory.path())
361            .unwrap()
362            .filter_map(Result::ok)
363            .filter(|entry| entry.file_name().to_string_lossy().contains(".corrupt-"))
364            .collect();
365        assert_eq!(backups.len(), 1);
366        assert_eq!(
367            std::fs::read_to_string(backups[0].path()).unwrap(),
368            "invalid json"
369        );
370    }
371
372    #[test]
373    fn failed_transaction_does_not_write_mutations() {
374        let directory = tempfile::tempdir().unwrap();
375        let path = directory.path().join("boxes.json");
376        BoxStateStore::from_records(path.clone(), vec![record("original")])
377            .save()
378            .unwrap();
379
380        let result = BoxStateStore::modify(&path, |store| {
381            store.records_mut().push(record("discarded"));
382            Err::<(), _>(std::io::Error::other("abort"))
383        });
384
385        assert!(result.is_err());
386        let persisted = BoxStateStore::load(&path).unwrap();
387        assert_eq!(persisted.records().len(), 1);
388        assert_eq!(persisted.records()[0].id, "original");
389    }
390
391    #[test]
392    fn save_preserves_runtime_owned_fields() {
393        let directory = tempfile::tempdir().unwrap();
394        let path = directory.path().join("boxes.json");
395        let mut value = record("runtime-field");
396        value.virtiofs_cache = Some("always".to_string());
397
398        BoxStateStore::from_records(path.clone(), vec![value])
399            .save()
400            .unwrap();
401
402        let persisted = BoxStateStore::load(&path).unwrap();
403        assert_eq!(
404            persisted.records()[0].virtiofs_cache.as_deref(),
405            Some("always")
406        );
407    }
408
409    #[test]
410    fn operation_lookup_ignores_legacy_records_and_finds_managed_intent() {
411        let operation_id = OperationId::new("operation-1").unwrap();
412        let managed = managed_record("managed", operation_id.clone());
413        let mut store =
414            BoxStateStore::from_records("/tmp/boxes.json", vec![record("legacy"), managed]);
415
416        assert_eq!(
417            store.find_by_operation_id(&operation_id).unwrap().id,
418            "managed"
419        );
420        store
421            .find_by_operation_id_mut(&operation_id)
422            .unwrap()
423            .status = "running".to_string();
424        assert_eq!(store.find_by_id("managed").unwrap().status, "running");
425        assert!(store
426            .find_by_operation_id(&OperationId::new("missing").unwrap())
427            .is_none());
428    }
429
430    #[test]
431    fn strict_load_rejects_duplicate_managed_operation_ids() {
432        let directory = tempfile::tempdir().unwrap();
433        let path = directory.path().join("boxes.json");
434        let operation_id = OperationId::new("operation-1").unwrap();
435        let records = vec![
436            managed_record("first", operation_id.clone()),
437            managed_record("second", operation_id),
438        ];
439        std::fs::write(&path, serde_json::to_vec(&records).unwrap()).unwrap();
440
441        let error = BoxStateStore::load(&path).unwrap_err();
442
443        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
444        assert!(error.to_string().contains("duplicate managed operation ID"));
445    }
446
447    #[test]
448    fn transaction_rejects_duplicate_operation_without_changing_disk() {
449        let directory = tempfile::tempdir().unwrap();
450        let path = directory.path().join("boxes.json");
451        let operation_id = OperationId::new("operation-1").unwrap();
452        BoxStateStore::from_records(
453            path.clone(),
454            vec![managed_record("first", operation_id.clone())],
455        )
456        .save()
457        .unwrap();
458
459        let error = BoxStateStore::modify(&path, |store| {
460            store
461                .records_mut()
462                .push(managed_record("second", operation_id));
463            Ok(())
464        })
465        .unwrap_err();
466
467        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
468        let persisted = BoxStateStore::load(&path).unwrap();
469        assert_eq!(persisted.records().len(), 1);
470        assert_eq!(persisted.records()[0].id, "first");
471    }
472
473    #[test]
474    fn strict_load_rejects_managed_plan_drift() {
475        let directory = tempfile::tempdir().unwrap();
476        let path = directory.path().join("boxes.json");
477        let mut record = managed_record("managed", OperationId::new("operation-1").unwrap());
478        record.managed_execution.as_mut().unwrap().plan =
479            a3s_box_core::resolve_execution(&a3s_box_core::BoxConfig::default()).unwrap();
480        std::fs::write(&path, serde_json::to_vec(&vec![record]).unwrap()).unwrap();
481
482        let error = BoxStateStore::load(&path).unwrap_err();
483
484        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
485        assert!(error.to_string().contains("does not match"));
486    }
487
488    #[test]
489    fn strict_load_rejects_unknown_managed_lifecycle_state() {
490        let directory = tempfile::tempdir().unwrap();
491        let path = directory.path().join("boxes.json");
492        let mut record = managed_record("managed", OperationId::new("operation-1").unwrap());
493        record.status = "future-state".to_string();
494        std::fs::write(&path, serde_json::to_vec(&vec![record]).unwrap()).unwrap();
495
496        let error = BoxStateStore::load(&path).unwrap_err();
497
498        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
499        assert!(error
500            .to_string()
501            .contains("unknown managed execution state"));
502    }
503
504    #[test]
505    fn strict_load_rejects_transition_without_matching_pending_operation() {
506        let directory = tempfile::tempdir().unwrap();
507        let path = directory.path().join("boxes.json");
508        let mut record = managed_record("managed", OperationId::new("operation-1").unwrap());
509        record.status = "pausing".to_string();
510        std::fs::write(&path, serde_json::to_vec(&vec![record]).unwrap()).unwrap();
511
512        let error = BoxStateStore::load(&path).unwrap_err();
513
514        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
515        assert!(error.to_string().contains("inconsistent pending operation"));
516    }
517
518    #[cfg(unix)]
519    #[test]
520    fn concurrent_transactions_do_not_lose_records() {
521        let directory = tempfile::tempdir().unwrap();
522        let path = directory.path().join("boxes.json");
523        let handles: Vec<_> = (0..8)
524            .map(|index| {
525                let path = path.clone();
526                std::thread::spawn(move || {
527                    BoxStateStore::modify(&path, |store| {
528                        store.records_mut().push(record(&format!("id-{index}")));
529                        Ok::<(), std::io::Error>(())
530                    })
531                    .unwrap();
532                })
533            })
534            .collect();
535
536        for handle in handles {
537            handle.join().unwrap();
538        }
539
540        let store = BoxStateStore::load(&path).unwrap();
541        assert_eq!(store.records().len(), 8);
542    }
543}
a3s_box_runtime/box_state.rs

a3s_box_runtime/
box_state.rs
