durability/

storage.rs

//! Storage abstraction for durability.
//!
//! Vocabulary note:
//! - Some durability primitives require **atomicity** (e.g. atomic rename/write) and
//!   **integrity** (checksums, framing).
//! - Stable-storage **durability** (survives power loss after reporting success)
//!   additionally requires explicit `fsync`/`sync_all` barriers and sometimes
//!   parent-directory sync after renames.

use crate::error::{PersistenceError, PersistenceResult};
use std::io::{Read, Write};
use std::path::PathBuf;

/// Make the **data** of `path` durable (`fdatasync`).
///
/// Uses `sync_data()` (fdatasync) rather than `sync_all()` (fsync). For
/// append-only logs, this is correct and faster: fdatasync skips unnecessary
/// metadata updates (mtime, atime) that don't affect crash recovery.
///
/// Requires `Directory::file_path()`. Returns `NotSupported` if unavailable.
pub fn sync_file<D: Directory + ?Sized>(dir: &D, path: &str) -> PersistenceResult<()> {
    let Some(p) = dir.file_path(path) else {
        return Err(PersistenceError::NotSupported(
            "sync_file requires Directory::file_path()".into(),
        ));
    };
    let f = std::fs::OpenOptions::new().read(true).open(&p)?;
    f.sync_data()?;
    Ok(())
}

/// Attempt to `fsync`/`sync_all` the parent directory of `path`.
///
/// This is the commonly-missed step needed to make *names* durable:
/// - durable file creation
/// - durable atomic rename
///
/// Notes:
/// - On some platforms/filesystems, syncing the directory is required for the rename/create to
///   survive power loss even after syncing the file itself.
/// - This requires `Directory::file_path()`. If unavailable, returns `NotSupported`.
pub fn sync_parent_dir<D: Directory + ?Sized>(dir: &D, path: &str) -> PersistenceResult<()> {
    let Some(p) = dir.file_path(path) else {
        return Err(PersistenceError::NotSupported(
            "sync_parent_dir requires Directory::file_path()".into(),
        ));
    };
    let Some(parent) = p.parent() else {
        return Err(PersistenceError::InvalidConfig(format!(
            "path has no parent directory: {p:?}"
        )));
    };
    let f = std::fs::File::open(parent)?;
    // Best-effort: if this fails due to platform-specific directory open semantics,
    // surface the error to the caller; stable storage requires an explicit decision.
    f.sync_all()?;
    Ok(())
}

/// Policy for when writers call `Write::flush()`.
///
/// Vocabulary note:
/// - `flush()` is not a stable-storage durability guarantee on most filesystems; it is best
///   treated as an IO boundary (push to OS / underlying writer).
/// - Stable-storage durability requires explicit `sync_all`/`fsync` barriers, which are not
///   expressible via the `Directory` trait today.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum FlushPolicy {
    /// Call `flush()` after each logical append operation.
    PerAppend,
    /// Call `flush()` every N logical append operations.
    ///
    /// `n=1` is equivalent to `PerAppend`. `n=0` is treated as `PerAppend`.
    EveryN(usize),
    /// Call `flush()` when the specified duration has elapsed since the last flush.
    ///
    /// Checked lazily on each `append()` call. If no appends arrive, no flush occurs.
    /// For background flushing independent of write activity, use an external timer
    /// calling [`WalWriter::flush`](crate::walog::WalWriter::flush) directly.
    Interval(std::time::Duration),
    /// Do not call `flush()` implicitly; callers may flush explicitly (if supported by the backend).
    Manual,
}

/// Trait for directory-like storage backends.
pub trait Directory: Send + Sync {
    /// Create a new file for writing (overwriting if it exists).
    fn create_file(&self, path: &str) -> PersistenceResult<Box<dyn Write + Send>>;
    /// Open an existing file for reading.
    fn open_file(&self, path: &str) -> PersistenceResult<Box<dyn Read + Send>>;
    /// Return whether a path exists.
    fn exists(&self, path: &str) -> bool;
    /// Delete a file or directory (directories recursively).
    fn delete(&self, path: &str) -> PersistenceResult<()>;
    /// Atomically rename/move a file.
    fn atomic_rename(&self, from: &str, to: &str) -> PersistenceResult<()>;
    /// Create a directory (and parents if needed).
    fn create_dir_all(&self, path: &str) -> PersistenceResult<()>;
    /// List entries in a directory.
    fn list_dir(&self, path: &str) -> PersistenceResult<Vec<String>>;
    /// Open a file for appending (creating it if missing).
    fn append_file(&self, path: &str) -> PersistenceResult<Box<dyn Write + Send>>;
    /// Atomically write bytes to a path.
    fn atomic_write(&self, path: &str, data: &[u8]) -> PersistenceResult<()>;
    /// Optional filesystem path for backends that support it.
    fn file_path(&self, path: &str) -> Option<PathBuf>;

    // -- Stable-storage durability helpers (default: delegate to free functions) --
    //
    // These require `file_path()` to return `Some`. Non-filesystem backends
    // get `NotSupported` from the defaults, which is correct -- stable-storage
    // durability is meaningless without a real filesystem.

    /// Attempt to make the file at `path` durable on stable storage.
    ///
    /// Default: delegates to [`sync_file`].
    /// Returns `NotSupported` if `file_path()` returns `None`.
    fn durable_sync_file(&self, path: &str) -> PersistenceResult<()> {
        sync_file(self, path)
    }

    /// Attempt to make the *name* of `path` durable (sync the parent directory).
    ///
    /// Default: delegates to [`sync_parent_dir`].
    /// Returns `NotSupported` if `file_path()` returns `None`.
    fn durable_sync_parent_dir(&self, path: &str) -> PersistenceResult<()> {
        sync_parent_dir(self, path)
    }

    /// Atomically rename and then sync the destination parent directory.
    ///
    /// Returns `NotSupported` if `file_path()` returns `None`.
    fn atomic_rename_durable(&self, from: &str, to: &str) -> PersistenceResult<()> {
        let from_path = match self.file_path(from) {
            Some(p) => p,
            None => {
                return Err(PersistenceError::NotSupported(
                    "atomic_rename_durable requires Directory::file_path()".into(),
                ));
            }
        };
        let to_path = match self.file_path(to) {
            Some(p) => p,
            None => {
                return Err(PersistenceError::NotSupported(
                    "atomic_rename_durable requires Directory::file_path()".into(),
                ));
            }
        };

        self.atomic_rename(from, to)?;
        let from_parent = from_path.parent();
        let to_parent = to_path.parent();
        if from_parent != to_parent {
            self.durable_sync_parent_dir(from)?;
        }
        self.durable_sync_parent_dir(to)?;
        Ok(())
    }

    /// Atomically write bytes to `path` with explicit durability barriers.
    ///
    /// Stronger than [`Directory::atomic_write`]: writes temp, syncs temp,
    /// renames, syncs parent directory.
    ///
    /// Returns `NotSupported` if `file_path()` returns `None`.
    fn atomic_write_durable(&self, path: &str, data: &[u8]) -> PersistenceResult<()> {
        if self.file_path(path).is_none() {
            return Err(PersistenceError::NotSupported(
                "atomic_write_durable requires Directory::file_path()".into(),
            ));
        }

        let tmp = format!("{path}.tmp");
        if let Err(e) = (|| -> PersistenceResult<()> {
            let mut w = self.create_file(&tmp)?;
            w.write_all(data)?;
            w.flush()?;
            Ok(())
        })() {
            let _ = self.delete(&tmp);
            return Err(e);
        }

        if let Err(e) = self.durable_sync_file(&tmp) {
            let _ = self.delete(&tmp);
            return Err(e);
        }

        if let Err(e) = self.atomic_rename_durable(&tmp, path) {
            let _ = self.delete(&tmp);
            return Err(e);
        }

        Ok(())
    }
}

/// Filesystem-backed `Directory` rooted at a local path.
pub struct FsDirectory {
    root: PathBuf,
}

impl FsDirectory {
    /// Create (or open) a filesystem directory backend rooted at `root`.
    pub fn new(root: impl Into<PathBuf>) -> PersistenceResult<Self> {
        let root = root.into();
        std::fs::create_dir_all(&root)?;
        Ok(Self { root })
    }

    /// Create a filesystem directory backend wrapped in `Arc<dyn Directory>`.
    pub fn arc(
        root: impl Into<std::path::PathBuf>,
    ) -> PersistenceResult<std::sync::Arc<dyn Directory>> {
        Ok(std::sync::Arc::new(Self::new(root)?))
    }

    fn resolve_path(&self, path: &str) -> PersistenceResult<PathBuf> {
        // Reject path traversal: `..`, absolute paths, and prefix components.
        for component in std::path::Path::new(path).components() {
            match component {
                std::path::Component::ParentDir
                | std::path::Component::RootDir
                | std::path::Component::Prefix(_) => {
                    return Err(PersistenceError::InvalidConfig(format!(
                        "path must not contain '..', absolute, or prefix components: {path}"
                    )));
                }
                _ => {}
            }
        }
        Ok(self.root.join(path))
    }
}

impl Directory for FsDirectory {
    fn create_file(&self, path: &str) -> PersistenceResult<Box<dyn Write + Send>> {
        let full_path = self.resolve_path(path)?;
        if let Some(parent) = full_path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        Ok(Box::new(std::fs::File::create(full_path)?))
    }

    fn open_file(&self, path: &str) -> PersistenceResult<Box<dyn Read + Send>> {
        let full_path = self.resolve_path(path)?;
        if !full_path.exists() {
            return Err(PersistenceError::NotFound(full_path.display().to_string()));
        }
        Ok(Box::new(std::fs::File::open(full_path)?))
    }

    fn exists(&self, path: &str) -> bool {
        self.resolve_path(path).map(|p| p.exists()).unwrap_or(false)
    }

    fn delete(&self, path: &str) -> PersistenceResult<()> {
        let full_path = self.resolve_path(path)?;
        if full_path.is_dir() {
            std::fs::remove_dir_all(full_path)?;
        } else if full_path.exists() {
            std::fs::remove_file(full_path)?;
        }
        Ok(())
    }

    fn atomic_rename(&self, from: &str, to: &str) -> PersistenceResult<()> {
        let from_path = self.resolve_path(from)?;
        let to_path = self.resolve_path(to)?;
        if let Some(parent) = to_path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        std::fs::rename(from_path, to_path)?;
        Ok(())
    }

    fn create_dir_all(&self, path: &str) -> PersistenceResult<()> {
        std::fs::create_dir_all(self.resolve_path(path)?)?;
        Ok(())
    }

    fn list_dir(&self, path: &str) -> PersistenceResult<Vec<String>> {
        let full_path = self.resolve_path(path)?;
        if !full_path.exists() {
            return Ok(Vec::new());
        }
        let entries = std::fs::read_dir(full_path)?;
        let mut out = Vec::new();
        for entry in entries {
            let entry = entry?;
            out.push(entry.file_name().to_string_lossy().to_string());
        }
        out.sort();
        Ok(out)
    }

    fn append_file(&self, path: &str) -> PersistenceResult<Box<dyn Write + Send>> {
        let full_path = self.resolve_path(path)?;
        if let Some(parent) = full_path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        let file = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(full_path)?;
        Ok(Box::new(file))
    }

    fn atomic_write(&self, path: &str, data: &[u8]) -> PersistenceResult<()> {
        let temp_path = format!("{path}.tmp");
        let full_temp_path = self.resolve_path(&temp_path)?;
        if let Some(parent) = full_temp_path.parent() {
            std::fs::create_dir_all(parent)?;
        }

        if let Err(e) = (|| -> PersistenceResult<()> {
            let mut temp_file = std::fs::File::create(&full_temp_path)?;
            temp_file.write_all(data)?;
            temp_file.sync_all()?;
            Ok(())
        })() {
            let _ = std::fs::remove_file(&full_temp_path);
            return Err(e);
        }

        let full_path = self.resolve_path(path)?;
        if let Err(e) = std::fs::rename(&full_temp_path, &full_path) {
            let _ = std::fs::remove_file(&full_temp_path);
            return Err(e.into());
        }

        if let Some(parent) = full_path.parent() {
            let parent_file = std::fs::File::open(parent)?;
            parent_file.sync_all()?;
        }
        Ok(())
    }

    fn file_path(&self, path: &str) -> Option<PathBuf> {
        self.resolve_path(path).ok()
    }
}

/// In-memory `Directory` used for tests.
#[derive(Clone, Default)]
pub struct MemoryDirectory {
    files: std::sync::Arc<std::sync::RwLock<std::collections::HashMap<String, Vec<u8>>>>,
}

impl MemoryDirectory {
    /// Create an empty in-memory directory.
    pub fn new() -> Self {
        Self::default()
    }

    /// Create an empty in-memory directory wrapped in `Arc<dyn Directory>`.
    pub fn arc() -> std::sync::Arc<dyn Directory> {
        std::sync::Arc::new(Self::new())
    }
}

impl Directory for MemoryDirectory {
    fn create_file(&self, path: &str) -> PersistenceResult<Box<dyn Write + Send>> {
        // Overwrite semantics: clear the file eagerly, then append in-place.
        self.files
            .write()
            .map_err(|_| PersistenceError::LockFailed {
                resource: "memory directory".to_string(),
                reason: "lock poisoned".to_string(),
            })?
            .insert(path.to_string(), Vec::new());

        Ok(Box::new(MemoryInPlaceWriter {
            files: self.files.clone(),
            path: path.to_string(),
        }))
    }

    fn open_file(&self, path: &str) -> PersistenceResult<Box<dyn Read + Send>> {
        let files = self
            .files
            .read()
            .map_err(|_| PersistenceError::LockFailed {
                resource: "memory directory".to_string(),
                reason: "lock poisoned".to_string(),
            })?;
        let data = files
            .get(path)
            .ok_or_else(|| PersistenceError::NotFound(path.to_string()))?
            .clone();
        Ok(Box::new(std::io::Cursor::new(data)))
    }

    fn exists(&self, path: &str) -> bool {
        self.files
            .read()
            .map(|f| f.contains_key(path))
            .unwrap_or(false)
    }

    fn delete(&self, path: &str) -> PersistenceResult<()> {
        let mut files = self
            .files
            .write()
            .map_err(|_| PersistenceError::LockFailed {
                resource: "memory directory".to_string(),
                reason: "lock poisoned".to_string(),
            })?;
        files.remove(path);
        // Also remove children (simulate remove_dir_all).
        let prefix = format!("{path}/");
        files.retain(|k, _| !k.starts_with(&prefix));
        Ok(())
    }

    fn atomic_rename(&self, from: &str, to: &str) -> PersistenceResult<()> {
        let mut files = self
            .files
            .write()
            .map_err(|_| PersistenceError::LockFailed {
                resource: "memory directory".to_string(),
                reason: "lock poisoned".to_string(),
            })?;
        let data = files
            .remove(from)
            .ok_or_else(|| PersistenceError::NotFound(from.to_string()))?;
        files.insert(to.to_string(), data);
        Ok(())
    }

    fn create_dir_all(&self, _path: &str) -> PersistenceResult<()> {
        Ok(())
    }

    fn list_dir(&self, path: &str) -> PersistenceResult<Vec<String>> {
        let files = self
            .files
            .read()
            .map_err(|_| PersistenceError::LockFailed {
                resource: "memory directory".to_string(),
                reason: "lock poisoned".to_string(),
            })?;
        let prefix = if path.is_empty() {
            "".to_string()
        } else {
            format!("{path}/")
        };
        let result: std::collections::BTreeSet<String> = files
            .keys()
            .filter(|k| k.starts_with(&prefix))
            .filter_map(|k| {
                let rest = k.strip_prefix(&prefix).unwrap_or(k);
                let first_component = rest.split('/').next().unwrap_or(rest);
                if first_component.is_empty() {
                    None
                } else {
                    Some(first_component.to_string())
                }
            })
            .collect();
        Ok(result.into_iter().collect())
    }

    fn append_file(&self, path: &str) -> PersistenceResult<Box<dyn Write + Send>> {
        // Ensure the file exists, then append in-place.
        {
            let mut files = self
                .files
                .write()
                .map_err(|_| PersistenceError::LockFailed {
                    resource: "memory directory".to_string(),
                    reason: "lock poisoned".to_string(),
                })?;
            files.entry(path.to_string()).or_insert_with(Vec::new);
        }
        Ok(Box::new(MemoryInPlaceWriter {
            files: self.files.clone(),
            path: path.to_string(),
        }))
    }

    fn atomic_write(&self, path: &str, data: &[u8]) -> PersistenceResult<()> {
        let mut files = self
            .files
            .write()
            .map_err(|_| PersistenceError::LockFailed {
                resource: "memory directory".to_string(),
                reason: "lock poisoned".to_string(),
            })?;
        files.insert(path.to_string(), data.to_vec());
        Ok(())
    }

    fn file_path(&self, _path: &str) -> Option<PathBuf> {
        None
    }
}

struct MemoryInPlaceWriter {
    files: std::sync::Arc<std::sync::RwLock<std::collections::HashMap<String, Vec<u8>>>>,
    path: String,
}

impl Write for MemoryInPlaceWriter {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        let mut files = self
            .files
            .write()
            .map_err(|_| std::io::Error::other("lock poisoned"))?;
        let entry = files.entry(self.path.clone()).or_insert_with(Vec::new);
        entry.extend_from_slice(buf);
        Ok(buf.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}