haematite-cli 0.21.3

//! STOR-003 HR4: the release driver.
//!
//! One thread for the service releases a branch's oldest history once its
//! advances since its last release reach the batch. It is woken by the write
//! that makes a branch due, never by a clock, and holds the storage lock for
//! one release at a time.
//!
//! What waits is a set of branch names, so a burst of writes holds one entry
//! a branch, bounded by the branches.
//!
//! A decision a crash left is the commit point: start rolls each forward as
//! written, on the caller's thread under the storage lock, before the thread
//! is spawned and so before the service admits a request, and fails by name
//! if one cannot be. The cost is the decided branches times three durable
//! operations (Apollo 14:24, HR4-fix2).
//!
//! The count of advances lives in memory and starts at none at open, so a
//! service restarted faster than the batch would never release. With both
//! keep settings set, every live branch is put in the set once at start:
//! the cost is the branches times at most `release_max_rows` + 1 rows read,
//! in the thread, one storage lock hold a branch (STOR-003 HR4-fix).
//!
//! With either keep setting unset nothing can ever be taken, so no branch is
//! ever due: the thread takes no lock and reads no rows for it, and the
//! status says once, by name, which setting is unset.
//!
//! A release that fails is not asked again until another batch of advances:
//! the count starts again whatever the outcome, and the failure is kept,
//! with its branch and time, until a later one replaces it.
//! Shutdown tells the thread and joins it: a release under way ends its
//! three operations, and no new one starts. A thread that ended any other
//! way is a named state in the status, never silence.

use std::collections::{BTreeMap, BTreeSet};
use std::num::{NonZeroU64, NonZeroUsize};
use std::sync::{Arc, Condvar, Mutex, MutexGuard, PoisonError};
use std::thread::JoinHandle;
use std::time::Duration;

use haematite::SyncCounts;
use haematite::branch::{Timestamp, current_timestamp};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};

#[cfg(test)]
use super::storage::{Crash, crash_at};
use super::storage::{
    Failure, IntentListing, Keep, ReleaseAnswer, ReleaseRules, ReleaseSource, Storage,
};

/// The most one release may name: past it, one release is a long pause.
const MAX_ROWS_MOST: u64 = 65_536;

/// The most days `keep.days` may name: a hundred years.
const DAYS_MOST: u64 = 36_500;

/// `retention` in the service's configuration.
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Config {
    #[serde(default = "release_max_rows")]
    pub release_max_rows: u64,
    #[serde(default = "release_batch")]
    pub release_batch: u64,
    #[serde(default)]
    pub keep: KeepConfig,
}

impl Default for Config {
    fn default() -> Self {
        Self {
            release_max_rows: release_max_rows(),
            release_batch: release_batch(),
            keep: KeepConfig::default(),
        }
    }
}

/// `retention.keep`: each left out keeps every version.
#[derive(Debug, Default, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct KeepConfig {
    pub versions: Option<u64>,
    pub days: Option<u64>,
}

const fn release_max_rows() -> u64 {
    1_024
}

const fn release_batch() -> u64 {
    256
}

/// The driver's settings, checked.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Settings {
    pub max_rows: NonZeroUsize,
    pub batch: NonZeroUsize,
    pub keep: Keep,
}

impl Config {
    /// # Errors
    /// Each setting refused by its name: zero where zero means nothing, or
    /// past its most.
    pub fn settings(&self) -> Result<Settings, String> {
        let max_rows = Some(self.release_max_rows)
            .filter(|rows| *rows <= MAX_ROWS_MOST)
            .and_then(|rows| usize::try_from(rows).ok())
            .and_then(NonZeroUsize::new)
            .ok_or_else(|| {
                format!(
                    "retention.release_max_rows must be a whole number from 1 to \
                     {MAX_ROWS_MOST}, not {}",
                    self.release_max_rows
                )
            })?;
        let batch = usize::try_from(self.release_batch)
            .ok()
            .and_then(NonZeroUsize::new)
            .ok_or_else(|| {
                format!(
                    "retention.release_batch must be at least 1, not {}",
                    self.release_batch
                )
            })?;
        let versions = match self.keep.versions {
            None => None,
            Some(versions) => Some(NonZeroU64::new(versions).ok_or_else(|| {
                "retention.keep.versions must be at least 1: the head is always kept".to_owned()
            })?),
        };
        if let Some(days) = self.keep.days
            && days > DAYS_MOST
        {
            return Err(format!(
                "retention.keep.days must be at most {DAYS_MOST}, not {days}"
            ));
        }
        Ok(Settings {
            max_rows,
            batch,
            keep: Keep {
                versions,
                days: self.keep.days,
            },
        })
    }
}

/// The last release that failed, kept beside the last that ended: a later
/// success does not erase it.
#[derive(Debug, Clone, Serialize)]
pub struct Failed {
    pub branch: String,
    pub at: Timestamp,
    pub failure: String,
}

/// The last release the thread ran to its end.
#[derive(Debug, Clone, Serialize)]
pub struct Last {
    pub branch: String,
    pub at: Timestamp,
    pub answer: ReleaseAnswer,
    /// It took nothing, and only the keep settings held it.
    pub kept_by_setting: bool,
}

#[derive(Debug, Default)]
struct State {
    waiting: BTreeSet<String>,
    releasing: bool,
    stopping: bool,
    releases: u64,
    last: Option<Last>,
    last_failure: Option<Failed>,
    died: Option<String>,
    durable: SyncCounts,
    /// For each branch the keep days hold, when its held row falls due: the
    /// thread wakes at the earliest, with no write and no restart.
    due: BTreeMap<String, Timestamp>,
    /// A walk of the discarded generations' markers is owed: one page an
    /// idle turn, from `cursor`, until it wraps (outside review 15).
    lap: bool,
    /// The last marker key the walk read; `None` at the first.
    cursor: Option<Vec<u8>>,
    #[cfg(test)]
    seam: Seam,
}

/// What the thread does next.
enum Next {
    Release(String),
    /// Read the page of markers after this key.
    Scan(Option<Vec<u8>>),
}

/// What a test asks of the thread (STOR-003 HR4).
#[cfg(test)]
#[derive(Debug, Default)]
struct Seam {
    kill: bool,
    hold: bool,
    held: bool,
    /// Make the next release stop at this point, as a crash would.
    crash: Option<Crash>,
    /// The storage locks the thread has taken.
    locks: u64,
    /// Nanoseconds the test has moved the clock on.
    clock: u64,
    /// The thread's waits that have ended.
    wakes: u64,
    /// The pages of markers the thread has read.
    pages: u64,
    /// The branches whose release took rows, in order.
    released: Vec<String>,
}

#[derive(Debug, Default)]
struct Shared {
    state: Mutex<State>,
    changed: Condvar,
}

impl Shared {
    fn lock(&self) -> MutexGuard<'_, State> {
        self.state.lock().unwrap_or_else(PoisonError::into_inner)
    }

    fn wait<'a>(&self, state: MutexGuard<'a, State>) -> MutexGuard<'a, State> {
        self.changed
            .wait(state)
            .unwrap_or_else(PoisonError::into_inner)
    }

    fn wait_timeout<'a>(
        &self,
        state: MutexGuard<'a, State>,
        timeout: Duration,
    ) -> MutexGuard<'a, State> {
        self.changed.wait_timeout(state, timeout).map_or_else(
            |poisoned| poisoned.into_inner().0,
            |(state, _timed_out)| state,
        )
    }
}

/// Now, as the thread reads it: the wall clock, moved on by a test.
fn now_in(state: &State) -> Timestamp {
    current_timestamp().saturating_add(clock_moved(state))
}

#[cfg(test)]
const fn clock_moved(state: &State) -> u64 {
    state.seam.clock
}

#[cfg(not(test))]
const fn clock_moved(_state: &State) -> u64 {
    0
}

/// The longest a test's thread waits before it reads the moved clock again;
/// a service's waits only until the row falls due.
#[cfg(test)]
const TEST_WAIT: Duration = Duration::from_millis(10);

/// The release thread and what it shares with the service.
#[derive(Debug)]
pub struct Releaser {
    shared: Arc<Shared>,
    thread: Option<JoinHandle<()>>,
    batch: NonZeroUsize,
    /// Which keep setting is unset, so nothing is ever released.
    unset: Option<&'static str>,
}

/// Which keep setting leaves every version kept, by name.
const fn unset(keep: Keep) -> Option<&'static str> {
    match (keep.versions.is_none(), keep.days.is_none()) {
        (false, false) => None,
        (true, false) => Some("retention.keep.versions is not set"),
        (false, true) => Some("retention.keep.days is not set"),
        (true, true) => Some("retention.keep.versions and retention.keep.days are not set"),
    }
}

impl Releaser {
    /// Finish every decision a crash left, then start the thread over
    /// `storage`, with both keep settings set every live branch waiting.
    ///
    /// # Errors
    /// The decisions unread or unfinished, or the thread unstarted, by
    /// name.
    pub fn start(storage: Arc<Mutex<Storage>>, settings: Settings) -> Result<Self, Failure> {
        let unset = unset(settings.keep);
        let shared = Arc::new(Shared::default());
        let mut writer = storage
            .lock()
            .map_err(|_| Failure::new("writer_unavailable", "writer state requires recovery"))?;
        let now = current_timestamp();
        let rules = ReleaseRules {
            keep: Some(settings.keep),
            max_rows: settings.max_rows,
            now,
        };
        for branch in writer.decided_branches()? {
            let answer = writer
                .release(&branch, None, &rules, &IntentListing)
                .map_err(|failure| {
                    Failure::new(
                        "release_decision_unfinished",
                        format!("{branch}: {failure}"),
                    )
                })?;
            let mut state = shared.lock();
            state.releases = state.releases.saturating_add(1);
            state.last = Some(Last {
                branch,
                at: now,
                answer,
                kept_by_setting: false,
            });
        }
        // Every live branch now; the names a discard left rows of as the
        // thread walks their markers, a page an idle turn
        // (docs/design/HAEM-DISCARDED-ROWS-RELEASE-2026-10-06.md).
        let live = if unset.is_none() {
            writer.branch_names()
        } else {
            Vec::new()
        };
        drop(writer);
        let mut state = shared.lock();
        state.waiting.extend(live);
        state.lap = unset.is_none();
        drop(state);
        let thread_shared = Arc::clone(&shared);
        let thread = std::thread::Builder::new()
            .name("haematite-release".to_owned())
            .spawn(move || {
                let ended = run(&thread_shared, &storage, settings);
                let mut state = thread_shared.lock();
                state.releasing = false;
                if !state.stopping {
                    state.died = Some(ended.err().unwrap_or_else(|| {
                        "release_thread_ended: it stopped without being told".to_owned()
                    }));
                }
                drop(state);
                thread_shared.changed.notify_all();
            })
            .map_err(|error| Failure::new("release_thread_unstarted", error.to_string()))?;
        Ok(Self {
            shared,
            thread: Some(thread),
            batch: settings.batch,
            unset,
        })
    }

    /// Whether `advanced` advances since a branch's last release make it
    /// due: never while a keep setting is unset.
    #[must_use]
    pub const fn due(&self, advanced: usize) -> bool {
        self.unset.is_none() && advanced >= self.batch.get()
    }

    /// `branch` is due: put it in the set, once, and wake the thread.
    pub fn wake(&self, branch: &str) {
        let added = self.shared.lock().waiting.insert(branch.to_owned());
        if added {
            self.shared.changed.notify_all();
        }
    }

    /// `branch` was discarded: its rows wait, and the markers are walked again
    /// from where the last walk stopped, since the discard may end what held
    /// another discarded generation (a fork from it).
    pub fn discarded(&self, branch: &str) {
        let mut state = self.shared.lock();
        state.waiting.insert(branch.to_owned());
        state.lap |= self.unset.is_none();
        drop(state);
        self.shared.changed.notify_all();
    }

    /// Whether the thread has taken a branch to release and not finished it:
    /// a write that finds the storage held then waits instead of being
    /// refused.
    #[must_use]
    pub fn is_releasing(&self) -> bool {
        self.shared.lock().releasing
    }

    /// The thread's state for the service's status answer.
    #[must_use]
    pub fn status(&self) -> Value {
        let finished = self.thread.as_ref().is_none_or(JoinHandle::is_finished);
        let state = self.shared.lock();
        let (name, reason) = match (&state.died, finished, state.stopping) {
            (Some(reason), _, _) => ("died", Some(reason.clone())),
            (None, true, true) => ("stopped", None),
            (None, true, false) => (
                "died",
                Some("release_thread_ended: it stopped without being told".to_owned()),
            ),
            (None, false, _) if state.releasing => ("releasing", None),
            (None, false, _) => ("waiting", None),
        };
        json!({
            "state": name,
            "died": reason,
            "waiting": state.waiting.len(),
            "releases": state.releases,
            "last": state.last,
            "last_failure": state.last_failure,
            "not_released": self.unset,
            "durable": {
                "record_installs": state.durable.record_installs,
                "barrier_device_flushes": state.durable.barrier_device_flushes,
                "ref_file_full_flushes": state.durable.ref_file_full_flushes,
                "directory_full_flushes": state.durable.directory_full_flushes,
            },
        })
    }
}

impl Drop for Releaser {
    /// Tell the thread and join it: a release under way ends its three
    /// operations, and no new one starts.
    fn drop(&mut self) {
        self.shared.lock().stopping = true;
        self.shared.changed.notify_all();
        if let Some(thread) = self.thread.take() {
            // A thread that panicked has nothing left to join; its state is
            // already gone with the service.
            drop(thread.join());
        }
    }
}

/// The thread: take a waiting branch, release it, until told to stop.
fn run(shared: &Shared, storage: &Mutex<Storage>, settings: Settings) -> Result<(), String> {
    loop {
        let mut state = shared.lock();
        let next = loop {
            #[cfg(test)]
            if state.seam.kill {
                return Err("release_thread_killed: by the test seam".to_owned());
            }
            if state.stopping {
                return Ok(());
            }
            if let Some(branch) = state.waiting.pop_first() {
                // Nothing can be taken while a keep setting is unset.
                if unset(settings.keep).is_some() {
                    continue;
                }
                // Set as the branch leaves the set, so no one sees neither.
                state.releasing = true;
                break Next::Release(branch);
            }
            // Nothing waits: each branch whose held row has fallen due is
            // asked again, and the thread sleeps until the earliest next one.
            let now = now_in(&state);
            let fallen: Vec<String> = state
                .due
                .iter()
                .filter(|(_branch, at)| **at <= now)
                .map(|(branch, _at)| branch.clone())
                .collect();
            if !fallen.is_empty() {
                for branch in fallen {
                    state.due.remove(&branch);
                    state.waiting.insert(branch);
                }
                continue;
            }
            if state.lap {
                state.releasing = true;
                break Next::Scan(state.cursor.clone());
            }
            state = match state.due.values().min().copied() {
                Some(at) => {
                    let timeout = Duration::from_nanos(at.saturating_sub(now));
                    #[cfg(test)]
                    let timeout = timeout.min(TEST_WAIT);
                    shared.wait_timeout(state, timeout)
                }
                None => shared.wait(state),
            };
            #[cfg(test)]
            {
                state.seam.wakes = state.seam.wakes.saturating_add(1);
            }
        };
        drop(state);
        match next {
            Next::Release(branch) => release_one(shared, storage, branch, settings)?,
            Next::Scan(after) => scan_one(shared, storage, after.as_deref(), settings)?,
        }
    }
}

/// One page of markers after `after`, under the storage lock: its names
/// with no record wait, and the walk moves past it; a page that reaches the
/// last marker ends the walk and wraps it. A generation whose rows are held
/// is asked once a walk, never every turn (outside review 15).
fn scan_one(
    shared: &Shared,
    storage: &Mutex<Storage>,
    after: Option<&[u8]>,
    settings: Settings,
) -> Result<(), String> {
    let writer = storage
        .lock()
        .map_err(|_| "release_storage_poisoned: the storage lock was poisoned".to_owned())?;
    let page = writer.dead_page(after, settings.max_rows);
    drop(writer);
    let mut state = shared.lock();
    state.releasing = false;
    #[cfg(test)]
    {
        state.seam.pages = state.seam.pages.saturating_add(1);
    }
    match page {
        Ok(page) => {
            state.waiting.extend(page.names);
            state.lap = !page.end;
            state.cursor = if page.end { None } else { page.last };
        }
        // Kept by name; the next discard walks again.
        Err(failure) => {
            state.lap = false;
            let at = now_in(&state);
            state.last_failure = Some(Failed {
                branch: "discarded generations".to_owned(),
                at,
                failure: failure.to_string(),
            });
        }
    }
    drop(state);
    shared.changed.notify_all();
    Ok(())
}

/// One release of `branch` under the storage lock, its outcome kept.
fn release_one(
    shared: &Shared,
    storage: &Mutex<Storage>,
    branch: String,
    settings: Settings,
) -> Result<(), String> {
    #[cfg(test)]
    {
        let mut state = shared.lock();
        state.seam.locks = state.seam.locks.saturating_add(1);
    }
    let mut writer = storage
        .lock()
        .map_err(|_| "release_storage_poisoned: the storage lock was poisoned".to_owned())?;
    #[cfg(test)]
    hold_here(shared);
    #[cfg(test)]
    {
        let crash = shared.lock().seam.crash.take();
        if let Some(point) = crash {
            crash_at(point);
        }
    }
    let before = writer.sync_counts();
    let now = now_in(&shared.lock());
    let rules = ReleaseRules {
        keep: Some(settings.keep),
        max_rows: settings.max_rows,
        now,
    };
    let outcome = writer.release(&branch, Some(settings.max_rows), &rules, &IntentListing);
    let made = writer.sync_counts().since(before);
    drop(writer);
    let mut state = shared.lock();
    state.releasing = false;
    state.durable = plus(state.durable, made);
    match outcome {
        Ok(answer) => {
            state.releases = state.releases.saturating_add(1);
            #[cfg(test)]
            if answer.released.is_some() {
                state.seam.released.push(branch.clone());
            }
            if answer.more {
                state.waiting.insert(branch.clone());
            }
            match answer.due_at {
                Some(at) => {
                    state.due.insert(branch.clone(), at);
                }
                None => {
                    state.due.remove(&branch);
                }
            }
            let kept_by_setting = answer.released.is_none()
                && !answer.held_by.is_empty()
                && answer.held_by.iter().all(|held| {
                    matches!(
                        held.source,
                        ReleaseSource::KeepVersions | ReleaseSource::KeepDays
                    )
                });
            state.last = Some(Last {
                branch,
                at: now,
                answer,
                kept_by_setting,
            });
        }
        // Kept by name and not asked again: the next batch of advances asks.
        Err(failure) => {
            state.due.remove(&branch);
            state.last_failure = Some(Failed {
                branch,
                at: now,
                failure: failure.to_string(),
            });
        }
    }
    drop(state);
    shared.changed.notify_all();
    Ok(())
}

const fn plus(left: SyncCounts, right: SyncCounts) -> SyncCounts {
    SyncCounts {
        record_installs: left.record_installs.saturating_add(right.record_installs),
        barrier_device_flushes: left
            .barrier_device_flushes
            .saturating_add(right.barrier_device_flushes),
        ref_file_full_flushes: left
            .ref_file_full_flushes
            .saturating_add(right.ref_file_full_flushes),
        directory_full_flushes: left
            .directory_full_flushes
            .saturating_add(right.directory_full_flushes),
    }
}

/// Wait here, the storage held, while a test holds the thread.
#[cfg(test)]
fn hold_here(shared: &Shared) {
    let mut state = shared.lock();
    if !state.seam.hold {
        return;
    }
    state.seam.held = true;
    shared.changed.notify_all();
    while state.seam.hold {
        state = shared.wait(state);
    }
    state.seam.held = false;
}

#[cfg(test)]
impl Releaser {
    /// Move the thread's clock on by `nanos`. Nothing is notified: the
    /// thread finds it when its wait for a due row ends.
    pub fn advance_clock(&self, nanos: u64) {
        let mut state = self.shared.lock();
        state.seam.clock = state.seam.clock.saturating_add(nanos);
        drop(state);
    }

    /// The thread's waits that have ended.
    #[must_use]
    pub fn wakes(&self) -> u64 {
        self.shared.lock().seam.wakes
    }

    /// The earliest time a held row falls due, if any.
    #[must_use]
    pub fn next_due(&self) -> Option<Timestamp> {
        self.shared.lock().due.values().min().copied()
    }

    /// Wait until `at_least` releases have ended, at most `bound`: whether
    /// they did.
    #[must_use]
    pub fn wait_releases(&self, at_least: u64, bound: Duration) -> bool {
        let until = std::time::Instant::now() + bound;
        let mut state = self.shared.lock();
        while state.releases < at_least {
            let left = until.saturating_duration_since(std::time::Instant::now());
            if left.is_zero() {
                return false;
            }
            state = self.shared.wait_timeout(state, left);
        }
        drop(state);
        true
    }

    /// End the thread as a fault would.
    pub fn kill(&self) {
        self.shared.lock().seam.kill = true;
        self.shared.changed.notify_all();
    }

    /// Hold the next release once it holds the storage.
    pub fn hold_next(&self) {
        self.shared.lock().seam.hold = true;
    }

    /// Wait until a release is held with the storage.
    pub fn wait_held(&self) {
        let mut state = self.shared.lock();
        while !state.seam.held {
            state = self.shared.wait(state);
        }
        drop(state);
    }

    /// Let a held release go on.
    pub fn let_go(&self) {
        self.shared.lock().seam.hold = false;
        self.shared.changed.notify_all();
    }

    /// Wait until nothing waits and no release runs, or the thread ended.
    pub fn wait_idle(&self) {
        let mut state = self.shared.lock();
        while (!state.waiting.is_empty() || state.releasing || state.lap) && state.died.is_none() {
            state = self.shared.wait(state);
        }
        drop(state);
    }

    /// Wait until the thread is a named ending.
    pub fn wait_died(&self) {
        let mut state = self.shared.lock();
        while state.died.is_none() {
            state = self.shared.wait(state);
        }
        drop(state);
    }

    /// Make the next release stop at `point`, as a crash would.
    pub fn fail_next(&self, point: Crash) {
        self.shared.lock().seam.crash = Some(point);
    }

    /// The pages of markers the thread has read.
    #[must_use]
    pub fn pages(&self) -> u64 {
        self.shared.lock().seam.pages
    }

    /// The branches whose release took rows, in order.
    #[must_use]
    pub fn released(&self) -> Vec<String> {
        self.shared.lock().seam.released.clone()
    }

    /// The storage locks the thread has taken.
    #[must_use]
    pub fn locks(&self) -> u64 {
        self.shared.lock().seam.locks
    }

    #[must_use]
    pub fn last_failure(&self) -> Option<Failed> {
        self.shared.lock().last_failure.clone()
    }

    #[must_use]
    pub fn releases(&self) -> u64 {
        self.shared.lock().releases
    }

    #[must_use]
    pub fn durable(&self) -> SyncCounts {
        self.shared.lock().durable
    }

    #[must_use]
    pub fn last(&self) -> Option<Last> {
        self.shared.lock().last.clone()
    }
}

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