use std::collections::{HashMap, HashSet};
use std::fs;
use std::io::{self, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, LazyLock, Mutex};
use std::time::Duration;
use serde::{Deserialize, Serialize};
use crate::backup::{hash_session, BackupStore, CapturedRegularFile};
use crate::error::AftError;
use crate::fs_lock;
const CHECKPOINT_LOCK_TIMEOUT: Duration = Duration::from_secs(30);
const MAX_NAMED_CHECKPOINTS_PER_SESSION: usize = 20;
const NAMED_CHECKPOINT_RETENTION_DAYS: u64 = 14;
const NAMED_CHECKPOINT_RETENTION_SECS: u64 = NAMED_CHECKPOINT_RETENTION_DAYS * 24 * 60 * 60;
const CHECKPOINT_SCHEMA_VERSION: u32 = 1;
const UNBOUND_HARNESS_SEGMENT: &str = "unbound";
static CHECKPOINT_MAINTENANCE_KEYS: LazyLock<Mutex<HashSet<(PathBuf, String)>>> =
LazyLock::new(|| Mutex::new(HashSet::new()));
pub const CHECKPOINT_RESTART_NOTICE: &str =
"no durable checkpoints found on disk; in-memory checkpoints do not survive restarts";
pub const CHECKPOINT_HYDRATED_NOTICE: &str =
"durable checkpoints are hydrated from disk and survive restarts";
pub fn checkpoint_durability(storage_path: &Path) -> String {
format!(
"durable on disk at {}; survives restarts",
storage_path.display()
)
}
#[derive(Debug, Clone)]
pub struct CheckpointInfo {
pub name: String,
pub file_count: usize,
pub created_at: u64,
pub storage_path: Option<PathBuf>,
pub evicted: Vec<String>,
pub skipped: Vec<(PathBuf, String)>,
}
#[derive(Debug, Clone)]
struct Checkpoint {
name: String,
file_contents: HashMap<PathBuf, CheckpointFile>,
created_at: u64,
created_order: u64,
}
#[derive(Debug, Clone)]
struct CheckpointFile {
metadata: Option<fs::Metadata>,
mode: Option<u32>,
kind: CheckpointFileKind,
}
#[derive(Debug, Serialize, Deserialize)]
struct DiskCheckpointMeta {
schema_version: u32,
session_id: String,
name: String,
created_at: u64,
created_order: u64,
files: Vec<DiskCheckpointFileMeta>,
}
#[derive(Debug, Serialize, Deserialize)]
struct DiskCheckpointFileMeta {
original_path: String,
blob: String,
kind: DiskCheckpointFileKind,
mode: Option<u32>,
target_is_dir: bool,
}
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
enum DiskCheckpointFileKind {
Regular,
Symlink,
}
#[derive(Debug, Clone)]
enum CheckpointFileKind {
Regular {
bytes: Arc<[u8]>,
},
Symlink {
target: PathBuf,
target_is_dir: bool,
},
}
impl CheckpointFile {
fn read(path: &Path) -> io::Result<Self> {
let metadata = fs::symlink_metadata(path)?;
let file_type = metadata.file_type();
if file_type.is_symlink() {
let target = fs::read_link(path)?;
let target_is_dir = fs::metadata(path)
.map(|target_metadata| target_metadata.is_dir())
.unwrap_or(false);
return Ok(Self {
mode: checkpoint_mode(&metadata),
metadata: Some(metadata),
kind: CheckpointFileKind::Symlink {
target,
target_is_dir,
},
});
}
if metadata.is_file() {
let capture = CapturedRegularFile::read(path)?.ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidInput,
"file changed while being captured",
)
})?;
return Ok(Self::from_fresh_capture(capture));
}
Err(io::Error::new(
io::ErrorKind::InvalidInput,
"not a regular file or symlink",
))
}
fn from_captured(path: &Path, capture: &mut CapturedRegularFile) -> io::Result<Self> {
capture.refresh_if_stale(path)?;
let metadata = capture.metadata().clone();
Ok(Self {
mode: checkpoint_mode(&metadata),
metadata: Some(metadata),
kind: CheckpointFileKind::Regular {
bytes: capture.shared_bytes(),
},
})
}
fn from_fresh_capture(capture: CapturedRegularFile) -> Self {
let metadata = capture.metadata().clone();
Self {
mode: checkpoint_mode(&metadata),
metadata: Some(metadata),
kind: CheckpointFileKind::Regular {
bytes: capture.shared_bytes(),
},
}
}
fn read_optional(path: &Path) -> io::Result<Option<Self>> {
match Self::read(path) {
Ok(snapshot) => Ok(Some(snapshot)),
Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(None),
Err(error) => Err(error),
}
}
fn from_disk(meta: &DiskCheckpointFileMeta, bytes: Vec<u8>) -> Result<Self, String> {
let kind = match &meta.kind {
DiskCheckpointFileKind::Regular => CheckpointFileKind::Regular {
bytes: bytes.into(),
},
DiskCheckpointFileKind::Symlink => {
let target = String::from_utf8(bytes)
.map(PathBuf::from)
.map_err(|error| format!("checkpoint symlink target is not UTF-8: {error}"))?;
CheckpointFileKind::Symlink {
target,
target_is_dir: meta.target_is_dir,
}
}
};
Ok(Self {
metadata: None,
mode: meta.mode,
kind,
})
}
}
#[cfg(unix)]
fn checkpoint_mode(metadata: &fs::Metadata) -> Option<u32> {
use std::os::unix::fs::PermissionsExt;
Some(metadata.permissions().mode())
}
#[cfg(not(unix))]
fn checkpoint_mode(_metadata: &fs::Metadata) -> Option<u32> {
None
}
#[derive(Debug)]
pub struct CheckpointStore {
checkpoints: HashMap<String, HashMap<String, Checkpoint>>,
lock_path: PathBuf,
lock_timeout: Duration,
storage_dir: Option<PathBuf>,
storage_harness: Option<String>,
blob_counter: AtomicU64,
}
struct CheckpointLockGuard {
guard: Option<fs_lock::LockGuard>,
scope_dir: Option<PathBuf>,
}
impl Drop for CheckpointLockGuard {
fn drop(&mut self) {
if let Some(guard) = self.guard.take() {
drop(guard);
}
if let Some(scope_dir) = &self.scope_dir {
remove_empty_scope_dir(scope_dir);
}
}
}
impl CheckpointStore {
pub fn new() -> Self {
let project_root = std::env::current_dir().unwrap_or_else(|_| std::env::temp_dir());
let project_key = crate::path_identity::project_scope_key(&project_root);
let storage_dir = crate::bash_background::storage_dir(None);
let lock_path = storage_dir
.join("checkpoints")
.join(project_key)
.join("checkpoint.lock");
let mut store = Self::with_lock_path(lock_path, CHECKPOINT_LOCK_TIMEOUT);
store.storage_dir = Some(storage_dir);
store.storage_harness = Some(UNBOUND_HARNESS_SEGMENT.to_string());
store
}
#[cfg(test)]
pub(crate) fn set_lock_path_for_test(&mut self, lock_path: PathBuf) {
self.storage_dir = lock_path.parent().map(Path::to_path_buf);
self.storage_harness = Some("test".to_string());
self.lock_path = lock_path;
}
pub fn set_storage_dir_for_harness(&mut self, dir: PathBuf, harness: crate::harness::Harness) {
let harness = harness.storage_segment();
if self.storage_dir.as_ref() == Some(&dir)
&& self.storage_harness.as_deref() == Some(&harness)
{
return;
}
if self.storage_dir.as_ref() == Some(&dir)
&& self.storage_harness.as_deref() == Some(UNBOUND_HARNESS_SEGMENT)
{
match self.acquire_mutation_lock() {
Ok(_lock) => migrate_unbound_checkpoint_namespace(&dir, &harness),
Err(error) => crate::slog_warn!(
"could not migrate unbound durable checkpoints into {}: {}",
harness,
error
),
}
}
self.storage_dir = Some(dir);
self.storage_harness = Some(harness);
self.checkpoints.clear();
}
fn with_lock_path(lock_path: PathBuf, lock_timeout: Duration) -> Self {
CheckpointStore {
checkpoints: HashMap::new(),
lock_path,
lock_timeout,
storage_dir: None,
storage_harness: None,
blob_counter: AtomicU64::new(0),
}
}
fn acquire_mutation_lock(&self) -> Result<CheckpointLockGuard, AftError> {
let scope_dir = self.lock_path.parent().map(Path::to_path_buf);
if let Some(parent) = scope_dir.as_deref() {
fs::create_dir_all(parent).map_err(|error| AftError::IoError {
path: parent.display().to_string(),
message: format!("failed to create checkpoint lock directory: {error}"),
})?;
}
let acquire_result = match fs_lock::try_acquire(&self.lock_path, self.lock_timeout) {
Err(fs_lock::AcquireError::Io(error)) if error.kind() == io::ErrorKind::NotFound => {
if let Some(parent) = scope_dir.as_deref() {
fs::create_dir_all(parent).map_err(|error| AftError::IoError {
path: parent.display().to_string(),
message: format!("failed to recreate checkpoint lock directory: {error}"),
})?;
}
fs_lock::try_acquire(&self.lock_path, self.lock_timeout)
}
result => result,
};
let guard = acquire_result.map_err(|error| match error {
fs_lock::AcquireError::Timeout => AftError::IoError {
path: self.lock_path.display().to_string(),
message: "timed out acquiring checkpoint mutation lock".to_string(),
},
fs_lock::AcquireError::Io(error) => AftError::IoError {
path: self.lock_path.display().to_string(),
message: format!("failed to acquire checkpoint mutation lock: {error}"),
},
})?;
Ok(CheckpointLockGuard {
guard: Some(guard),
scope_dir,
})
}
pub fn create(
&mut self,
session: &str,
name: &str,
files: Vec<PathBuf>,
backup_store: &BackupStore,
) -> Result<CheckpointInfo, AftError> {
self.create_impl(session, name, files, backup_store, None)
}
pub(crate) fn create_from_captures(
&mut self,
session: &str,
name: &str,
files: Vec<PathBuf>,
backup_store: &BackupStore,
captures: &mut HashMap<PathBuf, CapturedRegularFile>,
) -> Result<CheckpointInfo, AftError> {
self.create_impl(session, name, files, backup_store, Some(captures))
}
fn create_impl(
&mut self,
session: &str,
name: &str,
files: Vec<PathBuf>,
backup_store: &BackupStore,
mut captures: Option<&mut HashMap<PathBuf, CapturedRegularFile>>,
) -> Result<CheckpointInfo, AftError> {
let _mutation_lock = self.acquire_mutation_lock()?;
validate_checkpoint_name(name)?;
self.run_process_maintenance_once_locked()?;
self.hydrate_session_locked(session)?;
let explicit_request = !files.is_empty();
let file_list = if files.is_empty() {
backup_store.tracked_files(session)
} else {
files
};
let mut file_contents = HashMap::new();
let mut skipped: Vec<(PathBuf, String)> = Vec::new();
for path in &file_list {
let seeded = captures
.as_deref_mut()
.and_then(|captures| captures.get_mut(path))
.map(|capture| CheckpointFile::from_captured(path, capture));
let snapshot = match seeded {
Some(Err(error)) if error.kind() == io::ErrorKind::InvalidInput => {
if let Some(captures) = captures.as_deref_mut() {
captures.remove(path);
}
CheckpointFile::read(path)
}
Some(result) => result,
None => CheckpointFile::read(path),
};
match snapshot {
Ok(snapshot) => {
file_contents.insert(path.clone(), snapshot);
}
Err(e) => {
crate::slog_warn!(
"checkpoint {}: skipping unreadable file {}: {}",
name,
path.display(),
e
);
skipped.push((path.clone(), e.to_string()));
}
}
}
if explicit_request && file_contents.is_empty() && !skipped.is_empty() {
let (path, err) = &skipped[0];
return Err(AftError::FileNotFound {
path: format!("{}: {}", path.display(), err),
});
}
let created_at = current_timestamp();
let created_order = current_timestamp_nanos()
.saturating_add(self.blob_counter.fetch_add(1, Ordering::Relaxed));
let file_count = file_contents.len();
let checkpoint = Checkpoint {
name: name.to_string(),
file_contents,
created_at,
created_order,
};
let storage_path = self.durable_checkpoint_dir(session, name);
self.persist_checkpoint_locked(session, &checkpoint)?;
self.checkpoints
.entry(session.to_string())
.or_default()
.insert(name.to_string(), checkpoint);
let evicted = self.evict_excess_checkpoints_locked(session)?;
if skipped.is_empty() {
crate::slog_info!("checkpoint created: {} ({} files)", name, file_count);
} else {
crate::slog_info!(
"checkpoint created: {} ({} files, {} skipped)",
name,
file_count,
skipped.len()
);
}
Ok(CheckpointInfo {
name: name.to_string(),
file_count,
created_at,
storage_path,
evicted,
skipped,
})
}
pub fn restore(&mut self, session: &str, name: &str) -> Result<CheckpointInfo, AftError> {
let _mutation_lock = self.acquire_mutation_lock()?;
self.run_process_maintenance_once_locked()?;
self.hydrate_session_locked(session)?;
let storage_path = self.durable_checkpoint_dir(session, name);
let checkpoint = self.get(session, name)?;
let mut paths = checkpoint.file_contents.keys().cloned().collect::<Vec<_>>();
paths.sort();
restore_paths_atomically(checkpoint, &paths)?;
crate::slog_info!("checkpoint restored: {}", name);
Ok(CheckpointInfo {
name: checkpoint.name.clone(),
file_count: checkpoint.file_contents.len(),
created_at: checkpoint.created_at,
storage_path,
evicted: Vec::new(),
skipped: Vec::new(),
})
}
pub fn restore_validated(
&mut self,
session: &str,
name: &str,
validated_paths: &[PathBuf],
) -> Result<CheckpointInfo, AftError> {
let _mutation_lock = self.acquire_mutation_lock()?;
self.run_process_maintenance_once_locked()?;
self.hydrate_session_locked(session)?;
let storage_path = self.durable_checkpoint_dir(session, name);
let checkpoint = self.get(session, name)?;
for path in validated_paths {
checkpoint
.file_contents
.get(path)
.ok_or_else(|| AftError::FileNotFound {
path: path.display().to_string(),
})?;
}
restore_paths_atomically(checkpoint, validated_paths)?;
crate::slog_info!("checkpoint restored: {}", name);
Ok(CheckpointInfo {
name: checkpoint.name.clone(),
file_count: checkpoint.file_contents.len(),
created_at: checkpoint.created_at,
storage_path,
evicted: Vec::new(),
skipped: Vec::new(),
})
}
pub fn file_paths(&mut self, session: &str, name: &str) -> Result<Vec<PathBuf>, AftError> {
let _mutation_lock = self.acquire_mutation_lock()?;
self.run_process_maintenance_once_locked()?;
self.hydrate_session_locked(session)?;
let checkpoint = self.get(session, name)?;
Ok(checkpoint.file_contents.keys().cloned().collect())
}
pub fn absolute_file_paths(
&mut self,
session: &str,
name: &str,
) -> Result<Vec<PathBuf>, AftError> {
let mut paths: Vec<PathBuf> = self
.file_paths(session, name)?
.into_iter()
.map(absolute_checkpoint_path)
.collect();
paths.sort();
Ok(paths)
}
pub fn delete(&mut self, session: &str, name: &str) -> bool {
let _mutation_lock = match self.acquire_mutation_lock() {
Ok(lock) => lock,
Err(error) => {
crate::slog_warn!("checkpoint delete lock failed for {}: {}", name, error);
return false;
}
};
if let Err(error) = self.run_process_maintenance_once_locked() {
crate::slog_warn!(
"checkpoint delete maintenance failed for {}: {}",
name,
error
);
return false;
}
if let Err(error) = self.hydrate_session_locked(session) {
crate::slog_warn!("checkpoint delete hydration failed for {}: {}", name, error);
return false;
}
if self
.checkpoints
.get(session)
.is_none_or(|checkpoints| !checkpoints.contains_key(name))
{
return false;
}
if let Err(error) = self.remove_checkpoint_from_disk_locked(session, name) {
crate::slog_warn!("checkpoint delete failed for {}: {}", name, error);
return false;
}
let Some(session_checkpoints) = self.checkpoints.get_mut(session) else {
return false;
};
let removed = session_checkpoints.remove(name).is_some();
if session_checkpoints.is_empty() {
self.checkpoints.remove(session);
}
removed
}
pub fn list(&mut self, session: &str) -> Result<Vec<CheckpointInfo>, AftError> {
let _mutation_lock = self.acquire_mutation_lock()?;
self.run_process_maintenance_once_locked()?;
self.hydrate_session_locked(session)?;
let mut list = self
.checkpoints
.get(session)
.map(|checkpoints| {
checkpoints
.values()
.map(|checkpoint| CheckpointInfo {
name: checkpoint.name.clone(),
file_count: checkpoint.file_contents.len(),
created_at: checkpoint.created_at,
storage_path: self.durable_checkpoint_dir(session, &checkpoint.name),
evicted: Vec::new(),
skipped: Vec::new(),
})
.collect::<Vec<_>>()
})
.unwrap_or_default();
list.sort_by(|left, right| left.name.cmp(&right.name));
Ok(list)
}
pub fn total_count(&self) -> usize {
self.checkpoints
.values()
.map(|checkpoints| checkpoints.len())
.sum()
}
pub fn cleanup(&mut self) {
let _mutation_lock = match self.acquire_mutation_lock() {
Ok(lock) => lock,
Err(error) => {
crate::slog_warn!("checkpoint cleanup lock failed: {}", error);
return;
}
};
if let Err(error) = self.cleanup_locked() {
crate::slog_warn!("checkpoint cleanup failed: {}", error);
}
}
fn get(&self, session: &str, name: &str) -> Result<&Checkpoint, AftError> {
self.checkpoints
.get(session)
.and_then(|checkpoints| checkpoints.get(name))
.ok_or_else(|| AftError::CheckpointNotFound {
name: name.to_string(),
})
}
fn durable_checkpoints_dir(&self) -> Option<PathBuf> {
self.storage_dir
.as_ref()
.zip(self.storage_harness.as_ref())
.map(|(storage_dir, harness)| storage_dir.join(harness).join("checkpoints"))
}
fn durable_session_dir(&self, session: &str) -> Option<PathBuf> {
self.durable_checkpoints_dir()
.map(|checkpoints_dir| checkpoints_dir.join(hash_session(session)))
}
fn durable_checkpoint_dir(&self, session: &str, name: &str) -> Option<PathBuf> {
self.durable_session_dir(session)
.map(|session_dir| session_dir.join(name))
}
fn run_process_maintenance_once_locked(&mut self) -> Result<(), AftError> {
let Some(storage_dir) = self.storage_dir.clone() else {
return Ok(());
};
let Some(harness) = self.storage_harness.clone() else {
return Ok(());
};
if !CHECKPOINT_MAINTENANCE_KEYS
.lock()
.unwrap()
.insert((storage_dir, harness))
{
return Ok(());
}
self.cleanup_locked()
}
fn cleanup_locked(&mut self) -> Result<(), AftError> {
let now = current_timestamp();
self.checkpoints.retain(|_, session_checkpoints| {
session_checkpoints.retain(|_, checkpoint| {
now.saturating_sub(checkpoint.created_at) < NAMED_CHECKPOINT_RETENTION_SECS
});
!session_checkpoints.is_empty()
});
if let Some(checkpoints_dir) = self.durable_checkpoints_dir() {
sweep_expired_durable_checkpoints(&checkpoints_dir, now);
}
if let Some(checkpoints_root) = self.lock_path.parent().and_then(Path::parent) {
if checkpoints_root.file_name() == Some(std::ffi::OsStr::new("checkpoints")) {
sweep_empty_scope_dirs(checkpoints_root);
}
}
Ok(())
}
fn hydrate_session_locked(&mut self, session: &str) -> Result<(), AftError> {
let Some(session_dir) = self.durable_session_dir(session) else {
return Ok(());
};
if !session_dir.exists() {
self.checkpoints.remove(session);
return Ok(());
}
let entries = fs::read_dir(&session_dir).map_err(|error| AftError::IoError {
path: session_dir.display().to_string(),
message: format!("failed to read durable checkpoint session: {error}"),
})?;
let mut hydrated = HashMap::new();
for entry in entries {
let entry = entry.map_err(|error| AftError::IoError {
path: session_dir.display().to_string(),
message: format!("failed to read durable checkpoint entry: {error}"),
})?;
let checkpoint_dir = entry.path();
if !entry
.file_type()
.map_err(|error| AftError::IoError {
path: checkpoint_dir.display().to_string(),
message: format!("failed to inspect durable checkpoint entry: {error}"),
})?
.is_dir()
{
continue;
}
let name = entry.file_name().to_string_lossy().into_owned();
if !is_safe_checkpoint_name(&name) {
continue;
}
let meta_path = checkpoint_dir.join("meta.json");
if !meta_path.exists() {
continue;
}
let checkpoint = read_checkpoint_from_disk(&checkpoint_dir, session, &name)?;
hydrated.insert(name, checkpoint);
}
if hydrated.is_empty() {
self.checkpoints.remove(session);
} else {
self.checkpoints.insert(session.to_string(), hydrated);
self.evict_excess_checkpoints_locked(session)?;
}
Ok(())
}
fn persist_checkpoint_locked(
&self,
session: &str,
checkpoint: &Checkpoint,
) -> Result<(), AftError> {
let Some(checkpoint_dir) = self.durable_checkpoint_dir(session, &checkpoint.name) else {
return Ok(());
};
fs::create_dir_all(&checkpoint_dir).map_err(|error| AftError::IoError {
path: checkpoint_dir.display().to_string(),
message: format!("failed to create durable checkpoint directory: {error}"),
})?;
let mut files = Vec::with_capacity(checkpoint.file_contents.len());
for (index, (path, file)) in checkpoint.file_contents.iter().enumerate() {
let blob = format!(
"file_{}_{}_{}.blob",
checkpoint.created_order,
index,
self.blob_counter.fetch_add(1, Ordering::Relaxed)
);
let bytes = checkpoint_file_bytes(file);
write_temp_fsync_rename(&checkpoint_dir, &blob, &bytes).map_err(|error| {
AftError::IoError {
path: checkpoint_dir.join(&blob).display().to_string(),
message: format!("failed to write durable checkpoint blob: {error}"),
}
})?;
files.push(DiskCheckpointFileMeta {
original_path: path.display().to_string(),
blob,
kind: match &file.kind {
CheckpointFileKind::Regular { .. } => DiskCheckpointFileKind::Regular,
CheckpointFileKind::Symlink { .. } => DiskCheckpointFileKind::Symlink,
},
mode: file.mode,
target_is_dir: matches!(
&file.kind,
CheckpointFileKind::Symlink {
target_is_dir: true,
..
}
),
});
}
fsync_dir(&checkpoint_dir).map_err(|error| AftError::IoError {
path: checkpoint_dir.display().to_string(),
message: format!("failed to sync durable checkpoint blobs: {error}"),
})?;
let meta = DiskCheckpointMeta {
schema_version: CHECKPOINT_SCHEMA_VERSION,
session_id: session.to_string(),
name: checkpoint.name.clone(),
created_at: checkpoint.created_at,
created_order: checkpoint.created_order,
files,
};
let bytes = serde_json::to_vec_pretty(&meta).map_err(|error| AftError::IoError {
path: checkpoint_dir.join("meta.json").display().to_string(),
message: format!("failed to serialize durable checkpoint metadata: {error}"),
})?;
write_temp_fsync_rename(&checkpoint_dir, "meta.json", &bytes).map_err(|error| {
AftError::IoError {
path: checkpoint_dir.join("meta.json").display().to_string(),
message: format!("failed to write durable checkpoint metadata: {error}"),
}
})?;
fsync_dir(&checkpoint_dir).map_err(|error| AftError::IoError {
path: checkpoint_dir.display().to_string(),
message: format!("failed to sync durable checkpoint metadata: {error}"),
})?;
prune_unreferenced_checkpoint_blobs(&checkpoint_dir, &meta.files).map_err(|error| {
AftError::IoError {
path: checkpoint_dir.display().to_string(),
message: format!("failed to prune stale durable checkpoint blobs: {error}"),
}
})?;
Ok(())
}
fn remove_checkpoint_from_disk_locked(
&self,
session: &str,
name: &str,
) -> Result<(), AftError> {
let Some(checkpoint_dir) = self.durable_checkpoint_dir(session, name) else {
return Ok(());
};
match fs::remove_dir_all(&checkpoint_dir) {
Ok(()) => {
if let Some(session_dir) = checkpoint_dir.parent() {
let _ = fs::remove_dir(session_dir);
}
Ok(())
}
Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
Err(error) => Err(AftError::IoError {
path: checkpoint_dir.display().to_string(),
message: format!("failed to remove durable checkpoint: {error}"),
}),
}
}
fn evict_excess_checkpoints_locked(&mut self, session: &str) -> Result<Vec<String>, AftError> {
let Some(checkpoints) = self.checkpoints.get(session) else {
return Ok(Vec::new());
};
let overflow = checkpoints
.len()
.saturating_sub(MAX_NAMED_CHECKPOINTS_PER_SESSION);
let mut checkpoints = checkpoints
.values()
.map(|checkpoint| (checkpoint.created_order, checkpoint.name.clone()))
.collect::<Vec<_>>();
checkpoints.sort();
let evicted = checkpoints
.into_iter()
.take(overflow)
.map(|(_, name)| name)
.collect::<Vec<_>>();
for name in &evicted {
self.remove_checkpoint_from_disk_locked(session, name)?;
}
if let Some(checkpoints) = self.checkpoints.get_mut(session) {
for name in &evicted {
checkpoints.remove(name);
}
}
Ok(evicted)
}
pub fn session_is_empty(&self, session: &str) -> bool {
self.checkpoints.get(session).is_none_or(HashMap::is_empty)
}
}
fn migrate_unbound_checkpoint_namespace(storage_dir: &Path, harness: &str) {
let source = storage_dir
.join(UNBOUND_HARNESS_SEGMENT)
.join("checkpoints");
if !source.exists() {
return;
}
let target = storage_dir.join(harness).join("checkpoints");
if !target.exists() {
if let Some(parent) = target.parent() {
if let Err(error) = fs::create_dir_all(parent) {
crate::slog_warn!(
"failed to create durable checkpoint harness directory {}: {}",
parent.display(),
error
);
return;
}
}
if let Err(error) = fs::rename(&source, &target) {
crate::slog_warn!(
"failed to move unbound durable checkpoints into {}: {}",
target.display(),
error
);
}
return;
}
let Ok(boundary) = crate::walk_boundary::DeviceBoundary::for_root(&source) else {
crate::slog_warn!(
"cannot establish filesystem boundary for checkpoint migration {}",
source.display()
);
return;
};
let mut skipped_foreign_mounts = 0usize;
let Ok(session_entries) = fs::read_dir(&source) else {
return;
};
for session_entry in session_entries.flatten() {
let source_session = session_entry.path();
if !source_session.is_dir() {
continue;
}
if !boundary.should_descend(&source_session).unwrap_or(false) {
skipped_foreign_mounts += 1;
continue;
}
let target_session = target.join(session_entry.file_name());
if !target_session.exists() {
let _ = fs::rename(&source_session, &target_session);
continue;
}
let Ok(checkpoint_entries) = fs::read_dir(&source_session) else {
continue;
};
for checkpoint_entry in checkpoint_entries.flatten() {
let source_checkpoint = checkpoint_entry.path();
let target_checkpoint = target_session.join(checkpoint_entry.file_name());
if !target_checkpoint.exists() {
let _ = fs::rename(source_checkpoint, target_checkpoint);
}
}
let _ = fs::remove_dir(&source_session);
}
let _ = fs::remove_dir(&source);
if skipped_foreign_mounts > 0 {
crate::slog_warn!(
"checkpoint migration skipped {} foreign filesystem mount(s) below {}",
skipped_foreign_mounts,
source.display()
);
}
}
fn validate_checkpoint_name(name: &str) -> Result<(), AftError> {
if is_safe_checkpoint_name(name) {
Ok(())
} else {
Err(AftError::InvalidRequest {
message: "checkpoint name must be a single non-empty path component".to_string(),
})
}
}
fn is_safe_checkpoint_name(name: &str) -> bool {
matches!(
Path::new(name).components().collect::<Vec<_>>().as_slice(),
[std::path::Component::Normal(_)]
) && !name.chars().any(|character| {
character.is_control()
|| matches!(character, '\\' | ':' | '*' | '?' | '"' | '<' | '>' | '|')
})
}
fn is_safe_blob_name(name: &str) -> bool {
is_safe_checkpoint_name(name) && name.starts_with("file_") && name.ends_with(".blob")
}
fn read_checkpoint_from_disk(
checkpoint_dir: &Path,
session: &str,
expected_name: &str,
) -> Result<Checkpoint, AftError> {
let meta_path = checkpoint_dir.join("meta.json");
let bytes = fs::read(&meta_path).map_err(|error| AftError::IoError {
path: meta_path.display().to_string(),
message: format!("failed to read durable checkpoint metadata: {error}"),
})?;
let meta = serde_json::from_slice::<DiskCheckpointMeta>(&bytes).map_err(|error| {
AftError::IoError {
path: meta_path.display().to_string(),
message: format!("failed to parse durable checkpoint metadata: {error}"),
}
})?;
if meta.schema_version != CHECKPOINT_SCHEMA_VERSION {
return Err(AftError::IoError {
path: meta_path.display().to_string(),
message: format!(
"unsupported durable checkpoint metadata schema {}",
meta.schema_version
),
});
}
if meta.session_id != session
|| meta.name != expected_name
|| !is_safe_checkpoint_name(&meta.name)
{
return Err(AftError::IoError {
path: meta_path.display().to_string(),
message: "durable checkpoint metadata does not match its session or directory"
.to_string(),
});
}
let mut file_contents = HashMap::with_capacity(meta.files.len());
for file in &meta.files {
if !is_safe_blob_name(&file.blob) {
return Err(AftError::IoError {
path: meta_path.display().to_string(),
message: format!("invalid durable checkpoint blob name {}", file.blob),
});
}
let blob_path = checkpoint_dir.join(&file.blob);
let blob = fs::read(&blob_path).map_err(|error| AftError::IoError {
path: blob_path.display().to_string(),
message: format!("failed to read durable checkpoint blob: {error}"),
})?;
let path = PathBuf::from(&file.original_path);
let checkpoint_file =
CheckpointFile::from_disk(file, blob).map_err(|message| AftError::IoError {
path: blob_path.display().to_string(),
message,
})?;
if file_contents
.insert(path.clone(), checkpoint_file)
.is_some()
{
return Err(AftError::IoError {
path: meta_path.display().to_string(),
message: format!("duplicate durable checkpoint path {}", path.display()),
});
}
}
Ok(Checkpoint {
name: meta.name,
file_contents,
created_at: meta.created_at,
created_order: meta.created_order,
})
}
fn checkpoint_file_bytes(file: &CheckpointFile) -> Vec<u8> {
match &file.kind {
CheckpointFileKind::Regular { bytes } => bytes.to_vec(),
CheckpointFileKind::Symlink { target, .. } => {
target.as_os_str().to_string_lossy().as_bytes().to_vec()
}
}
}
fn write_temp_fsync_rename(dir: &Path, final_name: &str, bytes: &[u8]) -> io::Result<()> {
let tmp_name = format!(
".{}.{}.{}.tmp",
final_name,
std::process::id(),
current_timestamp_nanos()
);
let tmp_path = dir.join(tmp_name);
let final_path = dir.join(final_name);
{
let mut file = fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&tmp_path)?;
file.write_all(bytes)?;
file.sync_all()?;
}
fs::rename(tmp_path, final_path)
}
#[cfg(unix)]
fn fsync_dir(path: &Path) -> io::Result<()> {
fs::File::open(path)?.sync_all()
}
#[cfg(not(unix))]
fn fsync_dir(_path: &Path) -> io::Result<()> {
Ok(())
}
fn prune_unreferenced_checkpoint_blobs(
checkpoint_dir: &Path,
files: &[DiskCheckpointFileMeta],
) -> io::Result<()> {
let referenced = files
.iter()
.map(|file| file.blob.as_str())
.collect::<HashSet<_>>();
for entry in fs::read_dir(checkpoint_dir)? {
let entry = entry?;
let path = entry.path();
if !path.is_file() {
continue;
}
let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
continue;
};
if (name.starts_with("file_") && name.ends_with(".blob") && !referenced.contains(name))
|| name.contains(".tmp.")
|| name.ends_with(".tmp")
{
let _ = fs::remove_file(path);
}
}
Ok(())
}
fn sweep_expired_durable_checkpoints(checkpoints_dir: &Path, now: u64) {
let Ok(boundary) = crate::walk_boundary::DeviceBoundary::for_root(checkpoints_dir) else {
crate::slog_warn!(
"cannot establish filesystem boundary for checkpoint sweep {}",
checkpoints_dir.display()
);
return;
};
let mut skipped_foreign_mounts = 0usize;
let Ok(session_entries) = fs::read_dir(checkpoints_dir) else {
return;
};
for session_entry in session_entries.flatten() {
let session_dir = session_entry.path();
if !session_dir.is_dir() {
continue;
}
if !boundary.should_descend(&session_dir).unwrap_or(false) {
skipped_foreign_mounts += 1;
continue;
}
let Ok(checkpoint_entries) = fs::read_dir(&session_dir) else {
continue;
};
for checkpoint_entry in checkpoint_entries.flatten() {
let checkpoint_dir = checkpoint_entry.path();
if !checkpoint_dir.is_dir() {
continue;
}
if !boundary.should_descend(&checkpoint_dir).unwrap_or(false) {
skipped_foreign_mounts += 1;
continue;
}
let meta_path = checkpoint_dir.join("meta.json");
let Ok(bytes) = fs::read(&meta_path) else {
continue;
};
let Ok(meta) = serde_json::from_slice::<DiskCheckpointMeta>(&bytes) else {
continue;
};
if now.saturating_sub(meta.created_at) < NAMED_CHECKPOINT_RETENTION_SECS {
continue;
}
if let Err(error) = fs::remove_dir_all(&checkpoint_dir) {
crate::slog_warn!(
"failed to remove expired durable checkpoint {}: {}",
checkpoint_dir.display(),
error
);
}
}
let _ = fs::remove_dir(&session_dir);
}
if skipped_foreign_mounts > 0 {
crate::slog_warn!(
"checkpoint sweep skipped {} foreign filesystem mount(s) below {}",
skipped_foreign_mounts,
checkpoints_dir.display()
);
}
}
fn absolute_checkpoint_path(path: PathBuf) -> PathBuf {
if path.is_absolute() {
return normalize_checkpoint_path(&path);
}
let cwd = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
normalize_checkpoint_path(&cwd.join(path))
}
fn normalize_checkpoint_path(path: &Path) -> PathBuf {
let mut normalized = PathBuf::new();
for component in path.components() {
match component {
std::path::Component::CurDir => {}
std::path::Component::ParentDir => {
if !normalized.pop() {
normalized.push(component.as_os_str());
}
}
other => normalized.push(other.as_os_str()),
}
}
normalized
}
fn restore_paths_atomically(checkpoint: &Checkpoint, paths: &[PathBuf]) -> Result<(), AftError> {
let mut pre_restore_snapshot: HashMap<PathBuf, Option<CheckpointFile>> = HashMap::new();
for path in paths {
let current = CheckpointFile::read_optional(path).map_err(|error| AftError::IoError {
path: path.display().to_string(),
message: format!("failed to snapshot pre-restore file metadata: {error}"),
})?;
pre_restore_snapshot.insert(path.clone(), current);
}
let mut restored_paths: Vec<PathBuf> = Vec::new();
let mut created_dirs: Vec<PathBuf> = Vec::new();
for path in paths {
let snapshot =
checkpoint
.file_contents
.get(path)
.ok_or_else(|| AftError::FileNotFound {
path: path.display().to_string(),
})?;
if let Err(e) = write_restored_file(path, snapshot, &mut created_dirs) {
let mut rollback_errors = Vec::new();
if let Some(snapshot) = pre_restore_snapshot.get(path) {
if let Err(rollback_error) = restore_snapshot_file(path, snapshot.as_ref()) {
rollback_errors.push(format!("{}: {}", path.display(), rollback_error));
}
}
for restored_path in restored_paths.iter().rev() {
if let Some(snapshot) = pre_restore_snapshot.get(restored_path) {
if let Err(rollback_error) =
restore_snapshot_file(restored_path, snapshot.as_ref())
{
rollback_errors.push(format!(
"{}: {}",
restored_path.display(),
rollback_error
));
}
}
}
let dirs_rollback_ok = rollback_created_dirs(&created_dirs);
if rollback_errors.is_empty() && dirs_rollback_ok {
return Err(e);
}
return Err(AftError::IoError {
path: path.display().to_string(),
message: format!(
"{}; restore_checkpoint rollback_succeeded: {}; rollback_errors: {}",
e,
rollback_errors.is_empty() && dirs_rollback_ok,
if rollback_errors.is_empty() {
"none".to_string()
} else {
rollback_errors.join("; ")
}
),
});
}
restored_paths.push(path.clone());
}
Ok(())
}
fn restore_snapshot_file(path: &Path, snapshot: Option<&CheckpointFile>) -> Result<(), AftError> {
match snapshot {
Some(snapshot) => write_restored_file(path, snapshot, &mut Vec::new()),
None => remove_file_if_exists(path).map_err(|error| AftError::IoError {
path: path.display().to_string(),
message: format!("failed to remove file during checkpoint restore rollback: {error}"),
}),
}
}
fn write_restored_file(
path: &Path,
snapshot: &CheckpointFile,
created_dirs: &mut Vec<PathBuf>,
) -> Result<(), AftError> {
create_parent_dirs(path, created_dirs)?;
match &snapshot.kind {
CheckpointFileKind::Regular { bytes } => {
if path_is_symlink(path) {
remove_file_if_exists(path).map_err(|error| AftError::IoError {
path: path.display().to_string(),
message: format!("failed to replace symlink with regular file: {error}"),
})?;
}
fs::write(path, bytes).map_err(|error| AftError::IoError {
path: path.display().to_string(),
message: format!("failed to restore checkpoint file contents: {error}"),
})?;
restore_checkpoint_permissions(path, snapshot).map_err(|error| AftError::IoError {
path: path.display().to_string(),
message: format!("failed to restore checkpoint file permissions: {error}"),
})
}
CheckpointFileKind::Symlink {
target,
target_is_dir,
} => {
remove_file_if_exists(path).map_err(|error| AftError::IoError {
path: path.display().to_string(),
message: format!("failed to replace file with checkpoint symlink: {error}"),
})?;
create_symlink(target, path, *target_is_dir).map_err(|error| AftError::IoError {
path: path.display().to_string(),
message: format!("failed to restore checkpoint symlink: {error}"),
})
}
}
}
fn restore_checkpoint_permissions(path: &Path, snapshot: &CheckpointFile) -> io::Result<()> {
if let Some(metadata) = &snapshot.metadata {
return fs::set_permissions(path, metadata.permissions());
}
restore_checkpoint_mode(path, snapshot.mode)
}
#[cfg(unix)]
fn restore_checkpoint_mode(path: &Path, mode: Option<u32>) -> io::Result<()> {
use std::os::unix::fs::PermissionsExt;
if let Some(mode) = mode {
fs::set_permissions(path, fs::Permissions::from_mode(mode))?;
}
Ok(())
}
#[cfg(not(unix))]
fn restore_checkpoint_mode(_path: &Path, _mode: Option<u32>) -> io::Result<()> {
Ok(())
}
fn create_parent_dirs(path: &Path, created_dirs: &mut Vec<PathBuf>) -> Result<(), AftError> {
if let Some(parent) = path.parent() {
let missing_dirs = missing_parent_dirs(parent);
fs::create_dir_all(parent).map_err(|error| AftError::IoError {
path: parent.display().to_string(),
message: format!("failed to create checkpoint restore parent directories: {error}"),
})?;
created_dirs.extend(missing_dirs);
}
Ok(())
}
fn path_is_symlink(path: &Path) -> bool {
fs::symlink_metadata(path)
.map(|metadata| metadata.file_type().is_symlink())
.unwrap_or(false)
}
fn remove_file_if_exists(path: &Path) -> io::Result<()> {
match fs::remove_file(path) {
Ok(()) => Ok(()),
Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
Err(error) => Err(error),
}
}
#[cfg(unix)]
fn create_symlink(target: &Path, link: &Path, target_is_dir: bool) -> io::Result<()> {
let _ = target_is_dir;
std::os::unix::fs::symlink(target, link)
}
#[cfg(windows)]
fn create_symlink(target: &Path, link: &Path, target_is_dir: bool) -> io::Result<()> {
if target_is_dir {
std::os::windows::fs::symlink_dir(target, link)
} else {
std::os::windows::fs::symlink_file(target, link)
}
}
#[cfg(not(any(unix, windows)))]
fn create_symlink(_target: &Path, _link: &Path, _target_is_dir: bool) -> io::Result<()> {
Err(io::Error::new(
io::ErrorKind::Unsupported,
"checkpoint symlink restore is unsupported on this platform",
))
}
fn missing_parent_dirs(parent: &Path) -> Vec<PathBuf> {
let mut dirs = Vec::new();
let mut current = Some(parent);
while let Some(dir) = current {
if dir.as_os_str().is_empty() || dir.exists() {
break;
}
dirs.push(dir.to_path_buf());
current = dir.parent();
}
dirs
}
fn rollback_created_dirs(dirs: &[PathBuf]) -> bool {
let mut dirs = dirs.to_vec();
dirs.sort_by_key(|dir| std::cmp::Reverse(dir.components().count()));
dirs.dedup();
let mut ok = true;
for dir in dirs {
match std::fs::remove_dir(&dir) {
Ok(()) => {}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(_) => ok = false,
}
}
ok
}
fn remove_empty_scope_dir(scope_dir: &Path) {
let _ = fs::remove_dir(scope_dir);
}
fn sweep_empty_scope_dirs(checkpoints_root: &Path) {
let entries = match fs::read_dir(checkpoints_root) {
Ok(entries) => entries,
Err(_) => return,
};
for entry in entries.flatten() {
let Ok(file_type) = entry.file_type() else {
continue;
};
if file_type.is_dir() {
remove_empty_scope_dir(&entry.path());
}
}
}
fn current_timestamp() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
}
fn current_timestamp_nanos() -> u64 {
u64::try_from(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_nanos(),
)
.unwrap_or(u64::MAX)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::protocol::DEFAULT_SESSION_ID;
use std::fs;
fn temp_file(name: &str, content: &str) -> (PathBuf, tempfile::TempDir) {
let dir = tempfile::Builder::new()
.prefix("aft_checkpoint_tests_")
.tempdir()
.expect("create checkpoint temp dir");
let path = dir.path().join(name);
fs::write(&path, content).unwrap();
(path, dir)
}
fn fresh_checkpoint_store(storage: &Path) -> CheckpointStore {
let lock_path = storage
.join("checkpoints")
.join("test-project")
.join("checkpoint.lock");
let mut store = CheckpointStore::with_lock_path(lock_path, CHECKPOINT_LOCK_TIMEOUT);
store.set_storage_dir_for_harness(storage.to_path_buf(), crate::harness::Harness::Opencode);
store
}
fn checkpoint_store() -> (CheckpointStore, tempfile::TempDir) {
let dir = tempfile::tempdir().unwrap();
(fresh_checkpoint_store(dir.path()), dir)
}
fn checkpoint_file(content: &str) -> CheckpointFile {
let file = tempfile::NamedTempFile::new().unwrap();
fs::write(file.path(), content).unwrap();
CheckpointFile::read(file.path()).unwrap()
}
#[test]
fn create_and_restore_round_trip() {
let (path1, _dir1) = temp_file("cp_rt1.txt", "hello");
let (path2, _dir2) = temp_file("cp_rt2.txt", "world");
let backup_store = BackupStore::new();
let (mut store, _store_dir) = checkpoint_store();
let info = store
.create(
DEFAULT_SESSION_ID,
"snap1",
vec![path1.clone(), path2.clone()],
&backup_store,
)
.unwrap();
assert_eq!(info.name, "snap1");
assert_eq!(info.file_count, 2);
fs::write(&path1, "changed1").unwrap();
fs::write(&path2, "changed2").unwrap();
let info = store.restore(DEFAULT_SESSION_ID, "snap1").unwrap();
assert_eq!(info.file_count, 2);
assert_eq!(fs::read_to_string(&path1).unwrap(), "hello");
assert_eq!(fs::read_to_string(&path2).unwrap(), "world");
}
#[cfg(unix)]
#[test]
fn durable_checkpoint_hydrates_after_restart_with_bytes_and_mode() {
use std::os::unix::fs::PermissionsExt;
let files = tempfile::tempdir().unwrap();
let path = files.path().join("durable-mode.bin");
let original = b"draft decision\n\0byte exact\n";
fs::write(&path, original).unwrap();
let mut mode = fs::metadata(&path).unwrap().permissions();
mode.set_mode(0o600);
fs::set_permissions(&path, mode).unwrap();
let backup_store = BackupStore::new();
let (mut first, storage) = checkpoint_store();
let info = first
.create(
DEFAULT_SESSION_ID,
"restart-mode",
vec![path.clone()],
&backup_store,
)
.unwrap();
let durable_path = info.storage_path.expect("durable checkpoint path");
assert!(durable_path.join("meta.json").is_file());
assert!(
fs::read_dir(&durable_path)
.unwrap()
.flatten()
.any(|entry| entry.path().extension().is_some_and(|ext| ext == "blob")),
"checkpoint must persist one or more file blobs"
);
fs::write(&path, b"mutated\n").unwrap();
let mut changed_mode = fs::metadata(&path).unwrap().permissions();
changed_mode.set_mode(0o644);
fs::set_permissions(&path, changed_mode).unwrap();
drop(first);
let mut restarted = fresh_checkpoint_store(storage.path());
let listed = restarted.list(DEFAULT_SESSION_ID).unwrap();
assert_eq!(
listed.len(),
1,
"fresh store must hydrate durable checkpoint"
);
assert_eq!(listed[0].name, "restart-mode");
restarted
.restore(DEFAULT_SESSION_ID, "restart-mode")
.unwrap();
assert_eq!(fs::read(&path).unwrap(), original);
assert_eq!(
fs::metadata(&path).unwrap().permissions().mode() & 0o777,
0o600
);
}
#[cfg(unix)]
#[test]
fn durable_checkpoint_hydrates_symlink_without_following_target() {
let files = tempfile::tempdir().unwrap();
let target = files.path().join("target.txt");
let link = files.path().join("link.txt");
fs::write(&target, "target content").unwrap();
std::os::unix::fs::symlink(&target, &link).unwrap();
let backup_store = BackupStore::new();
let (mut first, storage) = checkpoint_store();
first
.create(
DEFAULT_SESSION_ID,
"restart-symlink",
vec![link.clone()],
&backup_store,
)
.unwrap();
fs::remove_file(&link).unwrap();
fs::write(&link, "plain replacement").unwrap();
drop(first);
let mut restarted = fresh_checkpoint_store(storage.path());
restarted
.restore(DEFAULT_SESSION_ID, "restart-symlink")
.unwrap();
assert!(fs::symlink_metadata(&link)
.unwrap()
.file_type()
.is_symlink());
assert_eq!(fs::read_link(&link).unwrap(), target);
assert_eq!(fs::read_to_string(&target).unwrap(), "target content");
}
#[test]
fn checkpoint_retention_evicts_oldest_name_from_memory_and_disk() {
let (path, _files) = temp_file("retention.txt", "version-0");
let backup_store = BackupStore::new();
let (mut store, storage) = checkpoint_store();
for index in 0..=MAX_NAMED_CHECKPOINTS_PER_SESSION {
fs::write(&path, format!("version-{index}")).unwrap();
let info = store
.create(
DEFAULT_SESSION_ID,
&format!("checkpoint-{index:02}"),
vec![path.clone()],
&backup_store,
)
.unwrap();
if index == MAX_NAMED_CHECKPOINTS_PER_SESSION {
assert_eq!(info.evicted, vec!["checkpoint-00"]);
} else {
assert!(info.evicted.is_empty());
}
}
let listed = store.list(DEFAULT_SESSION_ID).unwrap();
assert_eq!(listed.len(), MAX_NAMED_CHECKPOINTS_PER_SESSION);
assert!(listed.iter().all(|info| info.name != "checkpoint-00"));
let old_dir = storage
.path()
.join("opencode")
.join("checkpoints")
.join(hash_session(DEFAULT_SESSION_ID))
.join("checkpoint-00");
assert!(!old_dir.exists(), "evicted checkpoint must leave disk too");
}
#[test]
fn hydration_finishes_retention_after_interrupted_create() {
let (path, _files) = temp_file("interrupted-retention.txt", "checkpoint content");
let backup_store = BackupStore::new();
let (mut first, storage) = checkpoint_store();
for index in 0..MAX_NAMED_CHECKPOINTS_PER_SESSION {
first
.create(
DEFAULT_SESSION_ID,
&format!("checkpoint-{index:02}"),
vec![path.clone()],
&backup_store,
)
.unwrap();
}
let checkpoint = Checkpoint {
name: "checkpoint-20".to_string(),
file_contents: HashMap::from([(path, checkpoint_file("newest"))]),
created_at: current_timestamp(),
created_order: u64::MAX,
};
{
let _lock = first.acquire_mutation_lock().unwrap();
first
.persist_checkpoint_locked(DEFAULT_SESSION_ID, &checkpoint)
.unwrap();
}
drop(first);
let session_dir = storage
.path()
.join("opencode")
.join("checkpoints")
.join(hash_session(DEFAULT_SESSION_ID));
assert_eq!(fs::read_dir(&session_dir).unwrap().count(), 21);
let mut restarted = fresh_checkpoint_store(storage.path());
let listed = restarted.list(DEFAULT_SESSION_ID).unwrap();
assert_eq!(listed.len(), MAX_NAMED_CHECKPOINTS_PER_SESSION);
assert!(listed.iter().all(|info| info.name != "checkpoint-00"));
assert!(
!session_dir.join("checkpoint-00").exists(),
"hydration must finish interrupted retention on disk"
);
}
#[test]
fn cleanup_refuses_to_sweep_scope_dirs_outside_a_checkpoints_root() {
let temp_root = tempfile::tempdir().expect("temp root");
let victim = temp_root.path().join("innocent-empty-sibling");
fs::create_dir(&victim).expect("victim dir");
let scope_dir = temp_root.path().join("scope");
fs::create_dir(&scope_dir).expect("scope dir");
let lock_path = scope_dir.join("checkpoint.lock");
let mut store = CheckpointStore::with_lock_path(lock_path, CHECKPOINT_LOCK_TIMEOUT);
store.cleanup_locked().expect("cleanup");
assert!(
victim.exists(),
"cleanup must not sweep empty dirs outside a `checkpoints` root"
);
}
#[test]
fn cleanup_sweeps_durable_checkpoints_older_than_fourteen_days() {
let (path, _files) = temp_file("durable-gc.txt", "original");
let backup_store = BackupStore::new();
let (mut store, _storage) = checkpoint_store();
let info = store
.create(
DEFAULT_SESSION_ID,
"expired-durable",
vec![path],
&backup_store,
)
.unwrap();
let durable_path = info.storage_path.unwrap();
let meta_path = durable_path.join("meta.json");
let mut meta: DiskCheckpointMeta =
serde_json::from_slice(&fs::read(&meta_path).unwrap()).unwrap();
meta.created_at = current_timestamp()
.saturating_sub(NAMED_CHECKPOINT_RETENTION_SECS)
.saturating_sub(1);
fs::write(&meta_path, serde_json::to_vec_pretty(&meta).unwrap()).unwrap();
store.cleanup();
assert!(
!durable_path.exists(),
"fourteen-day cleanup must remove the durable checkpoint directory"
);
}
#[test]
fn durable_hydration_fails_when_a_referenced_blob_is_missing() {
let (path, _files) = temp_file("hydration-control.txt", "original");
let backup_store = BackupStore::new();
let (mut first, storage) = checkpoint_store();
let info = first
.create(
DEFAULT_SESSION_ID,
"hydration-control",
vec![path],
&backup_store,
)
.unwrap();
let durable_path = info.storage_path.unwrap();
let meta: DiskCheckpointMeta =
serde_json::from_slice(&fs::read(durable_path.join("meta.json")).unwrap()).unwrap();
fs::remove_file(durable_path.join(&meta.files[0].blob)).unwrap();
drop(first);
let mut restarted = fresh_checkpoint_store(storage.path());
let error = restarted.list(DEFAULT_SESSION_ID).unwrap_err();
match error {
AftError::IoError { message, .. } => {
assert!(message.contains("failed to read durable checkpoint blob"));
}
other => panic!("expected durable hydration I/O error, got {other:?}"),
}
}
#[test]
fn overwrite_existing_name() {
let (path, _dir) = temp_file("cp_overwrite.txt", "v1");
let backup_store = BackupStore::new();
let (mut store, _store_dir) = checkpoint_store();
store
.create(DEFAULT_SESSION_ID, "dup", vec![path.clone()], &backup_store)
.unwrap();
fs::write(&path, "v2").unwrap();
store
.create(DEFAULT_SESSION_ID, "dup", vec![path.clone()], &backup_store)
.unwrap();
fs::write(&path, "v3").unwrap();
store.restore(DEFAULT_SESSION_ID, "dup").unwrap();
assert_eq!(fs::read_to_string(&path).unwrap(), "v2");
}
#[test]
fn list_returns_metadata_scoped_to_session() {
let (path, _dir) = temp_file("cp_list.txt", "data");
let backup_store = BackupStore::new();
let (mut store, _store_dir) = checkpoint_store();
store
.create(DEFAULT_SESSION_ID, "a", vec![path.clone()], &backup_store)
.unwrap();
store
.create(DEFAULT_SESSION_ID, "b", vec![path.clone()], &backup_store)
.unwrap();
store
.create("other_session", "c", vec![path.clone()], &backup_store)
.unwrap();
let default_list = store.list(DEFAULT_SESSION_ID).unwrap();
assert_eq!(default_list.len(), 2);
let names: Vec<&str> = default_list.iter().map(|i| i.name.as_str()).collect();
assert!(names.contains(&"a"));
assert!(names.contains(&"b"));
let other_list = store.list("other_session").unwrap();
assert_eq!(other_list.len(), 1);
assert_eq!(other_list[0].name, "c");
}
#[test]
fn sessions_isolate_checkpoint_names() {
let (path_a, _dir_a) = temp_file("cp_isolated_a.txt", "a-original");
let (path_b, _dir_b) = temp_file("cp_isolated_b.txt", "b-original");
let backup_store = BackupStore::new();
let (mut store, _store_dir) = checkpoint_store();
store
.create("session_a", "snap", vec![path_a.clone()], &backup_store)
.unwrap();
store
.create("session_b", "snap", vec![path_b.clone()], &backup_store)
.unwrap();
fs::write(&path_a, "a-modified").unwrap();
fs::write(&path_b, "b-modified").unwrap();
store.restore("session_a", "snap").unwrap();
assert_eq!(fs::read_to_string(&path_a).unwrap(), "a-original");
assert_eq!(fs::read_to_string(&path_b).unwrap(), "b-modified");
fs::write(&path_a, "a-modified").unwrap();
store.restore("session_b", "snap").unwrap();
assert_eq!(fs::read_to_string(&path_a).unwrap(), "a-modified");
assert_eq!(fs::read_to_string(&path_b).unwrap(), "b-original");
}
#[test]
fn checkpoint_lock_scope_is_removed_after_release() {
let dir = tempfile::tempdir().unwrap();
let scope_dir = dir.path().join("checkpoints").join("project-scope");
let lock_path = scope_dir.join("checkpoint.lock");
let path = dir.path().join("checkpoint.txt");
fs::write(&path, "data").unwrap();
let backup_store = BackupStore::new();
let mut store = CheckpointStore::with_lock_path(lock_path, CHECKPOINT_LOCK_TIMEOUT);
store
.create(DEFAULT_SESSION_ID, "released", vec![path], &backup_store)
.unwrap();
assert!(!scope_dir.exists(), "released lock scope should be removed");
}
#[test]
fn cleanup_removes_expired_across_sessions() {
let (path, _dir) = temp_file("cp_cleanup.txt", "data");
let backup_store = BackupStore::new();
let (mut store, _store_dir) = checkpoint_store();
store
.create(
DEFAULT_SESSION_ID,
"recent",
vec![path.clone()],
&backup_store,
)
.unwrap();
store
.checkpoints
.entry("other".to_string())
.or_default()
.insert(
"old".to_string(),
Checkpoint {
name: "old".to_string(),
file_contents: HashMap::new(),
created_at: 1000, created_order: 1000,
},
);
assert_eq!(store.total_count(), 2);
store.cleanup();
assert_eq!(store.total_count(), 1);
assert_eq!(store.list(DEFAULT_SESSION_ID).unwrap()[0].name, "recent");
assert!(store.list("other").unwrap().is_empty());
}
#[test]
fn cleanup_sweeps_empty_scope_dirs_but_keeps_live_lock_scope() {
let dir = tempfile::tempdir().unwrap();
let checkpoints_root = dir.path().join("checkpoints");
let empty_a = checkpoints_root.join("empty-a");
let empty_b = checkpoints_root.join("empty-b");
let live_scope = checkpoints_root.join("live-scope");
fs::create_dir_all(&empty_a).unwrap();
fs::create_dir_all(&empty_b).unwrap();
fs::create_dir_all(&live_scope).unwrap();
fs::write(live_scope.join("checkpoint.lock"), "live lock").unwrap();
let lock_path = checkpoints_root
.join("current-scope")
.join("checkpoint.lock");
let mut store = CheckpointStore::with_lock_path(lock_path, CHECKPOINT_LOCK_TIMEOUT);
store.cleanup();
assert!(!empty_a.exists());
assert!(!empty_b.exists());
assert!(live_scope.is_dir());
assert!(live_scope.join("checkpoint.lock").is_file());
}
#[test]
fn cleanup_ignores_non_empty_scope_dir_removal_failure() {
let dir = tempfile::tempdir().unwrap();
let checkpoints_root = dir.path().join("checkpoints");
let scope_dir = checkpoints_root.join("racing-scope");
fs::create_dir_all(&scope_dir).unwrap();
fs::write(scope_dir.join("checkpoint.lock"), "lock appeared").unwrap();
let lock_path = checkpoints_root
.join("current-scope")
.join("checkpoint.lock");
let mut store = CheckpointStore::with_lock_path(lock_path, CHECKPOINT_LOCK_TIMEOUT);
store.cleanup();
assert!(scope_dir.is_dir());
assert!(scope_dir.join("checkpoint.lock").is_file());
}
#[test]
fn restore_nonexistent_returns_error() {
let (mut store, _store_dir) = checkpoint_store();
let result = store.restore(DEFAULT_SESSION_ID, "nope");
assert!(result.is_err());
match result.unwrap_err() {
AftError::CheckpointNotFound { name } => {
assert_eq!(name, "nope");
}
other => panic!("expected CheckpointNotFound, got: {:?}", other),
}
}
#[test]
fn restore_nonexistent_in_other_session_returns_error() {
let (path, _dir) = temp_file("cp_cross_session.txt", "data");
let backup_store = BackupStore::new();
let (mut store, _store_dir) = checkpoint_store();
store
.create("session_a", "only_a", vec![path], &backup_store)
.unwrap();
assert!(store.restore("session_b", "only_a").is_err());
}
#[test]
fn create_skips_missing_files_from_backup_tracked_set() {
let (readable, _readable_dir) = temp_file("cp_skip_readable.txt", "still_here");
let (deleted, _deleted_dir) = temp_file("cp_skip_deleted.txt", "about_to_vanish");
let deleted_canonical = fs::canonicalize(&deleted).unwrap();
let mut backup_store = BackupStore::new();
backup_store
.snapshot(DEFAULT_SESSION_ID, &readable, "auto")
.unwrap();
backup_store
.snapshot(DEFAULT_SESSION_ID, &deleted, "auto")
.unwrap();
fs::remove_file(&deleted).unwrap();
let (mut store, _store_dir) = checkpoint_store();
let info = store
.create(DEFAULT_SESSION_ID, "partial", vec![], &backup_store)
.expect("checkpoint should succeed despite one missing file");
assert_eq!(info.file_count, 1);
assert_eq!(info.skipped.len(), 1);
assert_eq!(info.skipped[0].0, deleted_canonical);
assert!(!info.skipped[0].1.is_empty());
}
#[test]
fn create_with_explicit_single_missing_file_errors() {
let dir = tempfile::tempdir().unwrap();
let missing = dir.path().join("cp_explicit_missing_does_not_exist.txt");
let backup_store = BackupStore::new();
let (mut store, _store_dir) = checkpoint_store();
let result = store.create(
DEFAULT_SESSION_ID,
"explicit",
vec![missing.clone()],
&backup_store,
);
assert!(result.is_err());
match result.unwrap_err() {
AftError::FileNotFound { path } => {
assert!(path.contains(&missing.display().to_string()));
}
other => panic!("expected FileNotFound, got: {:?}", other),
}
}
#[test]
fn create_with_explicit_mixed_files_keeps_readable_and_reports_skipped() {
let (good, _good_dir) = temp_file("cp_mixed_good.txt", "ok");
let missing_dir = tempfile::tempdir().unwrap();
let missing = missing_dir.path().join("cp_mixed_missing.txt");
let backup_store = BackupStore::new();
let (mut store, _store_dir) = checkpoint_store();
let info = store
.create(
DEFAULT_SESSION_ID,
"mixed",
vec![good.clone(), missing.clone()],
&backup_store,
)
.expect("mixed checkpoint should succeed when any file is readable");
assert_eq!(info.file_count, 1);
assert_eq!(info.skipped.len(), 1);
assert_eq!(info.skipped[0].0, missing);
}
#[test]
fn create_with_empty_files_uses_backup_tracked() {
let (path, _dir) = temp_file("cp_tracked.txt", "tracked_content");
let mut backup_store = BackupStore::new();
backup_store
.snapshot(DEFAULT_SESSION_ID, &path, "auto")
.unwrap();
let (mut store, _store_dir) = checkpoint_store();
let info = store
.create(DEFAULT_SESSION_ID, "from_tracked", vec![], &backup_store)
.unwrap();
assert!(info.file_count >= 1);
fs::write(&path, "modified").unwrap();
store.restore(DEFAULT_SESSION_ID, "from_tracked").unwrap();
assert_eq!(fs::read_to_string(&path).unwrap(), "tracked_content");
}
#[test]
fn restore_recreates_missing_parent_directories() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("nested").join("deeper").join("file.txt");
fs::create_dir_all(path.parent().unwrap()).unwrap();
fs::write(&path, "original nested content").unwrap();
let backup_store = BackupStore::new();
let (mut store, _store_dir) = checkpoint_store();
store
.create(
DEFAULT_SESSION_ID,
"nested",
vec![path.clone()],
&backup_store,
)
.unwrap();
fs::remove_dir_all(dir.path().join("nested")).unwrap();
store.restore(DEFAULT_SESSION_ID, "nested").unwrap();
assert_eq!(
fs::read_to_string(&path).unwrap(),
"original nested content"
);
}
#[cfg(unix)]
#[test]
fn checkpoint_restore_rolls_back_on_partial_failure() {
use std::os::unix::fs::PermissionsExt;
let dir = tempfile::tempdir().unwrap();
let path_a = dir.path().join("a.txt");
let path_b = dir.path().join("b.txt");
fs::write(&path_a, "checkpoint-a").unwrap();
fs::write(&path_b, "checkpoint-b").unwrap();
let backup_store = BackupStore::new();
let (mut store, _store_dir) = checkpoint_store();
store
.create(
DEFAULT_SESSION_ID,
"partial_failure",
vec![path_a.clone(), path_b.clone()],
&backup_store,
)
.unwrap();
fs::write(&path_a, "pre-restore-a").unwrap();
fs::write(&path_b, "pre-restore-b").unwrap();
let mut readonly = fs::metadata(&path_b).unwrap().permissions();
readonly.set_mode(0o444);
fs::set_permissions(&path_b, readonly).unwrap();
let result = store.restore(DEFAULT_SESSION_ID, "partial_failure");
let mut writable = fs::metadata(&path_b).unwrap().permissions();
writable.set_mode(0o644);
fs::set_permissions(&path_b, writable).unwrap();
assert!(result.is_err(), "restore should surface write failure");
assert_eq!(fs::read_to_string(&path_a).unwrap(), "pre-restore-a");
assert_eq!(fs::read_to_string(&path_b).unwrap(), "pre-restore-b");
}
#[test]
fn checkpoint_create_and_restore_use_mutation_lock() {
let dir = tempfile::tempdir().unwrap();
let lock_path = dir.path().join("locks").join("checkpoint.lock");
fs::create_dir_all(lock_path.parent().unwrap()).unwrap();
let mut store =
CheckpointStore::with_lock_path(lock_path.clone(), Duration::from_millis(50));
let backup_store = BackupStore::new();
let path = dir.path().join("locked.txt");
fs::write(&path, "original").unwrap();
let held_lock =
fs_lock::try_acquire(&lock_path, Duration::from_secs(1)).expect("hold checkpoint lock");
let create_result = store.create(
DEFAULT_SESSION_ID,
"locked",
vec![path.clone()],
&backup_store,
);
assert!(matches!(create_result, Err(AftError::IoError { .. })));
drop(held_lock);
store
.create(
DEFAULT_SESSION_ID,
"locked",
vec![path.clone()],
&backup_store,
)
.unwrap();
fs::write(&path, "changed").unwrap();
fs::create_dir_all(lock_path.parent().unwrap()).unwrap();
let held_lock =
fs_lock::try_acquire(&lock_path, Duration::from_secs(1)).expect("hold checkpoint lock");
let restore_result = store.restore(DEFAULT_SESSION_ID, "locked");
assert!(matches!(restore_result, Err(AftError::IoError { .. })));
drop(held_lock);
store.restore(DEFAULT_SESSION_ID, "locked").unwrap();
assert_eq!(fs::read_to_string(&path).unwrap(), "original");
}
#[cfg(unix)]
#[test]
fn checkpoint_restore_preserves_regular_file_permissions() {
use std::os::unix::fs::PermissionsExt;
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("mode.txt");
fs::write(&path, "original").unwrap();
let mut original_permissions = fs::metadata(&path).unwrap().permissions();
original_permissions.set_mode(0o600);
fs::set_permissions(&path, original_permissions).unwrap();
let backup_store = BackupStore::new();
let (mut store, _store_dir) = checkpoint_store();
store
.create(
DEFAULT_SESSION_ID,
"mode",
vec![path.clone()],
&backup_store,
)
.unwrap();
fs::write(&path, "changed").unwrap();
let mut changed_permissions = fs::metadata(&path).unwrap().permissions();
changed_permissions.set_mode(0o644);
fs::set_permissions(&path, changed_permissions).unwrap();
store.restore(DEFAULT_SESSION_ID, "mode").unwrap();
assert_eq!(fs::read_to_string(&path).unwrap(), "original");
let restored_mode = fs::metadata(&path).unwrap().permissions().mode() & 0o777;
assert_eq!(restored_mode, 0o600);
}
#[cfg(unix)]
#[test]
fn checkpoint_restore_recreates_symlink() {
let dir = tempfile::tempdir().unwrap();
let target = dir.path().join("target.txt");
let link = dir.path().join("link.txt");
fs::write(&target, "target content").unwrap();
std::os::unix::fs::symlink(&target, &link).unwrap();
let backup_store = BackupStore::new();
let (mut store, _store_dir) = checkpoint_store();
store
.create(
DEFAULT_SESSION_ID,
"symlink",
vec![link.clone()],
&backup_store,
)
.unwrap();
fs::remove_file(&link).unwrap();
fs::write(&link, "plain file").unwrap();
store.restore(DEFAULT_SESSION_ID, "symlink").unwrap();
assert!(fs::symlink_metadata(&link)
.unwrap()
.file_type()
.is_symlink());
assert_eq!(fs::read_link(&link).unwrap(), target);
assert_eq!(fs::read_to_string(&link).unwrap(), "target content");
}
#[test]
fn captured_regular_file_is_shared_by_checkpoint_and_backup() {
let (path, _dir) = temp_file("shared-capture.txt", "original bytes");
crate::backup::reset_capture_read_count(&path);
let mut capture = CapturedRegularFile::read(&path).unwrap().unwrap();
assert_eq!(crate::backup::capture_read_count(&path), 1);
let checkpoint = CheckpointFile::from_captured(&path, &mut capture).unwrap();
let mut backup = BackupStore::new();
backup
.snapshot_with_op_from_capture(
DEFAULT_SESSION_ID,
&path,
"shared capture",
Some("shared-op"),
&capture,
)
.unwrap();
let history = backup.history(DEFAULT_SESSION_ID, &path);
let CheckpointFileKind::Regular { bytes } = checkpoint.kind else {
panic!("regular capture must create a regular checkpoint");
};
assert_eq!(bytes.as_ref(), b"original bytes");
assert_eq!(history[0].content_bytes.as_ref(), b"original bytes");
assert!(Arc::ptr_eq(&bytes, &history[0].content_bytes));
assert_eq!(crate::backup::capture_read_count(&path), 1);
}
#[test]
fn stale_capture_refreshes_before_checkpoint_and_backup() {
let (path, _dir) = temp_file("stale-capture.txt", "old");
crate::backup::reset_capture_read_count(&path);
let mut capture = CapturedRegularFile::read(&path).unwrap().unwrap();
fs::write(&path, "fresh disk truth").unwrap();
let checkpoint = CheckpointFile::from_captured(&path, &mut capture).unwrap();
let mut backup = BackupStore::new();
backup
.snapshot_with_op_from_capture(
DEFAULT_SESSION_ID,
&path,
"freshened capture",
Some("fresh-op"),
&capture,
)
.unwrap();
let history = backup.history(DEFAULT_SESSION_ID, &path);
let CheckpointFileKind::Regular { bytes } = checkpoint.kind else {
panic!("regular capture must create a regular checkpoint");
};
assert_eq!(bytes.as_ref(), b"fresh disk truth");
assert_eq!(history[0].content_bytes.as_ref(), b"fresh disk truth");
assert!(Arc::ptr_eq(&bytes, &history[0].content_bytes));
assert_eq!(crate::backup::capture_read_count(&path), 2);
}
#[test]
fn checkpoint_restore_failure_removes_created_parent_dirs() {
let dir = tempfile::tempdir().unwrap();
let missing_root = dir.path().join("created");
let path_a = missing_root.join("nested").join("a.txt");
let path_b = dir.path().join("blocking-dir");
fs::create_dir(&path_b).unwrap();
let checkpoint = Checkpoint {
name: "dir-cleanup".to_string(),
file_contents: HashMap::from([
(path_a.clone(), checkpoint_file("checkpoint-a")),
(path_b.clone(), checkpoint_file("checkpoint-b")),
]),
created_at: current_timestamp(),
created_order: current_timestamp_nanos(),
};
let result = restore_paths_atomically(&checkpoint, &[path_a.clone(), path_b.clone()]);
assert!(
result.is_err(),
"second restore write should fail on directory"
);
assert!(!path_a.exists(), "restored file should be rolled back");
assert!(
!missing_root.exists(),
"new parent directories should be removed on rollback"
);
assert!(path_b.is_dir(), "pre-existing blocking directory remains");
}
}