velesdb-memory 0.15.0

use super::STATE_FILE;
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};

/// The file that marks a migration in progress.
pub const LOCK_FILE: &str = "migration.lock";

/// The persistent sibling whose OS lock serializes every canonical lock check.
///
/// Unlike [`LOCK_FILE`], this file is never removed. Its inode must stay stable:
/// the advisory lock on its open handle closes the delete/recreate ABA window
/// around the human-readable canonical record.
pub(in crate::migration) const LOCK_GUARD_FILE: &str = "migration.lock.guard";

const LOCK_FORMAT_VERSION: u32 = 1;
/// A genuine record is a few hundred bytes; anything past this is not one.
const MAX_LOCK_RECORD_BYTES: u64 = 4096;
/// Prefix of the legacy single-line record `holder` still echoes back.
const LEGACY_HOLDER_PREFIX: &str = "held_by=";
/// Longest legacy holder line `holder` echoes back verbatim.
const MAX_LEGACY_HOLDER_BYTES: usize = 256;
/// What `holder` reports for a record it will not echo.
const UNRECOGNISED_HOLDER: &str = "unrecognised lock record, content not shown";
static LOCK_SEQUENCE: AtomicU64 = AtomicU64::new(0);

/// Exclusive possession of a migration workspace.
///
/// The persistent OS guard is held before the canonical record is inspected
/// and remains held until explicit release or drop. A canonical record is
/// deliberately retained after drop or panic as fail-closed evidence.
#[derive(Debug)]
pub struct MigrationLock {
    path: PathBuf,
    token: String,
    guard: std::fs::File,
}

#[derive(Debug, serde::Serialize, serde::Deserialize)]
struct LockRecord {
    format_version: u32,
    held_by: String,
    token: String,
}

impl MigrationLock {
    /// Take the lock in `workspace` on behalf of `holder`.
    ///
    /// # Errors
    /// The OS guard is held, a canonical record remains, or the workspace is
    /// unwritable. Neither an active nor a dead lock is stolen automatically.
    pub fn acquire(workspace: &Path, holder: &str) -> Result<Self, String> {
        let path = workspace.join(LOCK_FILE);
        let guard = open_and_lock_guard(workspace)?;
        ensure_lock_record_absent(workspace, &path)?;
        let token = create_lock_record(&path, holder)?;
        Ok(Self { path, token, guard })
    }

    /// Who holds the lock in `workspace`, as recorded, or `None` when free.
    ///
    /// Also `None` when the record cannot be read through the bounded,
    /// symlink-refusing handle. A record that reads but is neither a lock
    /// record nor a single short legacy `held_by=` line is reported as
    /// unrecognised, its content withheld: `migration.lock` may be a hard
    /// link to any readable file, and this string lands in error messages.
    #[must_use]
    pub fn holder(workspace: &Path) -> Option<String> {
        read_lock_body(&workspace.join(LOCK_FILE)).map(|body| {
            serde_json::from_str::<LockRecord>(&body).map_or_else(
                |_| legacy_holder_line(&body).unwrap_or_else(|| UNRECOGNISED_HOLDER.to_owned()),
                |record| format!("{LEGACY_HOLDER_PREFIX}{}", record.held_by),
            )
        })
    }

    pub(super) fn verify_workspace(&self, workspace: &Path) -> Result<(), String> {
        let expected = workspace.join(LOCK_FILE);
        if self.path != expected || !self.owns_current_lock() {
            return Err(format!(
                "cannot write {STATE_FILE} without the exact live migration lock identity for {}; acquire MigrationLock for this exact workspace first",
                workspace.display()
            ));
        }
        Ok(())
    }

    fn owns_current_lock(&self) -> bool {
        read_lock_body(&self.path)
            .and_then(|body| serde_json::from_str::<LockRecord>(&body).ok())
            .is_some_and(|record| {
                record.format_version == LOCK_FORMAT_VERSION && record.token == self.token
            })
    }

    fn remove_if_owned(&self) -> Result<(), String> {
        if !self.owns_current_lock() {
            return Err(format!(
                "cannot release {LOCK_FILE}: the lock at {} is absent, invalid, or belongs to a later acquisition",
                self.path.display()
            ));
        }
        std::fs::remove_file(&self.path).map_err(|err| format!("cannot release {LOCK_FILE}: {err}"))
    }

    /// Release the lock.
    ///
    /// # Errors
    /// The canonical lock identity changed, the lock file cannot be removed,
    /// or the OS guard cannot be unlocked.
    pub fn release(self) -> Result<(), String> {
        self.remove_if_owned()?;
        fs2::FileExt::unlock(&self.guard).map_err(|err| {
            format!("removed {LOCK_FILE} but cannot unlock {LOCK_GUARD_FILE}: {err}")
        })
    }
}

/// Read the lock record through a handle that refuses a symlink (#2404, #2424).
///
/// The record is created once via `create_new` and afterwards only read or
/// removed, never reopened to write. A hard link planted before creation makes
/// `create_new` fail (`EEXIST`), so a later alias (a hard-link backup, `cp -al`)
/// has nothing to intercept and `nlink > 1` would only refuse an untampered
/// record (#2409 round 8); hence `_allow_hard_links`. The read is bounded so a
/// swapped-in huge file cannot exhaust memory, and a body past the bound is
/// refused whole rather than parsed truncated (possibly mid UTF-8 character).
fn read_lock_body(path: &Path) -> Option<String> {
    let mut options = std::fs::OpenOptions::new();
    options.read(true);
    let (file, _metadata) = crate::mutation::atomic_file::open_regular_file_allow_hard_links(
        path,
        "migration lock",
        options,
    )
    .ok()?;
    let mut body = String::new();
    file.take(MAX_LOCK_RECORD_BYTES + 1)
        .read_to_string(&mut body)
        .ok()?;
    (body.len() as u64 <= MAX_LOCK_RECORD_BYTES).then_some(body)
}

/// The body as a legacy `held_by=<name>` record: one line, short, no control
/// characters. Anything else is not echoed back (#2424).
fn legacy_holder_line(body: &str) -> Option<String> {
    let line = body.trim();
    let name = line.strip_prefix(LEGACY_HOLDER_PREFIX)?;
    let printable = !name.is_empty() && !line.chars().any(char::is_control);
    (printable && line.len() <= MAX_LEGACY_HOLDER_BYTES).then(|| line.to_owned())
}

fn open_and_lock_guard(workspace: &Path) -> Result<std::fs::File, String> {
    let guard_path = workspace.join(LOCK_GUARD_FILE);
    let guard = std::fs::OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(&guard_path)
        .map_err(|err| format!("cannot open persistent {LOCK_GUARD_FILE}: {err}"))?;
    validate_guard_file(&guard_path, &guard)?;
    lock_guard(workspace, &guard)?;
    Ok(guard)
}

fn validate_guard_file(path: &Path, guard: &std::fs::File) -> Result<(), String> {
    let path_metadata = std::fs::symlink_metadata(path)
        .map_err(|err| format!("cannot inspect {LOCK_GUARD_FILE}: {err}"))?;
    let handle_is_file = guard.metadata().is_ok_and(|metadata| metadata.is_file());
    if !path_metadata.file_type().is_symlink() && handle_is_file {
        return Ok(());
    }
    Err(format!(
        "refusing {LOCK_GUARD_FILE} at {}: the persistent guard must be a regular, non-symlink file",
        path.display()
    ))
}

fn lock_guard(workspace: &Path, guard: &std::fs::File) -> Result<(), String> {
    match fs2::FileExt::try_lock_exclusive(guard) {
        Ok(()) => Ok(()),
        Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => Err(format!(
            "a live migration still holds this workspace guard ({}). The OS guard is NOT stolen and deleting {LOCK_FILE} cannot release it; wait for the owner or stop it explicitly.",
            MigrationLock::holder(workspace).unwrap_or_else(|| "holder record missing".to_owned()),
        )),
        Err(err) => Err(format!("cannot lock {LOCK_GUARD_FILE}: {err}")),
    }
}

fn ensure_lock_record_absent(workspace: &Path, path: &Path) -> Result<(), String> {
    match std::fs::symlink_metadata(path) {
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(err) => Err(format!("cannot inspect {LOCK_FILE}: {err}")),
        Ok(_) => Err(format!(
            "a migration lock record remains in this workspace ({}). It is NOT stolen automatically: a dead process releases the OS guard but leaves this evidence behind. If you are certain no migration is running, delete {} yourself.",
            MigrationLock::holder(workspace).unwrap_or_else(|| "holder unknown".to_owned()),
            path.display()
        )),
    }
}

fn create_lock_record(path: &Path, holder: &str) -> Result<String, String> {
    let token = next_lock_token();
    let record = LockRecord {
        format_version: LOCK_FORMAT_VERSION,
        held_by: holder.to_owned(),
        token: token.clone(),
    };
    let body = serde_json::to_vec_pretty(&record)
        .map_err(|err| format!("cannot serialise {LOCK_FILE}: {err}"))?;
    // `read_lock_body` refuses a record past this bound, so one written
    // anyway could never be released: refuse it before anything exists.
    if body.len() as u64 > MAX_LOCK_RECORD_BYTES {
        return Err(format!(
            "refusing {LOCK_FILE}: holder name too long ({} byte record, limit {MAX_LOCK_RECORD_BYTES})",
            body.len()
        ));
    }
    let mut file = std::fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(path)
        .map_err(|err| format!("cannot create {LOCK_FILE}: {err}"))?;
    file.write_all(&body)
        .map_err(|err| format!("cannot write {LOCK_FILE}: {err}"))?;
    file.flush()
        .and_then(|()| file.sync_all())
        .map_err(|err| format!("cannot persist {LOCK_FILE}: {err}"))?;
    Ok(token)
}

fn next_lock_token() -> String {
    let sequence = LOCK_SEQUENCE.fetch_add(1, Ordering::Relaxed);
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |elapsed| elapsed.as_nanos());
    format!(
        "lock-v{LOCK_FORMAT_VERSION}-{:08x}-{nanos:032x}-{sequence:016x}",
        std::process::id()
    )
}

#[cfg(all(test, unix))]
#[path = "lock_tests.rs"]
mod lock_tests;