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        metadata
248            .validate()
249            .map_err(|error| format!("invalid managed execution {}: {error}", record.id))?;
250        ExecutionId::new(record.id.clone())
251            .map_err(|error| format!("invalid managed execution {}: {error}", record.id))?;
252        record
253            .managed_state()
254            .map_err(|error| format!("invalid managed execution {}: {error}", record.id))?;
255        if !operation_ids.insert(metadata.operation_id.clone()) {
256            return Err(format!(
257                "duplicate managed operation ID: {}",
258                metadata.operation_id
259            ));
260        }
261    }
262    Ok(())
263}
264
265#[derive(Clone, Copy, PartialEq, Eq)]
266enum CorruptionPolicy {
267    ReturnError,
268    Quarantine,
269}
270
271#[cfg(test)]
272mod tests {
273    use super::*;
274
275    fn record(id: &str) -> BoxRecord {
276        serde_json::from_value(serde_json::json!({
277            "id": id,
278            "short_id": BoxRecord::make_short_id(id),
279            "name": format!("box-{id}"),
280            "image": "alpine:latest",
281            "status": "created",
282            "pid": null,
283            "cpus": 1,
284            "memory_mb": 128,
285            "volumes": [],
286            "env": {},
287            "cmd": ["sh"],
288            "box_dir": format!("/tmp/{id}"),
289            "console_log": format!("/tmp/{id}/console.log"),
290            "created_at": "2026-07-14T12:00:00Z",
291            "started_at": null,
292            "auto_remove": false
293        }))
294        .unwrap()
295    }
296
297    fn managed_record(id: &str, operation_id: OperationId) -> BoxRecord {
298        let mut record = record(id);
299        let config = a3s_box_core::BoxConfig {
300            image: "alpine:latest".to_string(),
301            isolation: a3s_box_core::ExecutionIsolation::Sandbox,
302            ..Default::default()
303        };
304        record.managed_execution = Some(
305            crate::ManagedExecutionMetadata::new(
306                operation_id,
307                a3s_box_core::ExecutionGeneration::INITIAL,
308                a3s_box_core::CreateExecutionRequest {
309                    external_sandbox_id: format!("sandbox-{id}"),
310                    config,
311                    labels: Default::default(),
312                    policy: Default::default(),
313                    rootfs_snapshot_id: None,
314                },
315            )
316            .unwrap(),
317        );
318        record
319    }
320
321    #[test]
322    fn missing_state_is_empty_and_creates_parent() {
323        let directory = tempfile::tempdir().unwrap();
324        let path = directory.path().join("nested").join("boxes.json");
325
326        let store = BoxStateStore::load(&path).unwrap();
327
328        assert!(store.records().is_empty());
329        assert!(path.parent().unwrap().exists());
330        assert!(!path.exists());
331    }
332
333    #[test]
334    fn strict_load_reports_corruption_without_moving_file() {
335        let directory = tempfile::tempdir().unwrap();
336        let path = directory.path().join("boxes.json");
337        std::fs::write(&path, "invalid json").unwrap();
338
339        let error = BoxStateStore::load(&path).unwrap_err();
340
341        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
342        assert_eq!(std::fs::read_to_string(&path).unwrap(), "invalid json");
343    }
344
345    #[test]
346    fn compatibility_load_quarantines_corruption() {
347        let directory = tempfile::tempdir().unwrap();
348        let path = directory.path().join("boxes.json");
349        std::fs::write(&path, "invalid json").unwrap();
350
351        let store = BoxStateStore::load_or_quarantine(&path).unwrap();
352
353        assert!(store.records().is_empty());
354        assert!(!path.exists());
355        let backups: Vec<_> = std::fs::read_dir(directory.path())
356            .unwrap()
357            .filter_map(Result::ok)
358            .filter(|entry| entry.file_name().to_string_lossy().contains(".corrupt-"))
359            .collect();
360        assert_eq!(backups.len(), 1);
361        assert_eq!(
362            std::fs::read_to_string(backups[0].path()).unwrap(),
363            "invalid json"
364        );
365    }
366
367    #[test]
368    fn failed_transaction_does_not_write_mutations() {
369        let directory = tempfile::tempdir().unwrap();
370        let path = directory.path().join("boxes.json");
371        BoxStateStore::from_records(path.clone(), vec![record("original")])
372            .save()
373            .unwrap();
374
375        let result = BoxStateStore::modify(&path, |store| {
376            store.records_mut().push(record("discarded"));
377            Err::<(), _>(std::io::Error::other("abort"))
378        });
379
380        assert!(result.is_err());
381        let persisted = BoxStateStore::load(&path).unwrap();
382        assert_eq!(persisted.records().len(), 1);
383        assert_eq!(persisted.records()[0].id, "original");
384    }
385
386    #[test]
387    fn save_preserves_runtime_owned_fields() {
388        let directory = tempfile::tempdir().unwrap();
389        let path = directory.path().join("boxes.json");
390        let mut value = record("runtime-field");
391        value.virtiofs_cache = Some("always".to_string());
392
393        BoxStateStore::from_records(path.clone(), vec![value])
394            .save()
395            .unwrap();
396
397        let persisted = BoxStateStore::load(&path).unwrap();
398        assert_eq!(
399            persisted.records()[0].virtiofs_cache.as_deref(),
400            Some("always")
401        );
402    }
403
404    #[test]
405    fn operation_lookup_ignores_legacy_records_and_finds_managed_intent() {
406        let operation_id = OperationId::new("operation-1").unwrap();
407        let managed = managed_record("managed", operation_id.clone());
408        let mut store =
409            BoxStateStore::from_records("/tmp/boxes.json", vec![record("legacy"), managed]);
410
411        assert_eq!(
412            store.find_by_operation_id(&operation_id).unwrap().id,
413            "managed"
414        );
415        store
416            .find_by_operation_id_mut(&operation_id)
417            .unwrap()
418            .status = "running".to_string();
419        assert_eq!(store.find_by_id("managed").unwrap().status, "running");
420        assert!(store
421            .find_by_operation_id(&OperationId::new("missing").unwrap())
422            .is_none());
423    }
424
425    #[test]
426    fn strict_load_rejects_duplicate_managed_operation_ids() {
427        let directory = tempfile::tempdir().unwrap();
428        let path = directory.path().join("boxes.json");
429        let operation_id = OperationId::new("operation-1").unwrap();
430        let records = vec![
431            managed_record("first", operation_id.clone()),
432            managed_record("second", operation_id),
433        ];
434        std::fs::write(&path, serde_json::to_vec(&records).unwrap()).unwrap();
435
436        let error = BoxStateStore::load(&path).unwrap_err();
437
438        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
439        assert!(error.to_string().contains("duplicate managed operation ID"));
440    }
441
442    #[test]
443    fn transaction_rejects_duplicate_operation_without_changing_disk() {
444        let directory = tempfile::tempdir().unwrap();
445        let path = directory.path().join("boxes.json");
446        let operation_id = OperationId::new("operation-1").unwrap();
447        BoxStateStore::from_records(
448            path.clone(),
449            vec![managed_record("first", operation_id.clone())],
450        )
451        .save()
452        .unwrap();
453
454        let error = BoxStateStore::modify(&path, |store| {
455            store
456                .records_mut()
457                .push(managed_record("second", operation_id));
458            Ok(())
459        })
460        .unwrap_err();
461
462        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
463        let persisted = BoxStateStore::load(&path).unwrap();
464        assert_eq!(persisted.records().len(), 1);
465        assert_eq!(persisted.records()[0].id, "first");
466    }
467
468    #[test]
469    fn strict_load_rejects_managed_plan_drift() {
470        let directory = tempfile::tempdir().unwrap();
471        let path = directory.path().join("boxes.json");
472        let mut record = managed_record("managed", OperationId::new("operation-1").unwrap());
473        record.managed_execution.as_mut().unwrap().plan =
474            a3s_box_core::resolve_execution(&a3s_box_core::BoxConfig::default()).unwrap();
475        std::fs::write(&path, serde_json::to_vec(&vec![record]).unwrap()).unwrap();
476
477        let error = BoxStateStore::load(&path).unwrap_err();
478
479        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
480        assert!(error.to_string().contains("does not match"));
481    }
482
483    #[test]
484    fn strict_load_rejects_unknown_managed_lifecycle_state() {
485        let directory = tempfile::tempdir().unwrap();
486        let path = directory.path().join("boxes.json");
487        let mut record = managed_record("managed", OperationId::new("operation-1").unwrap());
488        record.status = "future-state".to_string();
489        std::fs::write(&path, serde_json::to_vec(&vec![record]).unwrap()).unwrap();
490
491        let error = BoxStateStore::load(&path).unwrap_err();
492
493        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
494        assert!(error
495            .to_string()
496            .contains("unknown managed execution state"));
497    }
498
499    #[test]
500    fn strict_load_rejects_transition_without_matching_pending_operation() {
501        let directory = tempfile::tempdir().unwrap();
502        let path = directory.path().join("boxes.json");
503        let mut record = managed_record("managed", OperationId::new("operation-1").unwrap());
504        record.status = "pausing".to_string();
505        std::fs::write(&path, serde_json::to_vec(&vec![record]).unwrap()).unwrap();
506
507        let error = BoxStateStore::load(&path).unwrap_err();
508
509        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
510        assert!(error.to_string().contains("inconsistent pending operation"));
511    }
512
513    #[cfg(unix)]
514    #[test]
515    fn concurrent_transactions_do_not_lose_records() {
516        let directory = tempfile::tempdir().unwrap();
517        let path = directory.path().join("boxes.json");
518        let handles: Vec<_> = (0..8)
519            .map(|index| {
520                let path = path.clone();
521                std::thread::spawn(move || {
522                    BoxStateStore::modify(&path, |store| {
523                        store.records_mut().push(record(&format!("id-{index}")));
524                        Ok::<(), std::io::Error>(())
525                    })
526                    .unwrap();
527                })
528            })
529            .collect();
530
531        for handle in handles {
532            handle.join().unwrap();
533        }
534
535        let store = BoxStateStore::load(&path).unwrap();
536        assert_eq!(store.records().len(), 8);
537    }
538}
a3s_box_runtime/box_state.rs

a3s_box_runtime/
box_state.rs
