use std::path::Path;
use std::time::{SystemTime, UNIX_EPOCH};
fn invalid_input(msg: &'static str) -> std::io::Error {
std::io::Error::new(std::io::ErrorKind::InvalidInput, msg)
}
pub fn try_atomic_write(
path: &Path,
bytes: &[u8],
permissions: Option<&std::fs::Permissions>,
) -> std::io::Result<()> {
try_atomic_write_if(path, bytes, permissions, || true)
}
pub fn try_atomic_write_if(
path: &Path,
bytes: &[u8],
permissions: Option<&std::fs::Permissions>,
may_commit: impl FnOnce() -> bool,
) -> std::io::Result<()> {
write_staged(path, bytes, permissions, may_commit, true)
}
pub fn try_atomic_write_unswept(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
write_staged(path, bytes, None, || true, false)
}
pub fn is_orphaned_temp(name: &str, modified: SystemTime, now: SystemTime) -> bool {
let Some(rest) = name
.strip_prefix('.')
.and_then(|name| name.split_once(".lean-ctx.tmp."))
.map(|(_, rest)| rest)
else {
return false;
};
let pid: Option<u32> = rest.split('.').next().and_then(|s| s.parse().ok());
pid.is_some_and(|p| !crate::process::is_alive(p))
|| now
.duration_since(modified)
.is_ok_and(|age| age > ORPHAN_AGE)
}
const ORPHAN_AGE: std::time::Duration = std::time::Duration::from_hours(1);
fn write_staged(
path: &Path,
bytes: &[u8],
permissions: Option<&std::fs::Permissions>,
may_commit: impl FnOnce() -> bool,
sweep: bool,
) -> std::io::Result<()> {
use std::io::Write;
let parent = path
.parent()
.ok_or_else(|| invalid_input("invalid path (no parent directory)"))?;
let filename = path
.file_name()
.ok_or_else(|| invalid_input("invalid path (no filename)"))?
.to_string_lossy();
if sweep {
cleanup_orphaned_temps(parent, &filename);
}
let pid = std::process::id();
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |d| d.as_nanos());
let tmp = parent.join(format!(".{filename}.lean-ctx.tmp.{pid}.{nanos}"));
{
let mut options = std::fs::OpenOptions::new();
options.write(true).create_new(true);
#[cfg(unix)]
if let Some(perms) = permissions {
use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
options.mode(perms.mode());
}
let mut f = options.open(&tmp)?;
f.write_all(bytes)?;
f.flush()?;
f.sync_all()?;
}
if let Some(perms) = permissions {
std::fs::set_permissions(&tmp, perms.clone())?;
}
if !may_commit() {
let _ = std::fs::remove_file(&tmp);
return Err(std::io::Error::new(
std::io::ErrorKind::TimedOut,
"atomic write was not admitted to commit",
));
}
#[cfg(windows)]
if let Err(e) = windows_replace(&tmp, path) {
let _ = std::fs::remove_file(&tmp);
return Err(e);
}
#[cfg(not(windows))]
if let Err(e) = std::fs::rename(&tmp, path) {
let _ = std::fs::remove_file(&tmp);
return Err(e);
}
#[cfg(unix)]
fsync_dir(parent);
Ok(())
}
pub fn replace_file(tmp: &Path, path: &Path) -> std::io::Result<()> {
#[cfg(windows)]
{
windows_replace(tmp, path)
}
#[cfg(not(windows))]
{
std::fs::rename(tmp, path)
}
}
#[cfg(windows)]
fn windows_replace(tmp: &Path, path: &Path) -> std::io::Result<()> {
use windows_sys::Win32::Storage::FileSystem::MOVEFILE_REPLACE_EXISTING;
windows_move(tmp, path, MOVEFILE_REPLACE_EXISTING)
}
#[cfg(windows)]
pub fn move_file_without_overwrite(source: &Path, destination: &Path) -> std::io::Result<()> {
use windows_sys::Win32::Storage::FileSystem::MOVEFILE_WRITE_THROUGH;
windows_move(source, destination, MOVEFILE_WRITE_THROUGH)
}
#[cfg(windows)]
fn windows_move(
tmp: &Path,
path: &Path,
flags: windows_sys::Win32::Storage::FileSystem::MOVE_FILE_FLAGS,
) -> std::io::Result<()> {
use std::os::windows::ffi::OsStrExt;
use windows_sys::Win32::Storage::FileSystem::MoveFileExW;
fn to_wide(p: &Path) -> Vec<u16> {
p.as_os_str()
.encode_wide()
.chain(std::iter::once(0))
.collect()
}
let tmp_w = to_wide(tmp);
let path_w = to_wide(path);
const ACCESS_DENIED: i32 = 5;
const SHARING_VIOLATION: i32 = 32;
let mut delay = std::time::Duration::from_millis(5);
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(1);
loop {
let ok = unsafe { MoveFileExW(tmp_w.as_ptr(), path_w.as_ptr(), flags) };
if ok != 0 {
return Ok(());
}
let error = std::io::Error::last_os_error();
let transient = matches!(
error.raw_os_error(),
Some(ACCESS_DENIED | SHARING_VIOLATION)
);
if !transient || std::time::Instant::now() >= deadline {
return Err(error);
}
std::thread::sleep(delay);
delay = (delay * 2).min(std::time::Duration::from_millis(100));
}
}
#[cfg(unix)]
fn fsync_dir(dir: &Path) {
if let Ok(f) = std::fs::File::open(dir) {
let _ = f.sync_all();
}
}
pub fn sync_directory(dir: &Path) -> std::io::Result<()> {
#[cfg(unix)]
{
std::fs::File::open(dir)?.sync_all()
}
#[cfg(not(unix))]
{
if std::fs::metadata(dir)?.is_dir() {
Ok(())
} else {
Err(std::io::Error::new(
std::io::ErrorKind::NotADirectory,
"sync target is not a directory",
))
}
}
}
#[cfg(any(test, feature = "test-support"))]
pub mod listings {
use std::cell::Cell;
thread_local! {
static SEEN: Cell<u64> = const { Cell::new(0) };
}
pub fn seen() {
SEEN.with(|n| n.set(n.get() + 1));
}
pub fn take() -> u64 {
SEEN.with(|n| n.replace(0))
}
}
fn cleanup_orphaned_temps(parent: &Path, filename: &str) {
const STALE_AGE: std::time::Duration = ORPHAN_AGE;
let prefix = format!(".{filename}.lean-ctx.tmp.");
#[cfg(any(test, feature = "test-support"))]
listings::seen();
let Ok(entries) = std::fs::read_dir(parent) else {
return;
};
let now = SystemTime::now();
for entry in entries.flatten() {
let name = entry.file_name();
let Some(rest) = name
.to_string_lossy()
.strip_prefix(&prefix)
.map(str::to_string)
else {
continue;
};
let pid: Option<u32> = rest.split('.').next().and_then(|s| s.parse().ok());
let pid_dead = pid.is_some_and(|p| !crate::process::is_alive(p));
let stale_by_age = entry
.metadata()
.and_then(|m| m.modified())
.ok()
.and_then(|m| now.duration_since(m).ok())
.is_some_and(|age| age > STALE_AGE);
if pid_dead || stale_by_age {
let _ = std::fs::remove_file(entry.path());
}
}
}
pub fn in_place_overwrite(
path: &Path,
bytes: &[u8],
permissions: Option<&std::fs::Permissions>,
) -> std::io::Result<()> {
use std::io::Write;
let mut opts = std::fs::OpenOptions::new();
opts.write(true).truncate(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
opts.custom_flags(libc::O_NOFOLLOW);
}
let mut f = opts.open(path)?;
f.write_all(bytes)?;
let _ = f.flush();
let _ = f.sync_all();
if let Some(perms) = permissions {
let _ = std::fs::set_permissions(path, perms.clone());
}
Ok(())
}
#[derive(Debug)]
pub enum StoreReadError {
Escapes,
Io(std::io::Error),
}
pub fn read_store_file(
store_root: &std::path::Path,
path: &std::path::Path,
) -> Result<String, StoreReadError> {
use std::io::Read;
let root = std::fs::canonicalize(store_root).map_err(StoreReadError::Io)?;
let parent = path.parent().ok_or(StoreReadError::Escapes)?;
let canonical_parent = std::fs::canonicalize(parent).map_err(StoreReadError::Io)?;
if !canonical_parent.starts_with(&root) {
return Err(StoreReadError::Escapes);
}
let link = std::fs::symlink_metadata(path).map_err(StoreReadError::Io)?;
if link.file_type().is_symlink() || !link.is_file() {
return Err(StoreReadError::Escapes);
}
let mut options = std::fs::OpenOptions::new();
options.read(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
options.custom_flags(libc::O_NOFOLLOW);
}
let mut file = options.open(path).map_err(|error| {
#[cfg(unix)]
if error.raw_os_error() == Some(libc::ELOOP) {
return StoreReadError::Escapes;
}
StoreReadError::Io(error)
})?;
let opened = file.metadata().map_err(StoreReadError::Io)?;
if !opened.is_file() {
return Err(StoreReadError::Escapes);
}
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
if opened.dev() != link.dev() || opened.ino() != link.ino() {
return Err(StoreReadError::Escapes);
}
}
let mut content = String::new();
file.read_to_string(&mut content)
.map_err(StoreReadError::Io)?;
Ok(content)
}
pub fn is_readonly_dir_error(e: &std::io::Error) -> bool {
if e.kind() == std::io::ErrorKind::PermissionDenied {
return true;
}
#[cfg(unix)]
{
matches!(
e.raw_os_error(),
Some(libc::EROFS | libc::EACCES | libc::EPERM)
)
}
#[cfg(not(unix))]
{
false
}
}
pub fn write_bytes_with_fallback(
path: &Path,
bytes: &[u8],
permissions: Option<&std::fs::Permissions>,
) -> Result<(), String> {
match try_atomic_write(path, bytes, permissions) {
Ok(()) => Ok(()),
Err(e) if is_readonly_dir_error(&e) && path.is_file() => {
in_place_overwrite(path, bytes, permissions).map_err(|fallback_err| {
format!(
"atomic write failed ({e}); in-place fallback also failed: {fallback_err} ({})",
path.display()
)
})
}
Err(e) => Err(format!("atomic write failed: {e} ({})", path.display())),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn readonly_dir_error_classification() {
assert!(is_readonly_dir_error(&std::io::Error::from(
std::io::ErrorKind::PermissionDenied
)));
assert!(!is_readonly_dir_error(&std::io::Error::from(
std::io::ErrorKind::NotFound
)));
#[cfg(unix)]
{
assert!(is_readonly_dir_error(&std::io::Error::from_raw_os_error(
libc::EROFS
)));
assert!(is_readonly_dir_error(&std::io::Error::from_raw_os_error(
libc::EACCES
)));
assert!(is_readonly_dir_error(&std::io::Error::from_raw_os_error(
libc::EPERM
)));
}
}
#[cfg(unix)]
#[test]
fn fsync_dir_succeeds_on_a_real_directory() {
let dir = tempfile::tempdir().unwrap();
fsync_dir(dir.path());
}
#[test]
fn cleanup_orphaned_temps_removes_dead_pid_leftovers() {
let dir = tempfile::tempdir().unwrap();
let filename = "cfg.toml";
let dead_pid = 999_999_999u32;
let orphan = dir
.path()
.join(format!(".{filename}.lean-ctx.tmp.{dead_pid}.123"));
std::fs::write(&orphan, b"stale").unwrap();
cleanup_orphaned_temps(dir.path(), filename);
assert!(
!orphan.exists(),
"orphaned temp with a dead PID must be removed"
);
}
#[test]
fn cleanup_orphaned_temps_keeps_fresh_temp_from_a_live_pid() {
let dir = tempfile::tempdir().unwrap();
let filename = "cfg.toml";
let my_pid = std::process::id();
let in_progress = dir
.path()
.join(format!(".{filename}.lean-ctx.tmp.{my_pid}.123"));
std::fs::write(&in_progress, b"still writing").unwrap();
cleanup_orphaned_temps(dir.path(), filename);
assert!(
in_progress.exists(),
"a live writer's own in-progress temp must survive"
);
}
#[test]
fn cleanup_orphaned_temps_removes_old_temp_even_with_a_live_pid() {
let dir = tempfile::tempdir().unwrap();
let filename = "cfg.toml";
let my_pid = std::process::id();
let ancient = dir
.path()
.join(format!(".{filename}.lean-ctx.tmp.{my_pid}.123"));
std::fs::write(&ancient, b"ancient").unwrap();
filetime::set_file_mtime(&ancient, filetime::FileTime::from_unix_time(0, 0)).unwrap();
cleanup_orphaned_temps(dir.path(), filename);
assert!(
!ancient.exists(),
"ancient temp must be removed regardless of PID liveness"
);
}
#[test]
fn cleanup_orphaned_temps_ignores_unrelated_files() {
let dir = tempfile::tempdir().unwrap();
let filename = "cfg.toml";
let unrelated = dir.path().join("other-file.txt");
std::fs::write(&unrelated, b"keep me").unwrap();
cleanup_orphaned_temps(dir.path(), filename);
assert!(unrelated.exists(), "non-matching files must be left alone");
}
#[test]
fn try_atomic_write_creates_and_replaces() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("cfg.toml");
try_atomic_write(&path, b"first", None).unwrap();
assert_eq!(std::fs::read(&path).unwrap(), b"first");
let strays: Vec<_> = std::fs::read_dir(dir.path())
.unwrap()
.flatten()
.filter(|e| e.file_name().to_string_lossy().contains(".lean-ctx.tmp."))
.collect();
assert!(strays.is_empty(), "temp file must not linger");
try_atomic_write(&path, b"second", None).unwrap();
assert_eq!(std::fs::read(&path).unwrap(), b"second");
let refused = try_atomic_write_if(&path, b"third", None, || false).unwrap_err();
assert_eq!(refused.kind(), std::io::ErrorKind::TimedOut);
assert_eq!(std::fs::read(&path).unwrap(), b"second");
assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 1);
}
#[cfg(unix)]
#[test]
fn in_place_overwrite_truncates_existing_file() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.jsonc");
std::fs::write(&path, b"longer original content").unwrap();
in_place_overwrite(&path, b"short", None).unwrap();
assert_eq!(std::fs::read(&path).unwrap(), b"short");
}
#[cfg(unix)]
#[test]
fn fallback_overwrites_when_parent_dir_is_readonly() {
use std::os::unix::fs::PermissionsExt;
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("cfg.toml");
std::fs::write(&path, b"original").unwrap();
std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o500)).unwrap();
let res = write_bytes_with_fallback(&path, b"updated", None);
let _ = std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o700));
res.expect("read-only-dir fallback must succeed");
assert_eq!(std::fs::read(&path).unwrap(), b"updated");
}
#[cfg(windows)]
fn hold_without_share_delete(
path: &Path,
hold: std::time::Duration,
) -> std::thread::JoinHandle<()> {
use std::os::windows::fs::OpenOptionsExt;
use windows_sys::Win32::Storage::FileSystem::FILE_SHARE_READ;
let handle = std::fs::OpenOptions::new()
.read(true)
.share_mode(FILE_SHARE_READ)
.open(path)
.expect("hold");
std::thread::spawn(move || {
std::thread::sleep(hold);
drop(handle);
})
}
#[cfg(windows)]
#[test]
fn publish_without_overwrite_waits_out_a_held_source() {
let dir = tempfile::tempdir().unwrap();
let source = dir.path().join(".journal.tmp");
let destination = dir.path().join("journal");
std::fs::write(&source, b"journal").unwrap();
let held = hold_without_share_delete(&source, std::time::Duration::from_millis(300));
move_file_without_overwrite(&source, &destination).expect("published after the hold");
held.join().unwrap();
assert_eq!(std::fs::read(&destination).unwrap(), b"journal");
assert!(!source.exists());
}
#[cfg(windows)]
#[test]
fn publish_without_overwrite_refuses_an_existing_destination_at_once() {
let dir = tempfile::tempdir().unwrap();
let source = dir.path().join(".journal.tmp");
let destination = dir.path().join("journal");
std::fs::write(&source, b"new").unwrap();
std::fs::write(&destination, b"owned").unwrap();
let started = std::time::Instant::now();
let error = move_file_without_overwrite(&source, &destination).unwrap_err();
assert_eq!(error.kind(), std::io::ErrorKind::AlreadyExists, "{error}");
assert!(started.elapsed() < std::time::Duration::from_millis(500));
assert_eq!(std::fs::read(&destination).unwrap(), b"owned");
assert!(source.exists(), "the caller removes the source");
}
#[cfg(windows)]
#[test]
fn replace_waits_out_a_held_target() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.toml");
std::fs::write(&path, b"old").unwrap();
let held = hold_without_share_delete(&path, std::time::Duration::from_millis(300));
try_atomic_write(&path, b"new", None).expect("replaced after the hold");
held.join().unwrap();
assert_eq!(std::fs::read(&path).unwrap(), b"new");
}
}