modelshelf 0.1.0

//! Advisory file locking and crash-safe atomic writes.
//!
//! Locks are always taken on the dedicated `registry.lock` file, never on
//! `registry.json` itself: the manifest is replaced via atomic rename, and a
//! lock held on a file that gets renamed away would protect nothing.
//!
//! Advisory locks coordinate processes that follow the convention. They do
//! not protect against writers that ignore it.

use std::fs::{File, OpenOptions, TryLockError};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

use crate::{Error, Result};

/// Default time to wait for the registry lock before failing with
/// [`Error::RegistryBusy`].
pub const DEFAULT_LOCK_TIMEOUT: Duration = Duration::from_secs(5);

const RETRY_INTERVAL: Duration = Duration::from_millis(25);

/// A held advisory lock. Released on drop.
#[derive(Debug)]
pub struct LockGuard {
    file: File,
}

impl Drop for LockGuard {
    fn drop(&mut self) {
        // Errors on unlock are unrecoverable and the OS releases the lock
        // when the descriptor closes anyway.
        let _ = self.file.unlock();
    }
}

fn open_lock_file(path: &Path) -> Result<File> {
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent).map_err(|e| Error::io(parent, e))?;
    }
    OpenOptions::new()
        .create(true)
        .read(true)
        .append(true) // never truncate; the file stays zero bytes
        .open(path)
        .map_err(|e| Error::io(path, e))
}

fn acquire(path: &Path, exclusive: bool, timeout: Duration) -> Result<LockGuard> {
    let file = open_lock_file(path)?;
    let start = Instant::now();
    loop {
        let attempt = if exclusive {
            file.try_lock()
        } else {
            file.try_lock_shared()
        };
        match attempt {
            Ok(()) => return Ok(LockGuard { file }),
            Err(TryLockError::WouldBlock) => {}
            Err(TryLockError::Error(e)) => return Err(Error::io(path, e)),
        }
        if start.elapsed() >= timeout {
            return Err(Error::RegistryBusy {
                path: path.to_path_buf(),
                timeout_ms: timeout.as_millis() as u64,
            });
        }
        std::thread::sleep(RETRY_INTERVAL);
    }
}

/// Acquire a shared (read) lock on `lock_path`.
pub fn shared(lock_path: &Path, timeout: Duration) -> Result<LockGuard> {
    acquire(lock_path, false, timeout)
}

/// Acquire an exclusive (write) lock on `lock_path`.
pub fn exclusive(lock_path: &Path, timeout: Duration) -> Result<LockGuard> {
    acquire(lock_path, true, timeout)
}

/// Write `bytes` to `dest` atomically: write to a sibling temp file, fsync it,
/// then rename over the destination. A crash at any point leaves either the
/// old complete file or the new complete file, never a partial one.
pub fn write_atomic(dest: &Path, bytes: &[u8]) -> Result<()> {
    let parent = dest
        .parent()
        .ok_or_else(|| Error::io(dest, std::io::Error::other("path has no parent")))?;
    std::fs::create_dir_all(parent).map_err(|e| Error::io(parent, e))?;

    let tmp: PathBuf = parent.join(format!(
        "{}.tmp-{}",
        dest.file_name().unwrap_or_default().to_string_lossy(),
        std::process::id()
    ));

    let result = (|| -> Result<()> {
        let mut f = File::create(&tmp).map_err(|e| Error::io(&tmp, e))?;
        f.write_all(bytes).map_err(|e| Error::io(&tmp, e))?;
        f.sync_all().map_err(|e| Error::io(&tmp, e))?;
        drop(f);
        rename_with_retry(&tmp, dest)?;
        sync_dir(parent);
        Ok(())
    })();

    if result.is_err() {
        let _ = std::fs::remove_file(&tmp);
    }
    result
}

/// `std::fs::rename` replaces the destination atomically on both Unix and
/// Windows (MOVEFILE_REPLACE_EXISTING), but on Windows it can transiently
/// fail with a sharing violation while a reader has the destination open.
/// Retry briefly instead of failing the whole write.
fn rename_with_retry(from: &Path, to: &Path) -> Result<()> {
    let start = Instant::now();
    loop {
        match std::fs::rename(from, to) {
            Ok(()) => return Ok(()),
            Err(e) if cfg!(windows) && start.elapsed() < Duration::from_secs(2) => {
                tracing::debug!("rename retry after {e}");
                std::thread::sleep(RETRY_INTERVAL);
            }
            Err(e) => return Err(Error::io(to, e)),
        }
    }
}

/// Best-effort fsync of a directory so the rename itself is durable (Unix
/// only; Windows has no direct equivalent and directory metadata is flushed
/// by the filesystem).
fn sync_dir(dir: &Path) {
    #[cfg(unix)]
    {
        if let Ok(d) = File::open(dir) {
            let _ = d.sync_all();
        }
    }
    #[cfg(not(unix))]
    {
        let _ = dir;
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn exclusive_blocks_exclusive() {
        let tmp = tempfile::tempdir().unwrap();
        let lock_path = tmp.path().join("registry.lock");
        let _g1 = exclusive(&lock_path, DEFAULT_LOCK_TIMEOUT).unwrap();
        let err = exclusive(&lock_path, Duration::from_millis(100)).unwrap_err();
        assert!(matches!(err, Error::RegistryBusy { .. }));
        assert_eq!(err.kind(), crate::ErrorKind::RegistryBusy);
    }

    #[test]
    fn shared_allows_shared() {
        let tmp = tempfile::tempdir().unwrap();
        let lock_path = tmp.path().join("registry.lock");
        let _g1 = shared(&lock_path, DEFAULT_LOCK_TIMEOUT).unwrap();
        let _g2 = shared(&lock_path, Duration::from_millis(100)).unwrap();
    }

    #[test]
    fn lock_released_on_drop() {
        let tmp = tempfile::tempdir().unwrap();
        let lock_path = tmp.path().join("registry.lock");
        drop(exclusive(&lock_path, DEFAULT_LOCK_TIMEOUT).unwrap());
        let _g = exclusive(&lock_path, Duration::from_millis(100)).unwrap();
    }

    #[test]
    fn write_atomic_replaces_content() {
        let tmp = tempfile::tempdir().unwrap();
        let dest = tmp.path().join("registry.json");
        write_atomic(&dest, b"one").unwrap();
        assert_eq!(std::fs::read(&dest).unwrap(), b"one");
        write_atomic(&dest, b"two").unwrap();
        assert_eq!(std::fs::read(&dest).unwrap(), b"two");
        // no temp files left behind
        let leftovers: Vec<_> = std::fs::read_dir(tmp.path())
            .unwrap()
            .filter_map(|e| e.ok())
            .filter(|e| e.file_name().to_string_lossy().contains(".tmp-"))
            .collect();
        assert!(leftovers.is_empty());
    }
}