haematite-cli 0.21.8

//! Confirmed immutable roots shared with readers independently of writer admission.
use super::{BRANCH_PREFIX, Captured, Failure, SERVICE, branch_name, history, identifier, version};
use crate::serve::protocol::{Selection, Version};
use haematite::branch::{KeptHead, RefRecord as BranchRefRecord};
use haematite::db::{ReadHold, ReadHoldRefused, ReadHolder, ReadOnlyDatabase};
use std::collections::BTreeMap;
use std::num::NonZeroUsize;
use std::ops::{Deref, DerefMut};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex, PoisonError, RwLock, RwLockReadGuard};
use std::time::Duration;

#[path = "../query/run/window_catalogue.rs"]
mod window_clock;

/// A record as published, with the heads it keeps (STOR-003 MD4).
pub type Published = (BranchRefRecord, Vec<KeptHead>);

#[derive(Clone)]
pub struct Catalogue {
    records: Arc<RwLock<BTreeMap<String, Published>>>,
    reader: Arc<ReadOnlyDatabase>,
    /// Holds what a read captured while it reads (STOR-003 H1b): the holder
    /// of the database the storage opened last, so a pass on it reads them.
    holder: ReadHolder,
    /// Moves once after every publish and every removal, so a follower of a
    /// head is woken by the change itself and never asks on a timer.
    changed: Arc<Changed>,
}

/// How many times the catalogue has changed, and the waiters woken by it.
///
/// The count only says that something changed: a waiter reads what it
/// follows again and compares it with what it last saw.
#[derive(Default)]
struct Changed {
    count: Mutex<u64>,
    moved: Condvar,
}

impl Changed {
    /// Count one change and wake every waiter. A count is a plain number no
    /// panic can leave half written, so a poisoned lock is taken as it is.
    fn bump(&self) {
        let mut count = self.count.lock().unwrap_or_else(PoisonError::into_inner);
        *count = count.wrapping_add(1);
        drop(count);
        self.moved.notify_all();
    }

    fn now(&self) -> u64 {
        *self.count.lock().unwrap_or_else(PoisonError::into_inner)
    }

    /// The count once it is no longer `seen`, or once `left` has passed with
    /// it unchanged; with no `left`, only once it moves.
    fn past(&self, seen: u64, left: Option<Duration>, cancelled: Option<&AtomicBool>) -> u64 {
        let unmoved = |count: &mut u64| {
            *count == seen && !cancelled.is_some_and(|closed| closed.load(Ordering::Acquire))
        };
        let held = || self.count.lock().unwrap_or_else(PoisonError::into_inner);
        left.map_or_else(
            || {
                *self
                    .moved
                    .wait_while(held(), unmoved)
                    .unwrap_or_else(PoisonError::into_inner)
            },
            |left| {
                *self
                    .moved
                    .wait_timeout_while(held(), left, unmoved)
                    .unwrap_or_else(PoisonError::into_inner)
                    .0
            },
        )
    }
}

/// A capture whose roots no reclaim pass deletes while this lives: dropped
/// when the answer it was read for is built, on every path.
pub struct Held {
    captured: Captured,
    _hold: ReadHold,
}

impl Deref for Held {
    type Target = Captured;
    fn deref(&self) -> &Captured {
        &self.captured
    }
}

impl DerefMut for Held {
    /// The shards may be reordered; the roots held are the ones captured.
    fn deref_mut(&mut self) -> &mut Captured {
        &mut self.captured
    }
}

impl Held {
    /// The capture, its hold let go: for an answer every read of which is
    /// done.
    #[must_use]
    pub fn into_captured(self) -> Captured {
        self.captured
    }
}

/// What a read does after its hold was refused (STOR-003 H1b).
#[derive(Debug, PartialEq, Eq)]
enum AfterRefusal {
    /// The caller asked for the head and this was the first refusal: the head
    /// moved past an open pass, so capture the head again, once.
    CaptureAgain,
    /// The version read is going: an open pass lists or drains and did not
    /// mark it (RX2: a pass still marking keeps it). A pass that lists or
    /// drains has marked every root any record, hold or lease keeps, so a
    /// root it did not mark is one the store itself has found nothing keeps
    /// (Apollo's ruling, 16:59). Answer `version_released` naming it. A
    /// caller that named a version is never answered from another.
    Released,
    /// The version's root is not on its shard (RX2, 6B): answer
    /// `version_missing`, never `version_released`, which only the store's
    /// own record of a release may say.
    Missing,
    /// Anything else, answered by its own name.
    Refused,
}

/// The rule every read follows: only a caller that asked for the head is
/// retried, and only once; a named version is never swapped for a newer one.
const fn after_refusal(asked_head: bool, retried: bool, refused: &ReadHoldRefused) -> AfterRefusal {
    match refused {
        ReadHoldRefused::NotRetained { .. } if asked_head && !retried => AfterRefusal::CaptureAgain,
        ReadHoldRefused::NotRetained { .. } => AfterRefusal::Released,
        ReadHoldRefused::Missing { .. } => AfterRefusal::Missing,
        _ => AfterRefusal::Refused,
    }
}

fn released(at: &Captured, refused: &ReadHoldRefused) -> Failure {
    Failure::new(
        "version_released",
        format!(
            "version {}.{} of branch {} was released by the store's tidying and cannot be \
             read: {refused}",
            at.version.generation, at.version.sequence, at.branch
        ),
    )
}

impl Catalogue {
    pub(in crate::serve) fn capture_companion_held(
        &self,
        branch: &str,
    ) -> Result<Option<Held>, Failure> {
        identifier(branch)?;
        let name = super::embedded::companion_name(branch);
        let captured = {
            let records = self.read()?;
            records.get(&name).map(|(record, _)| Captured {
                branch: name.clone(),
                version: version(record),
                shards: record.shards.clone(),
            })
        };
        let Some(captured) = captured else {
            return Ok(None);
        };
        let roots: Vec<_> = captured
            .shards
            .iter()
            .map(|shard| (shard.shard_id, shard.head))
            .collect();
        let hold = self.holder.hold(&roots).map_err(|refused| {
            if matches!(refused, ReadHoldRefused::NotRetained { .. }) {
                released(&captured, &refused)
            } else {
                Failure::new("read_hold_refused", refused.to_string())
            }
        })?;
        Ok(Some(Held {
            captured,
            _hold: hold,
        }))
    }

    pub(super) fn new(
        records: BTreeMap<String, Published>,
        reader: Arc<ReadOnlyDatabase>,
        holder: ReadHolder,
    ) -> Self {
        Self {
            records: Arc::new(RwLock::new(records)),
            reader,
            holder,
            changed: Arc::default(),
        }
    }

    /// How many times this catalogue has changed. Read it before the head it
    /// is compared with, then wait past it with [`Self::changed_past`]: a
    /// change made between the two is then never missed.
    #[must_use]
    pub fn changes(&self) -> u64 {
        self.changed.now()
    }

    /// Blocks until the catalogue has changed since `seen`, or until `left`
    /// has passed; with no `left`, until it changes. Answers the count now.
    /// Nothing here wakes on a timer: `left` is the caller's own deadline.
    #[must_use]
    pub fn changed_past(&self, seen: u64, left: Option<Duration>) -> u64 {
        self.changed.past(seen, left, None)
    }

    /// Wake a follower when its peer closes, even when no branch moves.
    pub(in crate::serve) fn wake_followers(&self) {
        // Holding the wait's mutex pairs the close signal with its predicate.
        let count = self
            .changed
            .count
            .lock()
            .unwrap_or_else(PoisonError::into_inner);
        self.changed.moved.notify_all();
        drop(count);
    }

    pub(in crate::serve) fn changed_past_or_closed(
        &self,
        seen: u64,
        left: Option<Duration>,
        cancelled: &AtomicBool,
    ) -> u64 {
        self.changed.past(seen, left, Some(cancelled))
    }

    /// The published version of `branch`, or `None` when no such branch is
    /// published.
    ///
    /// # Errors
    ///
    /// `invalid_identifier`; `catalogue_unavailable`.
    pub fn head(&self, branch: &str) -> Result<Option<Version>, Failure> {
        identifier(branch)?;
        Ok(self
            .read()?
            .get(&branch_name(branch))
            .map(|(record, _kept)| version(record)))
    }

    /// Capture `selection` and hold its roots until the returned [`Held`]
    /// drops: the one way a read captures what it reads (STOR-003 H1b). The
    /// answer names the version read, `Held::version`. A head refused because
    /// an open pass did not mark it is captured again once; a version the
    /// caller named, or a second refusal, is answered `version_released`.
    ///
    /// Costs, on top of the capture: one lock of the holds, then for each
    /// shard one stat and one lock of its domain; one lock when it drops.
    ///
    /// # Errors
    ///
    /// The capture's own; `version_released` naming the version; any other
    /// refusal by its name (`read_hold_database_closed` among them).
    pub fn capture_held(&self, selection: &Selection) -> Result<Held, Failure> {
        let asked_head = selection.version.is_none();
        let mut retried = false;
        loop {
            let captured = self.capture(selection)?;
            let roots: Vec<_> = captured
                .shards
                .iter()
                .map(|shard| (shard.shard_id, shard.head))
                .collect();
            let refused = match self.holder.hold(&roots) {
                Ok(hold) => {
                    #[cfg(test)]
                    after_held::reached();
                    return Ok(Held {
                        captured,
                        _hold: hold,
                    });
                }
                Err(refused) => refused,
            };
            match after_refusal(asked_head, retried, &refused) {
                AfterRefusal::CaptureAgain => retried = true,
                AfterRefusal::Released => return Err(released(&captured, &refused)),
                AfterRefusal::Missing => {
                    return Err(Failure::new(
                        "version_missing",
                        format!(
                            "version {}.{} of branch {} is not in the store: {refused}",
                            captured.version.generation, captured.version.sequence, captured.branch
                        ),
                    ));
                }
                AfterRefusal::Refused => {
                    return Err(Failure::new("read_hold_refused", refused.to_string()));
                }
            }
        }
    }
    fn read(&self) -> Result<RwLockReadGuard<'_, BTreeMap<String, Published>>, Failure> {
        self.records.read().map_err(|_| {
            Failure::new(
                "catalogue_unavailable",
                "catalogue requires process recovery",
            )
        })
    }
    /// Publishes `record` with the heads it keeps, as the ref store holds
    /// them once its install is fenced.
    pub(super) fn publish(
        &self,
        record: BranchRefRecord,
        kept: Vec<KeptHead>,
    ) -> Result<(), Failure> {
        self.records
            .write()
            .map_err(|_| {
                Failure::new(
                    "committed_unpublished",
                    "reference is durable but catalogue requires recovery",
                )
            })?
            .insert(record.name.clone(), (record, kept));
        self.changed.bump();
        Ok(())
    }
    pub(super) fn remove(&self, name: &str) -> Result<(), Failure> {
        self.records
            .write()
            .map_err(|_| {
                Failure::new(
                    "catalogue_unavailable",
                    "catalogue requires process recovery",
                )
            })?
            .remove(name);
        self.changed.bump();
        Ok(())
    }
    /// Captures live roots or seeks a kept row without holding catalogue locks during I/O.
    ///
    /// # Errors
    ///
    /// Refuses missing versions and corrupt rows; propagates storage and catalogue failures.
    pub fn capture(&self, selection: &Selection) -> Result<Captured, Failure> {
        identifier(&selection.branch)?;
        let (current, service) = {
            let visible = self.read()?;
            let current = visible.get(&branch_name(&selection.branch));
            if let Some((record, _kept)) = current
                && selection
                    .version
                    .is_none_or(|requested| requested == version(record))
            {
                return Ok(Captured {
                    branch: selection.branch.clone(),
                    version: version(record),
                    shards: record.shards.clone(),
                });
            }
            if current.is_none() && selection.version.is_none() {
                return Err(Failure::new("branch_missing", &selection.branch));
            }
            let service = visible
                .get(SERVICE)
                .map(|(record, _kept)| record.clone())
                .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
            let current = current.map(|(record, kept)| {
                let head = Captured {
                    branch: selection.branch.clone(),
                    version: version(record),
                    shards: record.shards.clone(),
                };
                (head, kept.clone())
            });
            drop(visible);
            (current, service)
        };
        let requested = selection.version.ok_or_else(|| {
            Failure::new(
                "version_missing",
                "kept capture requires an explicit version",
            )
        })?;
        // A version at or below the branch's mark was released (STOR-003
        // HR2): that answers first, whatever else could.
        let generation = requested.generation;
        if let Some(released) =
            history::read_released(&self.reader, &service, &selection.branch, generation)?
            && released.covers(requested)
        {
            return Err(released.refusal(&selection.branch, requested));
        }
        match &current {
            Some((head, _kept)) if requested.generation > head.version.generation => {
                return Err(Failure::new(
                    "generation_mismatch",
                    "branch generation differs",
                ));
            }
            // STOR-003 MD4: the current generation's record keeps the newest
            // versions below its head; they are read before any row.
            Some((head, kept)) if requested.generation == head.version.generation => {
                if let Some(captured) = history::capture_from_kept(head, kept, requested) {
                    return Ok(captured);
                }
            }
            _ => {}
        }
        if let Some(captured) =
            history::capture_kept(&self.reader, &service, &selection.branch, requested)?
        {
            return Ok(captured);
        }
        // An earlier generation, or one whose name was discarded, has ended:
        // with no row it is refused by that name, never as missing.
        let ended = current
            .as_ref()
            .is_none_or(|(head, _kept)| requested.generation < head.version.generation);
        if ended {
            return Err(history::generation_ended(&selection.branch, requested));
        }
        Err(Failure::new("version_missing", "version is not retained"))
    }

    /// # Errors
    ///
    /// Propagates any error from `read`.
    pub fn branches(&self, after: Option<&str>, limit: usize) -> Result<Vec<Captured>, Failure> {
        Ok(self
            .read()?
            .values()
            .filter_map(|(record, _kept)| {
                record
                    .name
                    .strip_prefix(BRANCH_PREFIX)
                    .map(|name| Captured {
                        branch: name.to_owned(),
                        version: version(record),
                        shards: record.shards.clone(),
                    })
            })
            .filter(|row| after.is_none_or(|after| row.branch.as_str() > after))
            .take(limit)
            .collect())
    }

    /// # Errors
    ///
    /// Refuses corrupt rows and cursors; propagates storage and catalogue failures.
    pub fn history(
        &self,
        branch: &str,
        after: Option<Version>,
        limit: usize,
    ) -> Result<Vec<Captured>, Failure> {
        identifier(branch)?;
        let Some(limit) = NonZeroUsize::new(limit) else {
            return Ok(Vec::new());
        };
        let (current, kept, service) = {
            let visible = self.read()?;
            let (record, kept) = visible
                .get(&branch_name(branch))
                .ok_or_else(|| Failure::new("branch_missing", branch))?;
            let current = Captured {
                branch: branch.to_owned(),
                version: version(record),
                shards: record.shards.clone(),
            };
            let kept = kept.clone();
            let service = visible
                .get(SERVICE)
                .map(|(record, _kept)| record.clone())
                .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
            drop(visible);
            (current, kept, service)
        };
        if let Some(after) = after
            && after.generation < current.version.generation
        {
            return Err(history::generation_ended(branch, after));
        }
        if after.is_some_and(|value| value.generation != current.version.generation) {
            return Err(Failure::new(
                "generation_mismatch",
                "history cursor belongs to another generation",
            ));
        }
        history::list_kept(&self.reader, &service, &current, &kept, after, limit)
    }
}

#[cfg(test)]
#[path = "catalogue_after_held.rs"]
pub mod after_held;

#[cfg(test)]
#[path = "catalogue_tests.rs"]
mod tests;