1use std::path::Path;
17use std::time::{SystemTime, UNIX_EPOCH};
18
19fn invalid_input(msg: &'static str) -> std::io::Error {
20 std::io::Error::new(std::io::ErrorKind::InvalidInput, msg)
21}
22
23pub fn try_atomic_write(
28 path: &Path,
29 bytes: &[u8],
30 permissions: Option<&std::fs::Permissions>,
31) -> std::io::Result<()> {
32 try_atomic_write_if(path, bytes, permissions, || true)
33}
34
35pub fn try_atomic_write_if(
42 path: &Path,
43 bytes: &[u8],
44 permissions: Option<&std::fs::Permissions>,
45 may_commit: impl FnOnce() -> bool,
46) -> std::io::Result<()> {
47 write_staged(path, bytes, permissions, may_commit, true)
48}
49
50pub fn try_atomic_write_unswept(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
55 write_staged(path, bytes, None, || true, false)
56}
57
58pub fn is_orphaned_temp(name: &str, modified: SystemTime, now: SystemTime) -> bool {
61 let Some(rest) = name
62 .strip_prefix('.')
63 .and_then(|name| name.split_once(".lean-ctx.tmp."))
64 .map(|(_, rest)| rest)
65 else {
66 return false;
67 };
68 let pid: Option<u32> = rest.split('.').next().and_then(|s| s.parse().ok());
69 pid.is_some_and(|p| !crate::process::is_alive(p))
70 || now
71 .duration_since(modified)
72 .is_ok_and(|age| age > ORPHAN_AGE)
73}
74
75const ORPHAN_AGE: std::time::Duration = std::time::Duration::from_hours(1);
76
77fn write_staged(
78 path: &Path,
79 bytes: &[u8],
80 permissions: Option<&std::fs::Permissions>,
81 may_commit: impl FnOnce() -> bool,
82 sweep: bool,
83) -> std::io::Result<()> {
84 use std::io::Write;
85
86 let parent = path
87 .parent()
88 .ok_or_else(|| invalid_input("invalid path (no parent directory)"))?;
89 let filename = path
90 .file_name()
91 .ok_or_else(|| invalid_input("invalid path (no filename)"))?
92 .to_string_lossy();
93
94 if sweep {
101 cleanup_orphaned_temps(parent, &filename);
102 }
103
104 let pid = std::process::id();
105 let nanos = SystemTime::now()
106 .duration_since(UNIX_EPOCH)
107 .map_or(0, |d| d.as_nanos());
108 let tmp = parent.join(format!(".{filename}.lean-ctx.tmp.{pid}.{nanos}"));
109
110 {
111 let mut options = std::fs::OpenOptions::new();
112 options.write(true).create_new(true);
113 #[cfg(unix)]
114 if let Some(perms) = permissions {
115 use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
116 options.mode(perms.mode());
117 }
118 let mut f = options.open(&tmp)?;
119 f.write_all(bytes)?;
120 f.flush()?;
121 f.sync_all()?;
122 }
123
124 if let Some(perms) = permissions {
125 std::fs::set_permissions(&tmp, perms.clone())?;
126 }
127
128 if !may_commit() {
129 let _ = std::fs::remove_file(&tmp);
130 return Err(std::io::Error::new(
131 std::io::ErrorKind::TimedOut,
132 "atomic write was not admitted to commit",
133 ));
134 }
135
136 #[cfg(windows)]
145 if let Err(e) = windows_replace(&tmp, path) {
146 let _ = std::fs::remove_file(&tmp);
147 return Err(e);
148 }
149
150 #[cfg(not(windows))]
151 if let Err(e) = std::fs::rename(&tmp, path) {
152 let _ = std::fs::remove_file(&tmp);
155 return Err(e);
156 }
157
158 #[cfg(unix)]
159 fsync_dir(parent);
160
161 Ok(())
162}
163
164pub fn replace_file(tmp: &Path, path: &Path) -> std::io::Result<()> {
168 #[cfg(windows)]
169 {
170 windows_replace(tmp, path)
171 }
172 #[cfg(not(windows))]
173 {
174 std::fs::rename(tmp, path)
175 }
176}
177
178#[cfg(windows)]
181fn windows_replace(tmp: &Path, path: &Path) -> std::io::Result<()> {
182 use windows_sys::Win32::Storage::FileSystem::MOVEFILE_REPLACE_EXISTING;
183 windows_move(tmp, path, MOVEFILE_REPLACE_EXISTING)
184}
185
186#[cfg(windows)]
193pub fn move_file_without_overwrite(source: &Path, destination: &Path) -> std::io::Result<()> {
194 use windows_sys::Win32::Storage::FileSystem::MOVEFILE_WRITE_THROUGH;
195 windows_move(source, destination, MOVEFILE_WRITE_THROUGH)
197}
198
199#[cfg(windows)]
201fn windows_move(
202 tmp: &Path,
203 path: &Path,
204 flags: windows_sys::Win32::Storage::FileSystem::MOVE_FILE_FLAGS,
205) -> std::io::Result<()> {
206 use std::os::windows::ffi::OsStrExt;
207 use windows_sys::Win32::Storage::FileSystem::MoveFileExW;
208
209 fn to_wide(p: &Path) -> Vec<u16> {
210 p.as_os_str()
211 .encode_wide()
212 .chain(std::iter::once(0))
213 .collect()
214 }
215
216 let tmp_w = to_wide(tmp);
217 let path_w = to_wide(path);
218
219 const ACCESS_DENIED: i32 = 5;
227 const SHARING_VIOLATION: i32 = 32;
228 let mut delay = std::time::Duration::from_millis(5);
229 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(1);
230 loop {
231 let ok = unsafe { MoveFileExW(tmp_w.as_ptr(), path_w.as_ptr(), flags) };
234 if ok != 0 {
235 return Ok(());
236 }
237 let error = std::io::Error::last_os_error();
238 let transient = matches!(
239 error.raw_os_error(),
240 Some(ACCESS_DENIED | SHARING_VIOLATION)
241 );
242 if !transient || std::time::Instant::now() >= deadline {
243 return Err(error);
244 }
245 std::thread::sleep(delay);
246 delay = (delay * 2).min(std::time::Duration::from_millis(100));
247 }
248}
249
250#[cfg(unix)]
253fn fsync_dir(dir: &Path) {
254 if let Ok(f) = std::fs::File::open(dir) {
255 let _ = f.sync_all();
256 }
257}
258
259pub fn sync_directory(dir: &Path) -> std::io::Result<()> {
266 #[cfg(unix)]
267 {
268 std::fs::File::open(dir)?.sync_all()
269 }
270 #[cfg(not(unix))]
271 {
272 if std::fs::metadata(dir)?.is_dir() {
273 Ok(())
274 } else {
275 Err(std::io::Error::new(
276 std::io::ErrorKind::NotADirectory,
277 "sync target is not a directory",
278 ))
279 }
280 }
281}
282
283#[cfg(any(test, feature = "test-support"))]
286pub mod listings {
287 use std::cell::Cell;
288
289 thread_local! {
290 static SEEN: Cell<u64> = const { Cell::new(0) };
291 }
292
293 pub fn seen() {
294 SEEN.with(|n| n.set(n.get() + 1));
295 }
296
297 pub fn take() -> u64 {
299 SEEN.with(|n| n.replace(0))
300 }
301}
302
303fn cleanup_orphaned_temps(parent: &Path, filename: &str) {
306 const STALE_AGE: std::time::Duration = ORPHAN_AGE;
307 let prefix = format!(".{filename}.lean-ctx.tmp.");
308 #[cfg(any(test, feature = "test-support"))]
309 listings::seen();
310
311 let Ok(entries) = std::fs::read_dir(parent) else {
312 return;
313 };
314 let now = SystemTime::now();
315 for entry in entries.flatten() {
316 let name = entry.file_name();
317 let Some(rest) = name
318 .to_string_lossy()
319 .strip_prefix(&prefix)
320 .map(str::to_string)
321 else {
322 continue;
323 };
324 let pid: Option<u32> = rest.split('.').next().and_then(|s| s.parse().ok());
325 let pid_dead = pid.is_some_and(|p| !crate::process::is_alive(p));
326 let stale_by_age = entry
327 .metadata()
328 .and_then(|m| m.modified())
329 .ok()
330 .and_then(|m| now.duration_since(m).ok())
331 .is_some_and(|age| age > STALE_AGE);
332 if pid_dead || stale_by_age {
333 let _ = std::fs::remove_file(entry.path());
334 }
335 }
336}
337
338pub fn in_place_overwrite(
343 path: &Path,
344 bytes: &[u8],
345 permissions: Option<&std::fs::Permissions>,
346) -> std::io::Result<()> {
347 use std::io::Write;
348
349 let mut opts = std::fs::OpenOptions::new();
350 opts.write(true).truncate(true);
351 #[cfg(unix)]
352 {
353 use std::os::unix::fs::OpenOptionsExt;
354 opts.custom_flags(libc::O_NOFOLLOW);
357 }
358
359 let mut f = opts.open(path)?;
360 f.write_all(bytes)?;
361 let _ = f.flush();
362 let _ = f.sync_all();
363
364 if let Some(perms) = permissions {
365 let _ = std::fs::set_permissions(path, perms.clone());
366 }
367 Ok(())
368}
369
370#[derive(Debug)]
372pub enum StoreReadError {
373 Escapes,
376 Io(std::io::Error),
377}
378
379pub fn read_store_file(
387 store_root: &std::path::Path,
388 path: &std::path::Path,
389) -> Result<String, StoreReadError> {
390 use std::io::Read;
391
392 let root = std::fs::canonicalize(store_root).map_err(StoreReadError::Io)?;
393 let parent = path.parent().ok_or(StoreReadError::Escapes)?;
394 let canonical_parent = std::fs::canonicalize(parent).map_err(StoreReadError::Io)?;
395 if !canonical_parent.starts_with(&root) {
396 return Err(StoreReadError::Escapes);
397 }
398 let link = std::fs::symlink_metadata(path).map_err(StoreReadError::Io)?;
399 if link.file_type().is_symlink() || !link.is_file() {
400 return Err(StoreReadError::Escapes);
401 }
402 let mut options = std::fs::OpenOptions::new();
403 options.read(true);
404 #[cfg(unix)]
405 {
406 use std::os::unix::fs::OpenOptionsExt;
407 options.custom_flags(libc::O_NOFOLLOW);
408 }
409 let mut file = options.open(path).map_err(|error| {
410 #[cfg(unix)]
412 if error.raw_os_error() == Some(libc::ELOOP) {
413 return StoreReadError::Escapes;
414 }
415 StoreReadError::Io(error)
416 })?;
417 let opened = file.metadata().map_err(StoreReadError::Io)?;
418 if !opened.is_file() {
419 return Err(StoreReadError::Escapes);
420 }
421 #[cfg(unix)]
422 {
423 use std::os::unix::fs::MetadataExt;
424 if opened.dev() != link.dev() || opened.ino() != link.ino() {
425 return Err(StoreReadError::Escapes);
426 }
427 }
428 let mut content = String::new();
429 file.read_to_string(&mut content)
430 .map_err(StoreReadError::Io)?;
431 Ok(content)
432}
433
434pub fn is_readonly_dir_error(e: &std::io::Error) -> bool {
438 if e.kind() == std::io::ErrorKind::PermissionDenied {
439 return true;
440 }
441 #[cfg(unix)]
442 {
443 matches!(
444 e.raw_os_error(),
445 Some(libc::EROFS | libc::EACCES | libc::EPERM)
446 )
447 }
448 #[cfg(not(unix))]
449 {
450 false
451 }
452}
453
454pub fn write_bytes_with_fallback(
459 path: &Path,
460 bytes: &[u8],
461 permissions: Option<&std::fs::Permissions>,
462) -> Result<(), String> {
463 match try_atomic_write(path, bytes, permissions) {
464 Ok(()) => Ok(()),
465 Err(e) if is_readonly_dir_error(&e) && path.is_file() => {
466 in_place_overwrite(path, bytes, permissions).map_err(|fallback_err| {
467 format!(
468 "atomic write failed ({e}); in-place fallback also failed: {fallback_err} ({})",
469 path.display()
470 )
471 })
472 }
473 Err(e) => Err(format!("atomic write failed: {e} ({})", path.display())),
474 }
475}
476
477#[cfg(test)]
478mod tests {
479 use super::*;
480
481 #[test]
482 fn readonly_dir_error_classification() {
483 assert!(is_readonly_dir_error(&std::io::Error::from(
484 std::io::ErrorKind::PermissionDenied
485 )));
486 assert!(!is_readonly_dir_error(&std::io::Error::from(
487 std::io::ErrorKind::NotFound
488 )));
489 #[cfg(unix)]
490 {
491 assert!(is_readonly_dir_error(&std::io::Error::from_raw_os_error(
492 libc::EROFS
493 )));
494 assert!(is_readonly_dir_error(&std::io::Error::from_raw_os_error(
495 libc::EACCES
496 )));
497 assert!(is_readonly_dir_error(&std::io::Error::from_raw_os_error(
498 libc::EPERM
499 )));
500 }
501 }
502
503 #[cfg(unix)]
504 #[test]
505 fn fsync_dir_succeeds_on_a_real_directory() {
506 let dir = tempfile::tempdir().unwrap();
507 fsync_dir(dir.path());
508 }
509
510 #[test]
513 fn cleanup_orphaned_temps_removes_dead_pid_leftovers() {
514 let dir = tempfile::tempdir().unwrap();
515 let filename = "cfg.toml";
516 let dead_pid = 999_999_999u32;
517 let orphan = dir
518 .path()
519 .join(format!(".{filename}.lean-ctx.tmp.{dead_pid}.123"));
520 std::fs::write(&orphan, b"stale").unwrap();
521
522 cleanup_orphaned_temps(dir.path(), filename);
523
524 assert!(
525 !orphan.exists(),
526 "orphaned temp with a dead PID must be removed"
527 );
528 }
529
530 #[test]
531 fn cleanup_orphaned_temps_keeps_fresh_temp_from_a_live_pid() {
532 let dir = tempfile::tempdir().unwrap();
533 let filename = "cfg.toml";
534 let my_pid = std::process::id();
535 let in_progress = dir
536 .path()
537 .join(format!(".{filename}.lean-ctx.tmp.{my_pid}.123"));
538 std::fs::write(&in_progress, b"still writing").unwrap();
539
540 cleanup_orphaned_temps(dir.path(), filename);
541
542 assert!(
543 in_progress.exists(),
544 "a live writer's own in-progress temp must survive"
545 );
546 }
547
548 #[test]
549 fn cleanup_orphaned_temps_removes_old_temp_even_with_a_live_pid() {
550 let dir = tempfile::tempdir().unwrap();
551 let filename = "cfg.toml";
552 let my_pid = std::process::id();
553 let ancient = dir
554 .path()
555 .join(format!(".{filename}.lean-ctx.tmp.{my_pid}.123"));
556 std::fs::write(&ancient, b"ancient").unwrap();
557 filetime::set_file_mtime(&ancient, filetime::FileTime::from_unix_time(0, 0)).unwrap();
558
559 cleanup_orphaned_temps(dir.path(), filename);
560
561 assert!(
562 !ancient.exists(),
563 "ancient temp must be removed regardless of PID liveness"
564 );
565 }
566
567 #[test]
568 fn cleanup_orphaned_temps_ignores_unrelated_files() {
569 let dir = tempfile::tempdir().unwrap();
570 let filename = "cfg.toml";
571 let unrelated = dir.path().join("other-file.txt");
572 std::fs::write(&unrelated, b"keep me").unwrap();
573
574 cleanup_orphaned_temps(dir.path(), filename);
575
576 assert!(unrelated.exists(), "non-matching files must be left alone");
577 }
578 #[test]
579 fn try_atomic_write_creates_and_replaces() {
580 let dir = tempfile::tempdir().unwrap();
581 let path = dir.path().join("cfg.toml");
582 try_atomic_write(&path, b"first", None).unwrap();
583 assert_eq!(std::fs::read(&path).unwrap(), b"first");
584 let strays: Vec<_> = std::fs::read_dir(dir.path())
586 .unwrap()
587 .flatten()
588 .filter(|e| e.file_name().to_string_lossy().contains(".lean-ctx.tmp."))
589 .collect();
590 assert!(strays.is_empty(), "temp file must not linger");
591 try_atomic_write(&path, b"second", None).unwrap();
592 assert_eq!(std::fs::read(&path).unwrap(), b"second");
593
594 let refused = try_atomic_write_if(&path, b"third", None, || false).unwrap_err();
597 assert_eq!(refused.kind(), std::io::ErrorKind::TimedOut);
598 assert_eq!(std::fs::read(&path).unwrap(), b"second");
599 assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 1);
600 }
601
602 #[cfg(unix)]
603 #[test]
604 fn in_place_overwrite_truncates_existing_file() {
605 let dir = tempfile::tempdir().unwrap();
606 let path = dir.path().join("config.jsonc");
607 std::fs::write(&path, b"longer original content").unwrap();
608 in_place_overwrite(&path, b"short", None).unwrap();
609 assert_eq!(std::fs::read(&path).unwrap(), b"short");
610 }
611
612 #[cfg(unix)]
613 #[test]
614 fn fallback_overwrites_when_parent_dir_is_readonly() {
615 use std::os::unix::fs::PermissionsExt;
616 let dir = tempfile::tempdir().unwrap();
617 let path = dir.path().join("cfg.toml");
618 std::fs::write(&path, b"original").unwrap();
619 std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o500)).unwrap();
622 let res = write_bytes_with_fallback(&path, b"updated", None);
623 let _ = std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o700));
624 res.expect("read-only-dir fallback must succeed");
625 assert_eq!(std::fs::read(&path).unwrap(), b"updated");
626 }
627
628 #[cfg(windows)]
631 fn hold_without_share_delete(
632 path: &Path,
633 hold: std::time::Duration,
634 ) -> std::thread::JoinHandle<()> {
635 use std::os::windows::fs::OpenOptionsExt;
636 use windows_sys::Win32::Storage::FileSystem::FILE_SHARE_READ;
637 let handle = std::fs::OpenOptions::new()
638 .read(true)
639 .share_mode(FILE_SHARE_READ)
640 .open(path)
641 .expect("hold");
642 std::thread::spawn(move || {
643 std::thread::sleep(hold);
644 drop(handle);
645 })
646 }
647
648 #[cfg(windows)]
651 #[test]
652 fn publish_without_overwrite_waits_out_a_held_source() {
653 let dir = tempfile::tempdir().unwrap();
654 let source = dir.path().join(".journal.tmp");
655 let destination = dir.path().join("journal");
656 std::fs::write(&source, b"journal").unwrap();
657 let held = hold_without_share_delete(&source, std::time::Duration::from_millis(300));
658 move_file_without_overwrite(&source, &destination).expect("published after the hold");
659 held.join().unwrap();
660 assert_eq!(std::fs::read(&destination).unwrap(), b"journal");
661 assert!(!source.exists());
662 }
663
664 #[cfg(windows)]
667 #[test]
668 fn publish_without_overwrite_refuses_an_existing_destination_at_once() {
669 let dir = tempfile::tempdir().unwrap();
670 let source = dir.path().join(".journal.tmp");
671 let destination = dir.path().join("journal");
672 std::fs::write(&source, b"new").unwrap();
673 std::fs::write(&destination, b"owned").unwrap();
674 let started = std::time::Instant::now();
675 let error = move_file_without_overwrite(&source, &destination).unwrap_err();
676 assert_eq!(error.kind(), std::io::ErrorKind::AlreadyExists, "{error}");
677 assert!(started.elapsed() < std::time::Duration::from_millis(500));
678 assert_eq!(std::fs::read(&destination).unwrap(), b"owned");
679 assert!(source.exists(), "the caller removes the source");
680 }
681
682 #[cfg(windows)]
684 #[test]
685 fn replace_waits_out_a_held_target() {
686 let dir = tempfile::tempdir().unwrap();
687 let path = dir.path().join("config.toml");
688 std::fs::write(&path, b"old").unwrap();
689 let held = hold_without_share_delete(&path, std::time::Duration::from_millis(300));
690 try_atomic_write(&path, b"new", None).expect("replaced after the hold");
691 held.join().unwrap();
692 assert_eq!(std::fs::read(&path).unwrap(), b"new");
693 }
694}