use std::fs;
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Instant, SystemTime, UNIX_EPOCH};
use anyhow::{Context, Result};
use blake3::Hash;
use serde::{Deserialize, Serialize};
pub const L1_LARGE_FILE_BYTES: u64 = 1024 * 1024;
const JOURNAL_EXT: &str = ".atomwrite.journal.json";
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(tag = "phase", rename_all = "snake_case")]
pub enum JournalEntry {
Started {
op_id: String,
op: JournalOp,
target: String,
checksum_before: Option<String>,
checksum_after: String,
pid: u32,
started_at_unix: u64,
},
Committed {
op_id: String,
committed_at_unix: u64,
},
Aborted {
op_id: String,
aborted_at_unix: u64,
reason: String,
},
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum JournalOp {
Write,
Edit,
Replace,
Set,
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, clap::ValueEnum)]
#[serde(rename_all = "snake_case")]
#[clap(rename_all = "snake_case")]
pub enum WalPolicy {
#[default]
Auto,
Always,
Never,
}
impl WalPolicy {
pub const fn as_str(&self) -> &'static str {
match self {
Self::Auto => "auto",
Self::Always => "always",
Self::Never => "never",
}
}
}
pub fn should_create_sidecar(target: &Path, op: JournalOp, policy: WalPolicy) -> bool {
match policy {
WalPolicy::Never => false,
WalPolicy::Always => true,
WalPolicy::Auto => {
let size = std::fs::metadata(target).map(|m| m.len()).unwrap_or(0);
if size > L1_LARGE_FILE_BYTES {
return true;
}
if matches!(op, JournalOp::Edit | JournalOp::Replace) {
return true;
}
if !directory_is_git_tracked(target) {
return true;
}
if size <= 4096 {
return false;
}
false
}
}
}
fn directory_is_git_tracked(target: &Path) -> bool {
let start = target.parent().unwrap_or_else(|| Path::new("."));
let mut current = Some(start);
let mut depth = 0u8;
while let Some(dir) = current {
if dir.join(".git").exists() {
return true;
}
depth += 1;
if depth > 16 {
return false;
}
current = dir.parent();
}
false
}
pub mod heuristics {
use super::*;
pub type Decision = bool;
pub fn h1_ttl(journal_committed_at_unix: u64) -> Decision {
let ttl_secs: u64 = std::env::var("ATOMWRITE_WAL_KEEP_SECS")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(0);
if ttl_secs == 0 {
return false;
}
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let age = now.saturating_sub(journal_committed_at_unix);
age < ttl_secs
}
pub fn h2_lru_within_cap(workspace_committed_count: u64, age_rank: u64) -> Decision {
let max_count: u64 = std::env::var("ATOMWRITE_WAL_MAX_COUNT")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(100);
age_rank < max_count || workspace_committed_count <= max_count
}
pub fn h3_rate_limit() -> Decision {
let max_per_min: u64 = std::env::var("ATOMWRITE_WAL_RATE_LIMIT")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(10);
if max_per_min == 0 {
return false;
}
static WINDOW_START: AtomicU64 = AtomicU64::new(0);
static WINDOW_COUNT: AtomicU64 = AtomicU64::new(0);
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let start = WINDOW_START.load(Ordering::Relaxed);
if now.saturating_sub(start) >= 60 {
WINDOW_START.store(now, Ordering::Relaxed);
WINDOW_COUNT.store(1, Ordering::Relaxed);
return false;
}
let count = WINDOW_COUNT.fetch_add(1, Ordering::Relaxed) + 1;
count > max_per_min
}
pub fn h4_sentinel(target: &Path) -> Decision {
let dir = target.parent().unwrap_or_else(|| Path::new("."));
dir.join(".atomwrite_no_wal").exists()
}
pub fn h5_archive(journal_committed_at_unix: u64) -> Decision {
let archive_days: u64 = std::env::var("ATOMWRITE_WAL_ARCHIVE_DAYS")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(7);
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let age_days = now.saturating_sub(journal_committed_at_unix) / 86_400;
age_days >= archive_days
}
}
pub fn heuristics_should_preserve(
target: &Path,
journal_committed_at_unix: u64,
workspace_committed_count: u64,
age_rank: u64,
) -> bool {
use heuristics::*;
h1_ttl(journal_committed_at_unix)
|| h2_lru_within_cap(workspace_committed_count, age_rank)
|| h3_rate_limit()
|| h4_sentinel(target)
|| h5_archive(journal_committed_at_unix)
}
pub fn journal_path(target: &Path) -> PathBuf {
let dir = target.parent().unwrap_or_else(|| Path::new("."));
let basename = target
.file_name()
.and_then(|s| s.to_str())
.unwrap_or("unknown");
dir.join(format!(".atomwrite.journal.{}{}", basename, JOURNAL_EXT))
}
pub fn generate_op_id() -> String {
let pid = std::process::id();
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
let input = format!("{}-{}", pid, nanos);
blake3::hash(input.as_bytes())
.to_hex()
.as_str()
.chars()
.take(16)
.collect()
}
pub fn journal_started(
target: &Path,
op: JournalOp,
checksum_before: Option<Hash>,
checksum_after: Hash,
) -> Result<String> {
let op_id = generate_op_id();
let started_at_unix = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let entry = JournalEntry::Started {
op_id: op_id.clone(),
op,
target: target.display().to_string(),
checksum_before: checksum_before.map(|h| h.to_hex().to_string()),
checksum_after: checksum_after.to_hex().to_string(),
pid: std::process::id(),
started_at_unix,
};
append_entry(target, &entry)?;
Ok(op_id)
}
pub fn journal_committed(target: &Path, op_id: &str) -> Result<()> {
let committed_at_unix = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let entry = JournalEntry::Committed {
op_id: op_id.to_owned(),
committed_at_unix,
};
append_entry(target, &entry)
}
pub fn journal_aborted(target: &Path, op_id: &str, reason: &str) -> Result<()> {
let aborted_at_unix = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let entry = JournalEntry::Aborted {
op_id: op_id.to_owned(),
aborted_at_unix,
reason: reason.to_owned(),
};
append_entry(target, &entry)
}
fn append_entry(target: &Path, entry: &JournalEntry) -> Result<()> {
let path = journal_path(target);
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)
.with_context(|| format!("failed to create journal dir {}", parent.display()))?;
}
let mut file = fs::OpenOptions::new()
.create(true)
.append(true)
.open(&path)
.with_context(|| format!("failed to open journal {}", path.display()))?;
let json = serde_json::to_string(entry)
.with_context(|| format!("failed to serialize journal entry for {}", target.display()))?;
writeln!(file, "{}", json)
.with_context(|| format!("failed to write journal entry to {}", path.display()))?;
file.sync_data()
.with_context(|| format!("failed to fsync journal {}", path.display()))?;
Ok(())
}
#[derive(Debug)]
pub struct JournalGuard {
path: PathBuf,
keep_on_drop: bool,
op_id: Option<String>,
committed_at_unix: Option<u64>,
}
impl JournalGuard {
pub fn inert() -> Self {
Self {
path: PathBuf::new(),
keep_on_drop: true,
op_id: None,
committed_at_unix: None,
}
}
pub fn keep(&mut self) {
self.keep_on_drop = true;
}
pub fn release(&mut self) {
self.keep_on_drop = false;
self.committed_at_unix = Some(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0),
);
}
#[allow(dead_code)]
pub fn op_id(&self) -> Option<&str> {
self.op_id.as_deref()
}
}
impl Drop for JournalGuard {
fn drop(&mut self) {
if self.keep_on_drop {
return;
}
if self.path.as_os_str().is_empty() {
return;
}
let committed_at = self.committed_at_unix.unwrap_or(0);
if heuristics_should_preserve(&self.path, committed_at, u64::MAX, u64::MAX) {
tracing::debug!(
path = %self.path.display(),
"G119 L4: heuristics voted to preserve sidecar; skipping remove"
);
return;
}
if let Err(e) = fs::remove_file(&self.path) {
tracing::debug!(path = %self.path.display(), error = %e,
"journal guard: sidecar removal failed (will be reaped later)");
}
}
}
pub fn journal_started_with_guard(
target: &Path,
op: JournalOp,
checksum_before: Option<Hash>,
checksum_after: Hash,
) -> Result<(String, JournalGuard)> {
let op_id = journal_started(target, op, checksum_before, checksum_after)?;
let path = journal_path(target);
let guard = JournalGuard {
path,
keep_on_drop: true,
op_id: Some(op_id.clone()),
committed_at_unix: None,
};
Ok((op_id, guard))
}
#[derive(Debug, Clone, Serialize, schemars::JsonSchema)]
pub struct WalStats {
pub total_journals: u64,
pub by_state: WalStateBreakdown,
pub oldest_journal_age_secs: u64,
pub total_size_bytes: u64,
pub by_directory: Vec<WalDirEntry>,
pub auto_heal_recommended: bool,
pub estimated_reclaim_bytes: u64,
}
#[derive(Debug, Clone, Default, Serialize, schemars::JsonSchema)]
pub struct WalStateBreakdown {
pub started: u64,
pub committed: u64,
pub aborted: u64,
pub malformed: u64,
}
#[derive(Debug, Clone, Serialize, schemars::JsonSchema)]
pub struct WalDirEntry {
pub path: String,
pub count: u64,
}
pub fn compute_wal_stats(workspace: &Path) -> Result<WalStats> {
use std::collections::BTreeMap;
let mut total: u64 = 0;
let mut by_state = WalStateBreakdown::default();
let mut oldest_unix: u64 = u64::MAX;
let mut total_size: u64 = 0;
let mut by_dir: BTreeMap<String, u64> = BTreeMap::new();
for path in walk_journal_paths(workspace)? {
let meta = std::fs::metadata(&path).ok();
total += 1;
total_size += meta.as_ref().map(|m| m.len()).unwrap_or(0);
let rel_dir = path
.parent()
.and_then(|p| p.strip_prefix(workspace).ok())
.map(|p| p.display().to_string())
.unwrap_or_else(|| ".".to_string());
*by_dir.entry(rel_dir).or_insert(0) += 1;
let (state, last_unix) = parse_journal_state(&path).unwrap_or(("malformed", 0));
match state {
"Committed" => by_state.committed += 1,
"Aborted" => by_state.aborted += 1,
"Started" => by_state.started += 1,
_ => by_state.malformed += 1,
}
if state != "malformed" && last_unix < oldest_unix {
oldest_unix = last_unix;
}
}
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let oldest_age = if oldest_unix == u64::MAX {
0
} else {
now.saturating_sub(oldest_unix)
};
let mut by_directory: Vec<WalDirEntry> = by_dir
.into_iter()
.map(|(path, count)| WalDirEntry { path, count })
.collect();
by_directory.sort_by(|a, b| b.count.cmp(&a.count));
by_directory.truncate(10);
let auto_heal_recommended = total > 100 || oldest_age > 7 * 86_400;
let estimated_reclaim_bytes = if auto_heal_recommended { total_size } else { 0 };
Ok(WalStats {
total_journals: total,
by_state,
oldest_journal_age_secs: oldest_age,
total_size_bytes: total_size,
by_directory,
auto_heal_recommended,
estimated_reclaim_bytes,
})
}
pub fn walk_journal_paths(workspace: &Path) -> Result<Vec<PathBuf>> {
if !workspace.exists() {
return Ok(Vec::new());
}
let mut out = Vec::new();
for entry in walkdir::WalkDir::new(workspace)
.into_iter()
.filter_map(Result::ok)
{
let path = entry.path();
if !path.is_file() {
continue;
}
if let Some(name) = path.file_name().and_then(|n| n.to_str()) {
if name.ends_with(JOURNAL_EXT) {
out.push(path.to_path_buf());
}
}
}
Ok(out)
}
fn parse_journal_state(path: &Path) -> Option<(&'static str, u64)> {
let content = std::fs::read_to_string(path).ok()?;
let mut state = "malformed";
let mut last_unix: u64 = 0;
for line in content.lines() {
let val: serde_json::Value = match serde_json::from_str(line) {
Ok(v) => v,
Err(_) => {
state = "malformed";
continue;
}
};
let phase = val.get("phase").and_then(|v| v.as_str()).unwrap_or("");
match phase {
"started" => {
state = "Started";
last_unix = val
.get("started_at_unix")
.and_then(|v| v.as_u64())
.unwrap_or(0);
}
"committed" => {
state = "Committed";
last_unix = val
.get("committed_at_unix")
.and_then(|v| v.as_u64())
.unwrap_or(0);
}
"aborted" => {
state = "Aborted";
last_unix = val
.get("aborted_at_unix")
.and_then(|v| v.as_u64())
.unwrap_or(0);
}
_ => {
state = "malformed";
}
}
}
Some((state, last_unix))
}
#[derive(Debug, Clone, Serialize, schemars::JsonSchema)]
pub struct AutoHealReport {
pub removed: u64,
pub preserved: u64,
pub malformed: u64,
pub bytes_reclaimed: u64,
pub threshold_secs: u64,
}
pub fn auto_heal_on_startup(
workspace: &Path,
threshold_secs: u64,
max_duration_ms: u64,
) -> Result<AutoHealReport> {
let start = Instant::now();
let mut removed = 0u64;
let mut preserved = 0u64;
let mut malformed = 0u64;
let mut bytes_reclaimed = 0u64;
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
for path in walk_journal_paths(workspace)? {
if start.elapsed().as_millis() as u64 > max_duration_ms {
break;
}
let (state, last_unix) = match parse_journal_state(&path) {
Some(s) => s,
None => {
malformed += 1;
continue;
}
};
match state {
"Committed" | "Aborted" => {
let age = now.saturating_sub(last_unix);
if age > threshold_secs {
if let Ok(meta) = std::fs::metadata(&path) {
bytes_reclaimed += meta.len();
}
match std::fs::remove_file(&path) {
Ok(_) => removed += 1,
Err(_) => preserved += 1,
}
} else {
preserved += 1;
}
}
"Started" => preserved += 1,
_ => malformed += 1,
}
}
Ok(AutoHealReport {
removed,
preserved,
malformed,
bytes_reclaimed,
threshold_secs,
})
}
#[derive(Debug, Clone, Serialize)]
#[allow(clippy::struct_field_names)]
pub struct OrphanJournalReport {
pub journal_path: String,
pub target: String,
pub op_id: String,
pub op: JournalOp,
pub expected_new_checksum: String,
pub checksum_before: Option<String>,
pub started_at_unix: u64,
pub pid: u32,
}
pub fn recover_orphan_journals(dir: &Path) -> Result<Vec<OrphanJournalReport>> {
let mut reports = Vec::new();
if !dir.exists() {
return Ok(reports);
}
let entries =
fs::read_dir(dir).with_context(|| format!("failed to read dir {}", dir.display()))?;
for entry in entries.flatten() {
let path = entry.path();
if !path.is_file() {
continue;
}
let Some(name) = path.file_name().and_then(|s| s.to_str()) else {
continue;
};
if !name.starts_with(".atomwrite.journal.") || !name.ends_with(JOURNAL_EXT) {
continue;
}
match parse_orphan(&path) {
Ok(Some(report)) => reports.push(report),
Ok(None) => {
}
Err(e) => {
tracing::warn!(path = %path.display(), error = %e, "failed to parse journal");
}
}
}
Ok(reports)
}
fn parse_orphan(path: &Path) -> Result<Option<OrphanJournalReport>> {
let content = fs::read_to_string(path)
.with_context(|| format!("failed to read journal {}", path.display()))?;
let mut last_started: Option<JournalEntry> = None;
for line in content.lines() {
if line.trim().is_empty() {
continue;
}
let entry: JournalEntry = serde_json::from_str(line)
.with_context(|| format!("invalid JSON in journal {}", path.display()))?;
match &entry {
JournalEntry::Started { .. } => last_started = Some(entry),
JournalEntry::Committed { .. } | JournalEntry::Aborted { .. } => {
last_started = None;
}
}
}
let Some(last) = last_started else {
return Ok(None);
};
let JournalEntry::Started {
op_id,
op,
target,
checksum_before,
checksum_after,
pid,
started_at_unix,
} = last
else {
return Ok(None);
};
Ok(Some(OrphanJournalReport {
journal_path: path.display().to_string(),
target,
op_id,
op,
expected_new_checksum: checksum_after,
checksum_before,
started_at_unix,
pid,
}))
}
#[cfg(test)]
pub(crate) fn read_entries(path: &Path) -> Result<Vec<JournalEntry>> {
let content = fs::read_to_string(path)
.with_context(|| format!("failed to read journal {}", path.display()))?;
content
.lines()
.filter(|l| !l.trim().is_empty())
.map(|l| serde_json::from_str(l).context("invalid JSON"))
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
#[test]
fn journal_path_appends_atomwrite_journal_json() {
let target = Path::new("/tmp/foo.txt");
let jp = journal_path(target);
assert!(jp.ends_with(".atomwrite.journal.foo.txt.atomwrite.journal.json"));
}
#[test]
fn journal_started_creates_sidecar_and_records_op_id() {
let tmp = TempDir::new().unwrap();
let target = tmp.path().join("file.txt");
let before = blake3::hash(b"old");
let after = blake3::hash(b"new");
let op_id = journal_started(&target, JournalOp::Write, Some(before), after).unwrap();
assert_eq!(op_id.len(), 16);
let jp = journal_path(&target);
assert!(jp.exists());
let entries = read_entries(&jp).unwrap();
assert_eq!(entries.len(), 1);
let JournalEntry::Started {
op_id: recorded_id,
op,
target: t,
checksum_before: cb,
checksum_after: ca,
pid,
started_at_unix,
} = &entries[0]
else {
panic!("expected Started entry");
};
assert_eq!(recorded_id, &op_id);
assert_eq!(*op, JournalOp::Write);
assert_eq!(t, &target.display().to_string());
assert_eq!(cb.as_deref(), Some(before.to_hex().to_string().as_str()));
assert_eq!(ca, &after.to_hex().to_string());
assert_eq!(*pid, std::process::id());
assert!(*started_at_unix > 0);
}
#[test]
fn journal_committed_after_started_does_not_orphan() {
let tmp = TempDir::new().unwrap();
let target = tmp.path().join("file.txt");
let op_id = journal_started(&target, JournalOp::Edit, None, blake3::hash(b"x")).unwrap();
journal_committed(&target, &op_id).unwrap();
let reports = recover_orphan_journals(tmp.path()).unwrap();
assert!(
reports.is_empty(),
"expected zero orphans, got {:?}",
reports
);
}
#[test]
fn orphan_detected_when_started_without_committed() {
let tmp = TempDir::new().unwrap();
let target = tmp.path().join("file.txt");
let op_id = journal_started(
&target,
JournalOp::Write,
Some(blake3::hash(b"old")),
blake3::hash(b"new"),
)
.unwrap();
let reports = recover_orphan_journals(tmp.path()).unwrap();
assert_eq!(reports.len(), 1);
let r = &reports[0];
assert_eq!(r.op_id, op_id);
assert_eq!(r.op, JournalOp::Write);
assert_eq!(r.target, target.display().to_string());
assert!(r.checksum_before.is_some());
assert_eq!(r.pid, std::process::id());
}
#[test]
fn journal_aborted_clears_orphan() {
let tmp = TempDir::new().unwrap();
let target = tmp.path().join("file.txt");
let op_id = journal_started(&target, JournalOp::Replace, None, blake3::hash(b"x")).unwrap();
journal_aborted(&target, &op_id, "caller cancelled").unwrap();
let reports = recover_orphan_journals(tmp.path()).unwrap();
assert!(reports.is_empty());
}
#[test]
fn generate_op_id_is_16_hex_chars_and_unique() {
let a = generate_op_id();
let b = generate_op_id();
assert_eq!(a.len(), 16);
assert_eq!(b.len(), 16);
assert_ne!(a, b);
assert!(a.chars().all(|c| c.is_ascii_hexdigit()));
}
#[test]
fn recover_on_empty_dir_returns_empty() {
let tmp = TempDir::new().unwrap();
let reports = recover_orphan_journals(tmp.path()).unwrap();
assert!(reports.is_empty());
}
#[test]
fn recover_on_missing_dir_returns_empty() {
let missing = std::env::temp_dir().join("atomwrite-test-missing-dir-xyz");
let _ = fs::remove_dir_all(&missing);
let reports = recover_orphan_journals(&missing).unwrap();
assert!(reports.is_empty());
}
#[test]
fn l1_never_policy_always_returns_false() {
let tmp = TempDir::new().unwrap();
let target = tmp.path().join("big.bin");
std::fs::write(&target, vec![0u8; 5_000_000]).unwrap();
assert!(!should_create_sidecar(
&target,
JournalOp::Write,
WalPolicy::Never
));
assert!(!should_create_sidecar(
&target,
JournalOp::Edit,
WalPolicy::Never
));
}
#[test]
fn l1_always_policy_always_returns_true() {
let tmp = TempDir::new().unwrap();
let target = tmp.path().join("small.txt");
std::fs::write(&target, "x").unwrap();
assert!(should_create_sidecar(
&target,
JournalOp::Write,
WalPolicy::Always
));
assert!(should_create_sidecar(
&target,
JournalOp::Set,
WalPolicy::Always
));
}
#[test]
fn l1_auto_policy_returns_true_for_large_file() {
let tmp = TempDir::new().unwrap();
let target = tmp.path().join("huge.bin");
std::fs::write(&target, vec![0u8; (L1_LARGE_FILE_BYTES + 1) as usize]).unwrap();
assert!(should_create_sidecar(
&target,
JournalOp::Write,
WalPolicy::Auto
));
}
#[test]
fn l1_auto_policy_returns_true_for_edit_op() {
let tmp = TempDir::new().unwrap();
let target = tmp.path().join("code.rs");
std::fs::write(&target, "fn x() {}").unwrap();
assert!(should_create_sidecar(
&target,
JournalOp::Edit,
WalPolicy::Auto
));
assert!(should_create_sidecar(
&target,
JournalOp::Replace,
WalPolicy::Auto
));
}
#[test]
fn l1_auto_policy_skips_trivial_file() {
let tmp = TempDir::new().unwrap();
let parent = tmp.path().join("gitty");
std::fs::create_dir_all(parent.join(".git")).unwrap();
let target = parent.join("small.txt");
std::fs::write(&target, "hi").unwrap();
assert!(!should_create_sidecar(
&target,
JournalOp::Write,
WalPolicy::Auto
));
}
#[test]
fn l4_h1_ttl_default_zero_returns_false() {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
assert!(!heuristics::h1_ttl(now));
}
#[test]
fn l4_h2_lru_within_cap_returns_true_when_count_low() {
let result = heuristics::h2_lru_within_cap(50, 25);
assert!(
result,
"sidecar within the LRU cap must be preserved (default cap=100)"
);
}
#[test]
fn l4_h2_lru_returns_true_when_count_at_or_below_default_cap() {
let result = heuristics::h2_lru_within_cap(100, 99);
assert!(result, "sidecar at the LRU cap boundary must be preserved");
}
#[test]
fn l4_h3_rate_limit_returns_false_below_threshold() {
let result = heuristics::h3_rate_limit();
assert!(
!result,
"first call in a fresh window must not be throttled (default K=10/min)"
);
}
#[test]
fn l4_h4_sentinel_returns_true_when_file_exists() {
let tmp = TempDir::new().unwrap();
let target = tmp.path().join("data.txt");
std::fs::write(tmp.path().join(".atomwrite_no_wal"), "").unwrap();
assert!(heuristics::h4_sentinel(&target));
}
#[test]
fn l4_h4_sentinel_returns_false_when_absent() {
let tmp = TempDir::new().unwrap();
let target = tmp.path().join("data.txt");
assert!(!heuristics::h4_sentinel(&target));
}
#[test]
fn l4_h5_archive_returns_false_for_recent_journal_under_default() {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let one_day_ago = now.saturating_sub(86_400);
let result = heuristics::h5_archive(one_day_ago);
assert!(
!result,
"1-day-old journal is below the default 7-day archive threshold"
);
}
#[test]
fn l4_h5_archive_returns_true_for_journal_older_than_7_days() {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let eight_days_ago = now.saturating_sub(8 * 86_400);
let result = heuristics::h5_archive(eight_days_ago);
assert!(
result,
"8-day-old journal is past the 7-day archive threshold"
);
}
#[test]
fn l4_engine_returns_false_when_all_heuristics_disabled() {
let tmp = TempDir::new().unwrap();
let target = tmp.path().join("file.txt");
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let very_high_rank: u64 = 10_000;
let very_high_count: u64 = 10_000;
assert!(!heuristics::h2_lru_within_cap(
very_high_count,
very_high_rank
));
assert!(!heuristics::h5_archive(now));
let _ = heuristics_should_preserve(&target, now, 0, 0);
}
#[test]
fn l3_auto_heal_on_empty_workspace_reports_zero() {
let tmp = TempDir::new().unwrap();
let report = auto_heal_on_startup(tmp.path(), 3600, 100).unwrap();
assert_eq!(report.removed, 0);
assert_eq!(report.preserved, 0);
assert_eq!(report.malformed, 0);
assert_eq!(report.threshold_secs, 3600);
}
#[test]
fn l3_auto_heal_reaps_old_committed_preserves_started() {
let tmp = TempDir::new().unwrap();
let committed_path = tmp
.path()
.join(".atomwrite.journal.committed.atomwrite.journal.json");
let started_path = tmp
.path()
.join(".atomwrite.journal.started.atomwrite.journal.json");
let old_unix = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
.saturating_sub(10_000);
std::fs::write(
&committed_path,
format!(
"{{\"phase\":\"started\",\"op_id\":\"a\",\"op\":\"write\",\"target\":\"x\",\"checksum_before\":null,\"checksum_after\":\"b\",\"pid\":1,\"started_at_unix\":{old_unix}}}\n\
{{\"phase\":\"committed\",\"op_id\":\"a\",\"committed_at_unix\":{old_unix}}}\n"
),
)
.unwrap();
let started_unix = old_unix;
std::fs::write(
&started_path,
format!(
"{{\"phase\":\"started\",\"op_id\":\"b\",\"op\":\"write\",\"target\":\"y\",\"checksum_before\":null,\"checksum_after\":\"c\",\"pid\":1,\"started_at_unix\":{started_unix}}}\n"
),
)
.unwrap();
let report = auto_heal_on_startup(tmp.path(), 1, 100).unwrap();
assert_eq!(report.removed, 1, "exactly the old Committed is reaped");
assert_eq!(report.preserved, 1, "Started is preserved");
assert!(!committed_path.exists(), "Committed sidecar is gone");
assert!(started_path.exists(), "Started sidecar survives");
assert!(report.bytes_reclaimed > 0);
}
#[test]
fn l3_auto_heal_respects_budget() {
let tmp = TempDir::new().unwrap();
for i in 0..50 {
let path = tmp
.path()
.join(format!(".atomwrite.journal.file{i}.atomwrite.journal.json"));
std::fs::write(
&path,
format!("{{\"phase\":\"committed\",\"op_id\":\"x{i}\",\"committed_at_unix\":1}}\n"),
)
.unwrap();
}
let start = Instant::now();
let report = auto_heal_on_startup(tmp.path(), 1, 100).unwrap();
let elapsed = start.elapsed();
assert!(
elapsed.as_millis() < 1000,
"50-sidecar heal should complete in <1s (budget was 100ms, allowed slack for slow CI)"
);
assert_eq!(report.removed, 50);
}
#[test]
fn l4_release_records_committed_at_unix() {
let mut g = JournalGuard {
path: PathBuf::from("/tmp/.atomwrite.journal.x.atomwrite.journal.json"),
keep_on_drop: true,
op_id: Some("op_test".into()),
committed_at_unix: None,
};
g.release();
let recorded = g.committed_at_unix.expect("release must record timestamp");
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
assert!(now.abs_diff(recorded) <= 2);
}
#[test]
fn l4_drop_preserves_sidecar_when_h4_sentinel_votes() {
let tmp = TempDir::new().unwrap();
std::fs::write(tmp.path().join(".atomwrite_no_wal"), "").unwrap();
let sidecar = tmp
.path()
.join(".atomwrite.journal.x.atomwrite.journal.json");
std::fs::write(&sidecar, "stub").unwrap();
{
let mut g = JournalGuard {
path: sidecar.clone(),
keep_on_drop: true,
op_id: Some("op".into()),
committed_at_unix: None,
};
g.release();
}
assert!(
sidecar.exists(),
"L4 (h4_sentinel via .atomwrite_no_wal) must preserve the sidecar on drop"
);
}
#[test]
fn l4_drop_removes_sidecar_when_no_heuristic_preserves() {
let tmp = TempDir::new().unwrap();
let sidecar = tmp
.path()
.join(".atomwrite.journal.x.atomwrite.journal.json");
std::fs::write(&sidecar, "stub").unwrap();
{
let mut g = JournalGuard {
path: sidecar.clone(),
keep_on_drop: true,
op_id: Some("op".into()),
committed_at_unix: None,
};
g.release();
}
assert!(
!sidecar.exists(),
"L2 must reap the sidecar when no L4 heuristic votes to preserve"
);
}
}