blockworx_store/

lock.rs

//! The advisory single-writer lock: a `lock` entry naming the process that
//! holds the container open for writing.
//!
//! Advisory on purpose. It stops the second *blockworx* from writing a log
//! the first is also appending to; it does not — and is not meant to —
//! stop the user's text editor, which is the point of a container made of
//! plain files.

use serde::{Deserialize, Serialize};

use super::container::LOCK;
use super::record::WallTime;
use super::storage::{Storage, ready_now};

/// Who holds the lock, as the entry records it.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct LockHolder {
    pub pid: u32,
    pub since: WallTime,
}

impl std::fmt::Display for LockHolder {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "process {}", self.pid)
    }
}

/// Whether the process named in a lock entry is still running.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Holder {
    Alive,
    Gone,
}

/// Who holds a container open for writing, in the words the reader is told
/// it in.
///
/// Not every lock names a process: a host with a lock manager of its own
/// answers whether a lock is held and nothing more, so what the reader is
/// owed is the *fact*, with a name on it where there is one.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Holding {
    Process(LockHolder),
    Elsewhere,
}

impl std::fmt::Display for Holding {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Holding::Process(held) => write!(f, "{held}"),
            Holding::Elsewhere => f.write_str("another view of this document"),
        }
    }
}

/// The answer to "may I write this container?".
pub enum Claim {
    Taken,
    Held(Holding),
}

/// Claim the writer's lock on `storage`'s container — what
/// [`Storage::claim`] does for a storage with no lock of its own.
///
/// A lock entry whose process is gone is *stale* — a writer that died
/// without releasing it — and is broken rather than left to make the
/// container unopenable forever. So is one that does not parse: it names
/// nobody, and a container cannot be held hostage by a corrupt byte.
///
/// # Errors
/// Failure to write the lock entry.
pub async fn claim<S: Storage + ?Sized>(storage: &S, now: WallTime) -> std::io::Result<Claim> {
    claim_probing(storage, now, holder).await
}

/// The seam the liveness probe enters by: a test can state "the holder is
/// gone" without needing a process that reliably is.
async fn claim_probing<S: Storage + ?Sized>(
    storage: &S,
    now: WallTime,
    probe: impl Fn(u32) -> Holder,
) -> std::io::Result<Claim> {
    if let Ok(Some(bytes)) = storage.read(&LOCK).await {
        let held = serde_json::from_slice::<LockHolder>(&bytes)
            .ok()
            .filter(|held| probe(held.pid) == Holder::Alive);
        if let Some(held) = held {
            return Ok(Claim::Held(Holding::Process(held)));
        }
        tracing::info!(
            lock = storage.names(&LOCK),
            "breaking a stale lock: the process that held it is gone",
        );
    }

    let held = LockHolder {
        pid: std::process::id(),
        since: now,
    };
    let mut line = serde_json::to_vec(&held).map_err(std::io::Error::other)?;
    line.push(b'\n');
    storage.write(&LOCK, &line).await?;
    Ok(Claim::Taken)
}

/// Give the lock back. A release that does not land leaves a stale entry,
/// which the next claim breaks — so there is nothing here for a caller to
/// do about it.
pub fn release<S: Storage + ?Sized>(storage: &S) {
    if let Err(why) = ready_now(storage.remove(&LOCK)) {
        tracing::warn!("the lock on {} was not released: {why}", storage.name());
    }
}

/// Linux answers this from `/proc`, which needs no dependency and no
/// `unsafe`. Everywhere else the honest answer is that we cannot tell, and
/// the safe one is to leave the lock alone: a lock wrongly broken means
/// two writers appending to one log, while a lock wrongly kept means one
/// read-only session and a file the user can delete.
#[cfg(target_os = "linux")]
fn holder(pid: u32) -> Holder {
    if std::path::Path::new(&format!("/proc/{pid}")).exists() {
        Holder::Alive
    } else {
        Holder::Gone
    }
}

#[cfg(not(target_os = "linux"))]
fn holder(_pid: u32) -> Holder {
    Holder::Alive
}

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

    fn read_holder<S: Storage>(storage: &S) -> LockHolder {
        let bytes = ready_now(storage.read(&LOCK))
            .expect("the read answers")
            .expect("the lock entry exists");
        serde_json::from_slice(&bytes).expect("and parses")
    }

    #[test]
    fn a_claim_is_held_until_it_is_released() {
        let storage = Memory::new("doc.bwx");

        match ready_now(claim(&storage, WallTime::from_unix_millis(7)))
            .expect("the lock is written")
        {
            Claim::Taken => {}
            Claim::Held(held) => panic!("a fresh container cannot be held by {held}"),
        }
        assert_eq!(
            read_holder(&storage),
            LockHolder {
                pid: std::process::id(),
                since: WallTime::from_unix_millis(7),
            },
        );

        match ready_now(storage.claim(WallTime::from_unix_millis(8)))
            .expect("the second claim answers")
        {
            Claim::Held(Holding::Process(held)) => assert_eq!(held.pid, std::process::id()),
            Claim::Held(held) => panic!("a lock entry names a process, not {held}"),
            Claim::Taken => panic!("a live holder's lock must not be broken"),
        }

        release(&storage);
        assert_eq!(
            ready_now(storage.read(&LOCK)).expect("the read answers"),
            None,
            "releasing the lock leaves the entry behind",
        );
    }

    /// The stale case, stated without needing a process that is reliably
    /// dead: the holder is gone, so the lock is broken and re-taken.
    #[test]
    fn a_lock_whose_holder_is_gone_is_broken() {
        let storage = Memory::new("doc.bwx");
        let stale = LockHolder {
            pid: 4242,
            since: WallTime::from_unix_millis(1),
        };
        ready_now(storage.write(
            &LOCK,
            &serde_json::to_vec(&stale).expect("a holder serializes"),
        ))
        .expect("the stale lock is planted");
        assert_eq!(
            read_holder(&storage),
            stale,
            "precondition: someone holds it"
        );

        let claim = ready_now(claim_probing(
            &storage,
            WallTime::from_unix_millis(9),
            |_| Holder::Gone,
        ))
        .expect("the claim answers");
        assert!(matches!(claim, Claim::Taken));
        assert_eq!(read_holder(&storage).pid, std::process::id());
    }

    /// A lock entry that names nobody readable is not a lock.
    #[test]
    fn an_unreadable_lock_is_broken() {
        let storage = Memory::new("doc.bwx");
        ready_now(storage.write(&LOCK, b"not json at all")).expect("the garbage is planted");

        let claim =
            ready_now(claim(&storage, WallTime::from_unix_millis(9))).expect("the claim answers");
        assert!(matches!(claim, Claim::Taken));
    }

    /// The production probe, on the one platform that can answer it.
    #[cfg(target_os = "linux")]
    #[test]
    fn the_probe_tells_this_process_from_one_that_cannot_exist() {
        assert_eq!(holder(std::process::id()), Holder::Alive);
        // Above every platform's pid ceiling, so it names no process now
        // and cannot come to name one.
        assert_eq!(holder(i32::MAX as u32), Holder::Gone);
    }
}