liminal-rs 0.16.2

use std::fs::{self, File};
use std::path::{Path, PathBuf};
use std::sync::Arc;

use rustix::fs::{FlockOperation, Mode, OFlags, flock, open};
use rustix::io::Errno;

use super::{DurabilityError, EphemeralHaematiteStore};

/// The immutable work done by the root's startup sweep.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct EphemeralSweepCounts {
    /// Dead entries removed before opening a store.
    pub removed_count: u64,
    /// File bytes removed from dead entries.
    pub removed_bytes: u64,
    /// Entries retained because their holder lock was busy.
    pub held_count: u64,
}

/// Owns the root lock until every store opened in it has closed.
#[derive(Debug)]
pub struct EphemeralRoot {
    path: PathBuf,
    lock: File,
    sweep: EphemeralSweepCounts,
}

impl EphemeralRoot {
    /// Locks the throwaway root and removes only entries with no live holder.
    ///
    /// # Errors
    /// The parent is absent, a lock is held, or directory work fails.
    pub fn open(persistence_path: &Path) -> Result<Arc<Self>, DurabilityError> {
        let path = persistence_path.join("ephemeral");
        match fs::create_dir(&path) {
            Ok(()) => {}
            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
                if !fs::symlink_metadata(&path).map_err(root_error)?.is_dir() {
                    return Err(root_error("ephemeral root is not a directory"));
                }
            }
            Err(error) => return Err(root_error(error)),
        }
        let lock = lock_file(&path.join("root.lock"))?;
        flock(&lock, FlockOperation::NonBlockingLockExclusive).map_err(|error| {
            root_error(format!(
                "ephemeral root lock held or unavailable at {}: {error}",
                path.display()
            ))
        })?;
        let sweep = sweep(&path)?;
        tracing::info!(
            removed_count = sweep.removed_count,
            removed_bytes = sweep.removed_bytes,
            held_count = sweep.held_count,
            "ephemeral root startup sweep"
        );
        Ok(Arc::new(Self { path, lock, sweep }))
    }

    /// Opens a store that retains both the root and its own holder lock.
    ///
    /// # Errors
    /// A store or its holder lock cannot be opened.
    pub fn open_store(
        self: &Arc<Self>,
        shard_count: usize,
    ) -> Result<EphemeralHaematiteStore, DurabilityError> {
        super::store::open_ephemeral_owned(Arc::clone(self), shard_count)
    }

    /// The immutable startup sweep counts, retained for the root's life.
    #[must_use]
    pub const fn sweep_counts(&self) -> EphemeralSweepCounts {
        self.sweep
    }

    /// The directory owned by this root lock.
    #[must_use]
    pub fn path(&self) -> &Path {
        &self.path
    }
}

impl Drop for EphemeralRoot {
    fn drop(&mut self) {
        if let Err(error) = flock(&self.lock, FlockOperation::Unlock) {
            tracing::error!(path=%self.path.display(), %error, "ephemeral root unlock failed");
        }
    }
}

pub(super) fn root_error(error: impl std::fmt::Display) -> DurabilityError {
    DurabilityError::EphemeralStoreOpen(format!("ephemeral root: {error}"))
}

pub(super) fn lock_file(path: &Path) -> Result<File, DurabilityError> {
    let fd = open(
        path,
        OFlags::RDWR | OFlags::CREATE | OFlags::CLOEXEC | OFlags::NOFOLLOW,
        Mode::RUSR | Mode::WUSR,
    )
    .map_err(|error| root_error(format!("could not open lock {}: {error}", path.display())))?;
    Ok(File::from(fd))
}

fn sweep(path: &Path) -> Result<EphemeralSweepCounts, DurabilityError> {
    let mut counts = EphemeralSweepCounts::default();
    for entry in fs::read_dir(path).map_err(root_error)? {
        let entry = entry.map_err(root_error)?;
        if !entry
            .file_name()
            .to_string_lossy()
            .starts_with("liminal-durability-")
        {
            continue;
        }
        if !entry.file_type().map_err(root_error)?.is_dir() {
            return Err(root_error(format!(
                "throwaway entry is not a directory: {}",
                entry.path().display()
            )));
        }
        let entry_path = entry.path();
        let holder = lock_file(&entry_path.join("holder.lock"))?;
        match flock(&holder, FlockOperation::NonBlockingLockExclusive) {
            Ok(()) => {
                let bytes = directory_bytes(&entry_path)?;
                fs::remove_dir_all(&entry_path).map_err(root_error)?;
                counts.removed_count = counts
                    .removed_count
                    .checked_add(1)
                    .ok_or_else(|| root_error("removed count overflow"))?;
                counts.removed_bytes = counts
                    .removed_bytes
                    .checked_add(bytes)
                    .ok_or_else(|| root_error("removed bytes overflow"))?;
                tracing::info!(path=%entry_path.display(), bytes, "removed dead ephemeral entry");
            }
            Err(error) if error == Errno::WOULDBLOCK => {
                counts.held_count = counts
                    .held_count
                    .checked_add(1)
                    .ok_or_else(|| root_error("held count overflow"))?;
                tracing::info!(path=%entry_path.display(), "kept live ephemeral entry");
            }
            Err(error) => return Err(root_error(error)),
        }
    }
    Ok(counts)
}

fn directory_bytes(path: &Path) -> Result<u64, DurabilityError> {
    let mut bytes = 0_u64;
    for entry in fs::read_dir(path).map_err(root_error)? {
        let entry = entry.map_err(root_error)?;
        let metadata = fs::symlink_metadata(entry.path()).map_err(root_error)?;
        let size = if metadata.is_dir() {
            directory_bytes(&entry.path())?
        } else {
            metadata.len()
        };
        bytes = bytes
            .checked_add(size)
            .ok_or_else(|| root_error("entry bytes overflow"))?;
    }
    Ok(bytes)
}

#[derive(Debug)]
pub(super) struct EphemeralOwnership {
    pub root: Arc<EphemeralRoot>,
    pub holder: File,
}

impl Drop for EphemeralOwnership {
    fn drop(&mut self) {
        if let Err(error) = flock(&self.holder, FlockOperation::Unlock) {
            tracing::error!(root=%self.root.path().display(), %error, "ephemeral holder unlock failed");
        }
    }
}