use std::ffi::OsStr;
use std::fs::{self, File};
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use std::sync::{Mutex, MutexGuard};
use chrono::{DateTime, Utc};
use serde::Deserialize;
use tracing::warn;
use uuid::Uuid;
use arcbox_atomic_file::AtomicWriteError;
use super::phase::{
ComputerPhase, ComputerProvisionOutcome, ComputerRecord, ComputerTransition, ExistingProvision,
ProvisionIntent, RECORD_VERSION, classify_existing_provision, validate_record,
};
use crate::error::{ComputerError, Result};
use crate::sandbox::types::IdleAction;
use crate::sandbox::{ComputerSpec, validate_id};
const RECORDS_DIR: &str = "sandbox-records";
pub struct DurableCommit<T> {
pub(in crate::sandbox) value: T,
pub durability_error: Option<String>,
}
impl<T> DurableCommit<T> {
pub fn confirmed(self, operation: &str) -> Result<T> {
match self.durability_error {
Some(error) => Err(ComputerError::Unavailable(format!(
"{operation} is visible but its durability is unconfirmed: {error}"
))),
None => Ok(self.value),
}
}
}
pub struct ComputerRecordStore {
root: PathBuf,
lock: Mutex<()>,
}
impl ComputerRecordStore {
pub fn new(data_dir: &Path) -> Result<Self> {
let root = data_dir.join(RECORDS_DIR);
fs::create_dir_all(&root)?;
fs::set_permissions(&root, fs::Permissions::from_mode(0o700))?;
File::open(data_dir)?.sync_all()?;
File::open(&root)?.sync_all()?;
Ok(Self {
root,
lock: Mutex::new(()),
})
}
#[cfg(test)]
pub(in crate::sandbox) fn load(&self, id: &str) -> Result<Option<ComputerRecord>> {
let _guard = self.lock()?;
self.load_unlocked(id)
}
pub(in crate::sandbox) fn load_all(&self) -> Result<Vec<ComputerRecord>> {
let _guard = self.lock()?;
let mut paths = fs::read_dir(&self.root)?
.map(|entry| entry.map(|entry| entry.path()))
.collect::<std::io::Result<Vec<_>>>()?;
paths.sort();
paths
.into_iter()
.filter(|path| path.extension() == Some(OsStr::new("json")))
.map(|path| {
if !fs::symlink_metadata(&path)?.file_type().is_file() {
return Err(ComputerError::Config(format!(
"sandbox record is not a regular file: {}",
path.display()
)));
}
let id = path.file_stem().and_then(OsStr::to_str).ok_or_else(|| {
ComputerError::Config(format!(
"sandbox record has an invalid file name: {}",
path.display()
))
})?;
self.load_unlocked(id)?.ok_or_else(|| {
ComputerError::Other(format!(
"sandbox record disappeared while loading: {}",
path.display()
))
})
})
.collect()
}
pub(in crate::sandbox) fn replay_provision(
&self,
id: &str,
request_key: &str,
) -> Result<Option<ComputerProvisionOutcome>> {
validate_provision_input(id, request_key)?;
let _guard = self.lock()?;
let Some(record) = self.load_unlocked(id)? else {
return Ok(None);
};
match classify_existing_provision(&record, request_key)? {
ExistingProvision::Pending => Ok(None),
ExistingProvision::Replay => {
self.confirm_root_unlocked("provision replay")?;
Ok(record.provision_outcome)
}
ExistingProvision::Blocked => {
self.confirm_root_unlocked("provision collision")?;
Err(ComputerError::AlreadyExists(id.to_owned()))
}
}
}
pub(crate) fn provision_intent(
&self,
id: &str,
request_key: &str,
effective_spec: ComputerSpec,
) -> Result<ProvisionIntent> {
validate_provision_input(id, request_key)?;
let _guard = self.lock()?;
if let Some(record) = self.load_unlocked(id)? {
return Ok(match classify_existing_provision(&record, request_key)? {
ExistingProvision::Pending => ProvisionIntent::Resume(record),
ExistingProvision::Replay => {
self.confirm_root_unlocked("provision replay")?;
ProvisionIntent::Replay(record)
}
ExistingProvision::Blocked => {
self.confirm_root_unlocked("provision collision")?;
ProvisionIntent::Blocked(record)
}
});
}
let record = ComputerRecord::new(id, request_key, effective_spec);
if let Some(error) = self.save_unlocked(&record)? {
warn!(
sandbox_id = id,
error, "provision intent is visible but directory fsync failed"
);
}
Ok(ProvisionIntent::Created(record))
}
pub fn transition(
&self,
id: &str,
generation: Uuid,
transition: ComputerTransition,
) -> Result<DurableCommit<ComputerRecord>> {
let _guard = self.lock()?;
let mut record = self
.load_unlocked(id)?
.ok_or_else(|| ComputerError::NotFound(id.to_owned()))?;
if record.generation != generation {
return Err(generation_mismatch(id, record.generation, generation));
}
record.apply(transition)?;
let durability_error = self.save_unlocked(&record)?;
Ok(DurableCommit {
value: record,
durability_error,
})
}
pub fn update_lifecycle(
&self,
id: &str,
generation: Uuid,
ttl_deadline: Option<DateTime<Utc>>,
idle_timeout_seconds: u32,
on_idle: IdleAction,
) -> Result<DurableCommit<()>> {
let _guard = self.lock()?;
let mut record = self
.load_unlocked(id)?
.ok_or_else(|| ComputerError::NotFound(id.to_owned()))?;
if record.generation != generation {
return Err(generation_mismatch(id, record.generation, generation));
}
record.ttl_deadline = ttl_deadline;
record.effective_spec.idle_timeout_seconds = idle_timeout_seconds;
record.effective_spec.on_idle = on_idle;
let durability_error = self.save_unlocked(&record)?;
Ok(DurableCommit {
value: (),
durability_error,
})
}
pub fn abort_provision(&self, id: &str, generation: Uuid) -> Result<DurableCommit<()>> {
self.delete_record(id, generation, ComputerPhase::Creating)
}
pub fn finish_remove(&self, id: &str, generation: Uuid) -> Result<DurableCommit<()>> {
self.delete_record(id, generation, ComputerPhase::Removing)
}
pub(in crate::sandbox) fn cancel_pending_or_missing(
&self,
id: &str,
) -> Result<DurableCommit<()>> {
let _guard = self.lock()?;
let Some(record) = self.load_unlocked(id)? else {
return Ok(self.confirm_absence_unlocked());
};
if record.phase != ComputerPhase::Creating || record.provision_outcome.is_some() {
return Err(ComputerError::AlreadyExists(id.to_owned()));
}
self.delete_unlocked(id)
}
fn lock(&self) -> Result<MutexGuard<'_, ()>> {
self.lock
.lock()
.map_err(|_| ComputerError::Other("sandbox record store mutex poisoned".into()))
}
fn load_unlocked(&self, id: &str) -> Result<Option<ComputerRecord>> {
validate_id("sandbox id", id)?;
let path = self.record_path(id);
let metadata = match fs::symlink_metadata(&path) {
Ok(metadata) => metadata,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(error.into()),
};
if !metadata.file_type().is_file() {
return Err(ComputerError::Config(format!(
"sandbox record is not a regular file: {}",
path.display()
)));
}
if metadata.permissions().mode() & 0o077 != 0 {
fs::set_permissions(&path, fs::Permissions::from_mode(0o600))?;
}
let bytes = match fs::read(&path) {
Ok(bytes) => bytes,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(error.into()),
};
#[derive(Deserialize)]
struct RecordHeader {
version: u32,
}
let header: RecordHeader = serde_json::from_slice(&bytes)?;
if header.version != RECORD_VERSION {
return Err(ComputerError::Config(format!(
"unsupported sandbox record version {} for {id}",
header.version
)));
}
let record: ComputerRecord = serde_json::from_slice(&bytes)?;
validate_record(id, &record)?;
Ok(Some(record))
}
fn save_unlocked(&self, record: &ComputerRecord) -> Result<Option<String>> {
validate_record(&record.id, record)?;
let bytes = serde_json::to_vec_pretty(record)?;
match arcbox_atomic_file::write(&self.record_path(&record.id), &bytes) {
Ok(()) => Ok(None),
Err(error @ AtomicWriteError::NotCommitted { .. }) => Err(error.into()),
Err(error @ AtomicWriteError::DurabilityUncertain { .. }) => {
Ok(Some(error.to_string()))
}
}
}
fn record_path(&self, id: &str) -> PathBuf {
self.root.join(format!("{id}.json"))
}
fn delete_record(
&self,
id: &str,
generation: Uuid,
expected_phase: ComputerPhase,
) -> Result<DurableCommit<()>> {
let _guard = self.lock()?;
let Some(record) = self.load_unlocked(id)? else {
return Ok(self.confirm_absence_unlocked());
};
if record.generation != generation {
return Err(generation_mismatch(id, record.generation, generation));
}
if record.phase != expected_phase {
return Err(ComputerError::WrongState {
id: id.to_owned(),
expected: expected_phase.as_str().into(),
actual: record.phase.as_str().into(),
});
}
self.delete_unlocked(id)
}
fn delete_unlocked(&self, id: &str) -> Result<DurableCommit<()>> {
fs::remove_file(self.record_path(id))?;
let durability_error = self
.sync_root_unlocked()
.err()
.map(|error| error.to_string());
Ok(DurableCommit {
value: (),
durability_error,
})
}
fn sync_root_unlocked(&self) -> std::io::Result<()> {
File::open(&self.root).and_then(|directory| directory.sync_all())
}
fn confirm_root_unlocked(&self, operation: &str) -> Result<()> {
self.sync_root_unlocked().map_err(|error| {
ComputerError::Unavailable(format!(
"{operation} durability could not be confirmed: {error}"
))
})
}
fn confirm_absence_unlocked(&self) -> DurableCommit<()> {
DurableCommit {
value: (),
durability_error: self
.sync_root_unlocked()
.err()
.map(|error| error.to_string()),
}
}
}
fn validate_provision_input(id: &str, request_key: &str) -> Result<()> {
validate_id("sandbox id", id)?;
if request_key.is_empty() {
return Err(ComputerError::Config(
"sandbox provision request key must not be empty".into(),
));
}
Ok(())
}
fn generation_mismatch(id: &str, expected: Uuid, actual: Uuid) -> ComputerError {
ComputerError::WrongState {
id: id.to_owned(),
expected: format!("generation {expected}"),
actual: format!("generation {actual}"),
}
}
#[cfg(test)]
mod tests {
use super::*;
fn spec(id: &str) -> ComputerSpec {
ComputerSpec {
id: Some(id.to_owned()),
ttl_seconds: 60,
..ComputerSpec::default()
}
}
fn created(intent: ProvisionIntent) -> ComputerRecord {
match intent {
ProvisionIntent::Created(record) => record,
other => panic!("expected created intent, got {other:?}"),
}
}
#[test]
fn records_survive_reopen_and_ignore_truncated_temporary_files() {
let data_dir = tempfile::tempdir().unwrap();
let store = ComputerRecordStore::new(data_dir.path()).unwrap();
let record = created(store.provision_intent("box", "key", spec("box")).unwrap());
fs::write(store.root.join(".box.interrupted.tmp"), b"{").unwrap();
let reopened = ComputerRecordStore::new(data_dir.path()).unwrap();
assert_eq!(reopened.load("box").unwrap(), Some(record));
assert_eq!(reopened.load("other").unwrap(), None);
assert_eq!(reopened.load_all().unwrap().len(), 1);
assert_eq!(
fs::metadata(reopened.record_path("box"))
.unwrap()
.permissions()
.mode()
& 0o777,
0o600
);
assert_eq!(
fs::metadata(&reopened.root).unwrap().permissions().mode() & 0o777,
0o700
);
}
#[test]
fn ready_restore_outcome_replays_after_store_reopen() {
let data_dir = tempfile::tempdir().unwrap();
let store = ComputerRecordStore::new(data_dir.path()).unwrap();
let record = created(
store
.provision_intent("restored", "restore-key", spec("restored"))
.unwrap(),
);
let outcome = ComputerProvisionOutcome {
ip_address: "192.0.2.8".into(),
};
store
.transition(
"restored",
record.generation,
ComputerTransition::ReadyWithOutcome(outcome.clone()),
)
.unwrap();
let reopened = ComputerRecordStore::new(data_dir.path()).unwrap();
assert_eq!(
reopened
.replay_provision("restored", "restore-key")
.unwrap(),
Some(outcome)
);
assert!(matches!(
reopened.replay_provision("restored", "different-restore-key"),
Err(ComputerError::AlreadyExists(id)) if id == "restored"
));
}
#[test]
fn load_all_rejects_corrupt_and_unsupported_records() {
let data_dir = tempfile::tempdir().unwrap();
let store = ComputerRecordStore::new(data_dir.path()).unwrap();
let mut record = created(store.provision_intent("box", "key", spec("box")).unwrap());
let path = store.record_path("box");
fs::write(&path, b"{").unwrap();
assert!(matches!(store.load_all(), Err(ComputerError::Json(_))));
record.version = RECORD_VERSION + 1;
fs::write(&path, serde_json::to_vec(&record).unwrap()).unwrap();
assert!(matches!(
store.load_all(),
Err(ComputerError::Config(message))
if message.contains("unsupported sandbox record version")
));
}
#[test]
fn matching_intent_resumes_then_replays_but_a_different_key_collides() {
let data_dir = tempfile::tempdir().unwrap();
let store = ComputerRecordStore::new(data_dir.path()).unwrap();
assert_eq!(store.replay_provision("missing", "key").unwrap(), None);
let mut original = spec("box");
original.rootfs = "/first/rootfs.ext4".into();
let record = created(store.provision_intent("box", "key", original).unwrap());
assert_eq!(store.replay_provision("box", "key").unwrap(), None);
let mut refreshed = spec("box");
refreshed.rootfs = "/new/rootfs.ext4".into();
assert!(matches!(
store.provision_intent("box", "key", refreshed).unwrap(),
ProvisionIntent::Resume(found)
if found.generation == record.generation
&& found.effective_spec.rootfs == "/first/rootfs.ext4"
));
assert!(matches!(
store.provision_intent("box", "other", spec("box")),
Err(ComputerError::AlreadyExists(id)) if id == "box"
));
assert!(matches!(
store.replay_provision("box", "other"),
Err(ComputerError::AlreadyExists(id)) if id == "box"
));
let mut with_secret = spec("box");
with_secret.env.insert("TOKEN".into(), "secret".into());
let store2 = ComputerRecordStore::new(data_dir.path()).unwrap();
store2
.provision_intent("secret-box", "secret-key", {
with_secret.id = Some("secret-box".into());
with_secret
})
.unwrap();
let secret_record = store2.load("secret-box").unwrap().unwrap();
store2
.transition(
"secret-box",
secret_record.generation,
ComputerTransition::Failed("boot failed".into()),
)
.unwrap();
assert!(
store2
.load("secret-box")
.unwrap()
.unwrap()
.effective_spec
.env
.is_empty()
);
store
.transition(
"box",
record.generation,
ComputerTransition::Starting(ComputerProvisionOutcome {
ip_address: "192.0.2.2".into(),
}),
)
.unwrap();
assert_eq!(
store.replay_provision("box", "key").unwrap(),
Some(ComputerProvisionOutcome {
ip_address: "192.0.2.2".into()
})
);
assert!(matches!(
store.provision_intent("box", "key", spec("box")).unwrap(),
ProvisionIntent::Replay(found)
if found.provision_outcome == Some(ComputerProvisionOutcome {
ip_address: "192.0.2.2".into()
})
));
store
.transition(
"box",
record.generation,
ComputerTransition::Failed("boot failed".into()),
)
.unwrap();
assert!(matches!(
store.replay_provision("box", "key"),
Err(ComputerError::AlreadyExists(id)) if id == "box"
));
assert!(matches!(
store.provision_intent("box", "key", spec("box")).unwrap(),
ProvisionIntent::Blocked(found) if found.phase == ComputerPhase::Failed
));
}
#[test]
fn transitions_enforce_generation_and_lifecycle_edges() {
let data_dir = tempfile::tempdir().unwrap();
let store = ComputerRecordStore::new(data_dir.path()).unwrap();
let record = created(store.provision_intent("box", "key", spec("box")).unwrap());
assert!(
store
.transition("box", Uuid::new_v4(), ComputerTransition::Ready)
.is_err()
);
assert!(
store
.transition("box", record.generation, ComputerTransition::Ready)
.is_err()
);
assert_eq!(
store.load("box").unwrap().unwrap().phase,
ComputerPhase::Creating
);
store
.transition(
"box",
record.generation,
ComputerTransition::Starting(ComputerProvisionOutcome {
ip_address: String::new(),
}),
)
.unwrap();
store
.transition("box", record.generation, ComputerTransition::Ready)
.unwrap();
assert!(store.replay_provision("box", "key").unwrap().is_some());
store
.transition("box", record.generation, ComputerTransition::Stopping)
.unwrap();
assert!(matches!(
store.replay_provision("box", "key"),
Err(ComputerError::AlreadyExists(id)) if id == "box"
));
store
.transition("box", record.generation, ComputerTransition::Stopped)
.unwrap();
assert!(matches!(
store.replay_provision("box", "key"),
Err(ComputerError::AlreadyExists(id)) if id == "box"
));
store
.transition("box", record.generation, ComputerTransition::Stopped)
.unwrap()
.confirmed("idempotent stopped retry")
.unwrap();
store
.transition("box", record.generation, ComputerTransition::Removing)
.unwrap();
assert!(matches!(
store.replay_provision("box", "key"),
Err(ComputerError::AlreadyExists(id)) if id == "box"
));
store.finish_remove("box", record.generation).unwrap();
assert_eq!(store.load("box").unwrap(), None);
assert_eq!(store.replay_provision("box", "key").unwrap(), None);
let recreated = created(
store
.provision_intent("box", "new-key", spec("box"))
.unwrap(),
);
assert_ne!(recreated.generation, record.generation);
let failed = created(
store
.provision_intent("failed", "key", spec("failed"))
.unwrap(),
);
store
.transition(
"failed",
failed.generation,
ComputerTransition::Failed("interrupted".into()),
)
.unwrap();
assert!(matches!(
store.replay_provision("failed", "key"),
Err(ComputerError::AlreadyExists(id)) if id == "failed"
));
}
#[test]
fn pause_resume_transitions_carry_the_snapshot_and_paused_at() {
let data_dir = tempfile::tempdir().unwrap();
let store = ComputerRecordStore::new(data_dir.path()).unwrap();
let record = created(store.provision_intent("box", "key", spec("box")).unwrap());
let generation = record.generation;
store
.transition(
"box",
generation,
ComputerTransition::Starting(ComputerProvisionOutcome {
ip_address: "192.0.2.2".into(),
}),
)
.unwrap();
store
.transition("box", generation, ComputerTransition::Ready)
.unwrap();
assert!(
store
.transition(
"box",
generation,
ComputerTransition::Paused {
snapshot_id: "snap".into()
}
)
.is_err()
);
store
.transition("box", generation, ComputerTransition::Pausing)
.unwrap();
let paused = store
.transition(
"box",
generation,
ComputerTransition::Paused {
snapshot_id: "snap".into(),
},
)
.unwrap()
.value;
assert_eq!(paused.phase, ComputerPhase::Paused);
assert_eq!(paused.pause_snapshot_id.as_deref(), Some("snap"));
assert!(paused.paused_at.is_some());
assert!(matches!(
store.replay_provision("box", "key"),
Err(ComputerError::AlreadyExists(_))
));
let resuming = store
.transition("box", generation, ComputerTransition::Resuming)
.unwrap()
.value;
assert_eq!(resuming.pause_snapshot_id.as_deref(), Some("snap"));
let reverted = store
.transition(
"box",
generation,
ComputerTransition::Paused {
snapshot_id: "snap".into(),
},
)
.unwrap()
.value;
assert_eq!(reverted.paused_at, paused.paused_at);
store
.transition("box", generation, ComputerTransition::Resuming)
.unwrap();
let ready = store
.transition("box", generation, ComputerTransition::Ready)
.unwrap()
.value;
assert_eq!(ready.phase, ComputerPhase::Ready);
assert_eq!(ready.pause_snapshot_id, None);
assert_eq!(ready.paused_at, None);
store
.transition("box", generation, ComputerTransition::Pausing)
.unwrap();
let repaused = store
.transition(
"box",
generation,
ComputerTransition::Paused {
snapshot_id: "snap2".into(),
},
)
.unwrap()
.value;
assert!(repaused.paused_at.is_some());
assert!(repaused.paused_at >= paused.paused_at);
}
#[test]
fn failed_pause_reverts_to_ready_and_paused_cannot_stop() {
let data_dir = tempfile::tempdir().unwrap();
let store = ComputerRecordStore::new(data_dir.path()).unwrap();
let record = created(store.provision_intent("box", "key", spec("box")).unwrap());
let generation = record.generation;
store
.transition(
"box",
generation,
ComputerTransition::Starting(ComputerProvisionOutcome {
ip_address: String::new(),
}),
)
.unwrap();
store
.transition("box", generation, ComputerTransition::Ready)
.unwrap();
store
.transition("box", generation, ComputerTransition::Pausing)
.unwrap();
store
.transition("box", generation, ComputerTransition::Ready)
.unwrap();
store
.transition("box", generation, ComputerTransition::Pausing)
.unwrap();
store
.transition(
"box",
generation,
ComputerTransition::Paused {
snapshot_id: "snap".into(),
},
)
.unwrap();
assert!(
store
.transition("box", generation, ComputerTransition::Stopping)
.is_err()
);
store
.transition("box", generation, ComputerTransition::Removing)
.unwrap();
store.finish_remove("box", generation).unwrap();
}
#[test]
fn pre_ack_abort_releases_only_the_expected_generation() {
let data_dir = tempfile::tempdir().unwrap();
let store = ComputerRecordStore::new(data_dir.path()).unwrap();
let record = created(store.provision_intent("box", "key", spec("box")).unwrap());
assert!(store.abort_provision("box", Uuid::new_v4()).is_err());
store
.abort_provision("box", record.generation)
.unwrap()
.confirmed("test abort")
.unwrap();
assert!(store.load("box").unwrap().is_none());
store
.cancel_pending_or_missing("box")
.unwrap()
.confirmed("idempotent remove")
.unwrap();
}
}