haematite_cli/serve/storage/

release.rs

//! STOR-003 HR1: what a branch's versions are relied on by, which a release
//! never passes (graph-and-query-surface.md, "What a version is relied on
//! by"): the head; a branch forked from it, while that branch exists; an open
//! merge's three moments; a discard asked for and not finished; and the
//! embedding job's position in each declared field (V4).
//!
//! The cost is one tree read for each live branch (its fork origin), one
//! paged read of the open merge rows, and the pending discards; never in
//! proportion to the branch's history. The branches read are counted so the
//! answer can show it.

use std::collections::BTreeMap;
use std::num::{NonZeroU64, NonZeroUsize};

use haematite::SyncCounts;
use haematite::branch::history::Row;
use haematite::branch::{CommitDurability, CommitRequest, KeptHead, Timestamp};
use haematite::db::ReadOnlyDatabase;
use haematite::tree::Hash;
use serde::Serialize;

use super::branches::provenance;
use super::dead;
use super::embedded;
use super::history::{self, Decision, Released};
use super::{
    BRANCH_PREFIX, BranchRefRecord, Captured, Failure, SERVICE, Storage, branch_name, identifier,
    open_branch, version,
};
use crate::serve::protocol::Version;

/// What relies on a version.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum Source {
    Head,
    Fork,
    Merge,
    Discard,
    /// The embedding job's finished `through` or its open walk's target,
    /// named `collection.field`: its next page reads both.
    Embed,
    /// Another branch's companion position naming this branch, as a fork's
    /// inherits its parent's (F2): named `branch:collection.field`.
    EmbedOf,
    /// The branch's own setting to keep its newest versions (STOR-003 HR3).
    KeepVersions,
    /// The branch's own setting to keep versions superseded recently.
    KeepDays,
}

/// One thing relying on a version of the branch: what, by name, and which.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct HeldBy {
    pub source: Source,
    /// The branch for a head or a fork, the merge id, or the discarded branch.
    pub name: String,
    pub version: Version,
}

/// Everything relying on a branch's versions, oldest first.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Floor {
    /// Ordered by (generation, sequence), oldest first. The head is among
    /// them when asked of [`Storage::relied_on`]; asked of
    /// [`Storage::relied_on_by_others`] it is not, and the list may be empty.
    pub held_by: Vec<HeldBy>,
    /// Live branches whose fork origin was read.
    pub branches_read: usize,
}

impl Floor {
    /// The oldest version something relies on: a release takes only what is
    /// below it. `None` when nothing relies on any version: every version may
    /// go.
    pub fn oldest(&self) -> Option<&HeldBy> {
        self.held_by.first()
    }

    /// The oldest version relied on in `generation` (DR-6): a version holds
    /// the rows of its own generation only, so a hold on an older generation
    /// never holds a newer one back.
    pub fn oldest_in(&self, generation: u64) -> Option<Version> {
        self.held_by
            .iter()
            .map(|held| held.version)
            .filter(|version| version.generation == generation)
            .min_by_key(|version| order(*version))
    }
}

/// The pending discards as the store lists them from their intents
/// (STOR-003 H1b2-B): what the release driver reads its floor with.
pub struct IntentListing;

impl PendingDiscards for IntentListing {
    fn pending(&self, storage: &Storage) -> Result<Vec<(String, Version)>, Failure> {
        storage.pending_discards()
    }
}

/// The discards asked for and not finished, as (branch, version) each.
pub trait PendingDiscards {
    /// # Errors
    /// Whatever stops the pending discards from being read.
    fn pending(&self, storage: &Storage) -> Result<Vec<(String, Version)>, Failure>;
}

/// The pending discards until the intent listing in `discards.rs` can be
/// called from here (STOR-003 H1b2): refused by name, so no release ever runs
/// without them. Nothing outside the tests asks for a floor yet (HR3 will),
/// so only they read it.
#[cfg(test)]
pub struct DiscardsUnread;

#[cfg(test)]
impl PendingDiscards for DiscardsUnread {
    fn pending(&self, _storage: &Storage) -> Result<Vec<(String, Version)>, Failure> {
        Err(Failure::new(
            "release_discards_unread",
            "the pending discards cannot be read yet, so nothing is released",
        ))
    }
}

impl Storage {
    /// Everything relying on `branch`'s versions now, its head among them
    /// (STOR-003 HR1).
    ///
    /// # Errors
    /// `branch_missing`; whatever reading a fork origin, a merge row or the
    /// pending discards answers.
    pub fn relied_on(
        &self,
        branch: &str,
        discards: &dyn PendingDiscards,
    ) -> Result<Floor, Failure> {
        identifier(branch)?;
        let head = self
            .refs
            .get(&branch_name(branch))
            .ok_or_else(|| Failure::new("branch_missing", branch))?;
        let head = HeldBy {
            source: Source::Head,
            name: branch.to_owned(),
            version: version(head),
        };
        let mut floor = self.relied_on_by_others(branch, false, discards)?;
        floor.held_by.push(head);
        floor
            .held_by
            .sort_by_key(|held| (held.version.generation, held.version.sequence));
        Ok(floor)
    }

    /// Everything other than `branch`'s own head relying on its versions:
    /// the branches forked from it, the open merges pinning it, and its
    /// pending discard unless `skip_own_discard` (STOR-003 `HR1b`). Its ref may
    /// be hidden or gone, as a discard being finished has it.
    ///
    /// An empty floor means nothing relies on any version of `branch`, so
    /// every version of it may go; it never means that nothing may.
    ///
    /// # Errors
    /// Whatever reading a fork origin, a merge row or the pending discards
    /// answers.
    pub(super) fn relied_on_by_others(
        &self,
        branch: &str,
        skip_own_discard: bool,
        discards: &dyn PendingDiscards,
    ) -> Result<Floor, Failure> {
        identifier(branch)?;
        let mut held_by = Vec::new();
        let mut branches_read = 0_usize;
        for record in self.refs.list_prefix(BRANCH_PREFIX) {
            let Some(fork) = record.name.strip_prefix(BRANCH_PREFIX) else {
                continue;
            };
            if fork == branch {
                continue;
            }
            let at = Captured {
                branch: fork.to_owned(),
                version: version(record),
                shards: record.shards.clone(),
            };
            branches_read = branches_read.saturating_add(1);
            if let Some(origin) = provenance(&self.reader, &at)?
                && origin.branch == branch
            {
                held_by.push(HeldBy {
                    source: Source::Fork,
                    name: fork.to_owned(),
                    version: origin.version,
                });
            }
        }
        for plan in self.open_merge_plans()? {
            for pin in [&plan.origin, &plan.source, &plan.target] {
                if pin.branch == branch {
                    held_by.push(HeldBy {
                        source: Source::Merge,
                        name: plan.id.clone(),
                        version: pin.version,
                    });
                }
            }
        }
        // One pending discard at most names a branch (reconcile refuses two),
        // so skipping the branch's entries skips exactly its own.
        for (discarded, at) in discards.pending(self)? {
            if discarded == branch && !skip_own_discard {
                held_by.push(HeldBy {
                    source: Source::Discard,
                    name: discarded,
                    version: at,
                });
            }
        }
        held_by.extend(self.embed_positions(branch)?);
        held_by.sort_by_key(|held| (held.version.generation, held.version.sequence));
        Ok(Floor {
            held_by,
            branches_read,
        })
    }
}

impl Storage {
    /// The versions of `branch` the embedding job's positions hold, each on
    /// the branch it is a version of (F2b): a position's `through` on the
    /// branch it names, from its own companion (V4) or another's, as a
    /// fork's inherited one names its parent (F2); an open walk's target on
    /// the companion's owner. Every companion is read, one
    /// paged read of its position keys each, so a companion whose branch is
    /// pending discard holds until that discard finishes; the reads are in
    /// proportion to the companions and their declared fields.
    fn embed_positions(&self, branch: &str) -> Result<Vec<HeldBy>, Failure> {
        let mut held = Vec::new();
        for record in self.refs.list_prefix(embedded::COMPANION_PREFIX) {
            let Some(owner) = record.name.strip_prefix(embedded::COMPANION_PREFIX) else {
                continue;
            };
            for ((collection, field), position) in self.companion_positions(record)? {
                let field = format!("{collection}.{field}");
                // `through` is a version of the branch the position names.
                if position.branch == branch
                    && let Some(through) = position.through
                {
                    let (source, name) = if owner == branch {
                        (Source::Embed, field.clone())
                    } else {
                        (Source::EmbedOf, format!("{owner}:{field}"))
                    };
                    held.push(HeldBy {
                        source,
                        name,
                        version: through,
                    });
                }
                // An open walk's target is always the owner's own version
                // (F-2, V2c), whatever branch the position names.
                if owner == branch
                    && let Some(open) = position.toward
                {
                    held.push(HeldBy {
                        source: Source::Embed,
                        name: field,
                        version: open.version,
                    });
                }
            }
        }
        Ok(held)
    }

    /// Every position in a companion, by (collection, field), in key order:
    /// one paged read of its position keys. The one reader of that layout,
    /// for the floor and for a fork's inheriting (F1b).
    ///
    /// # Errors
    /// `embed_position_key_corrupt` for a key that is not UTF-8 or names no
    /// field; `embed_position_corrupt`; whatever a read answers.
    pub(super) fn companion_positions(
        &self,
        companion: &BranchRefRecord,
    ) -> Result<BTreeMap<(String, String), embedded::Position>, Failure> {
        let limit = NonZeroUsize::new(POSITIONS_A_PAGE)
            .ok_or_else(|| Failure::new("embed_positions_page", "a page must be nonzero"))?;
        let mut positions = BTreeMap::new();
        for shard in &companion.shards {
            let mut from = b"e\0".to_vec();
            loop {
                let page = self
                    .reader
                    .range_page_at(shard.shard_id, shard.head, &from, b"e\x01", limit)
                    .map_err(Failure::observer)?;
                for (key, bytes) in page.entries {
                    let field = position_field(&companion.name, &key)?;
                    positions.insert(field, embedded::read_position(&bytes)?);
                }
                let Some(next) = page.next_from else {
                    break;
                };
                from = next;
            }
        }
        Ok(positions)
    }
}

/// The (collection, field) a position key names, refused by name, never
/// guessed, when it is not UTF-8 or has no field part. The key's bytes are
/// never printed.
fn position_field(companion: &str, key: &[u8]) -> Result<(String, String), Failure> {
    let corrupt = || {
        Failure::new(
            "embed_position_key_corrupt",
            format!("{companion} holds a position key that names no collection and field"),
        )
    };
    let named = key.strip_prefix(b"e\0".as_slice()).ok_or_else(corrupt)?;
    let named = std::str::from_utf8(named).map_err(|_error| corrupt())?;
    let (collection, field) = named.split_once('\0').ok_or_else(corrupt)?;
    Ok((collection.to_owned(), field.to_owned()))
}

/// Position keys read a page at a time.
const POSITIONS_A_PAGE: usize = 64;

/// What a branch keeps, resolved from the store's settings and the branch's
/// exception (ruled 07:24). Each setting left `None` keeps all (the design's
/// default), so the default keeps every version.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Keep {
    /// The newest this many versions, the head included.
    pub versions: Option<NonZeroU64>,
    /// Versions superseded fewer than this many days ago.
    pub days: Option<u64>,
}

/// How one release judges and bounds what it takes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Rules {
    /// `None` judges no keep settings at all, as a finishing discard asks
    /// (ruled 08:05); never the same as `Some(Keep::default())`, which keeps
    /// every version.
    pub keep: Option<Keep>,
    /// `retention.release_max_rows`: the most one release may name.
    pub max_rows: NonZeroUsize,
    /// Now, in nanoseconds since the epoch, as row timestamps are.
    pub now: Timestamp,
}

/// The oldest and newest version one release took.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
pub struct Span {
    pub first: Version,
    pub last: Version,
}

/// What `history.release` answers (ruled 07:24, 8).
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct Answer {
    pub branch: String,
    pub released: Option<Span>,
    pub marker: Option<Released>,
    /// What stopped the release short of its limit; empty when nothing did.
    pub held_by: Vec<HeldBy>,
    /// Rows still qualify beyond the limit.
    pub more: bool,
    pub branches_read: usize,
    /// Whether a reclaim pass was asked for after the release.
    pub pass_asked: bool,
    /// Why asking for that pass was refused, when it was.
    pub pass_refused: Option<String>,
    /// When the row the keep days hold falls due: when it was superseded,
    /// plus the days. `None` when no row is held by days with a known
    /// superseding time. Read from the rows this release read, nothing more.
    pub due_at: Option<Timestamp>,
}

const NANOS_A_DAY: u64 = 86_400 * 1_000_000_000;

/// Where a branch's first release starts: below every version it has.
const LOWEST: Version = Version {
    generation: 0,
    sequence: 0,
};

const fn order(version: Version) -> (u64, u64) {
    (version.generation, version.sequence)
}

const fn row_version(row: &Row) -> Version {
    Version {
        generation: row.key.generation,
        sequence: row.key.sequence,
    }
}

/// The newest a release may reach: `version` and below never, and when it
/// was written. A release passes the head; a finishing discard passes the
/// version after its intent's, so the intent's own row may go.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Below {
    pub version: Version,
    pub written: Timestamp,
}

/// What the oldest `cap` of `rows` may go, judged against the floor and the
/// keep settings: the versions taken, what stopped them, and whether the row
/// after the cap would have gone too. `rows` runs from the oldest kept row
/// towards `below`, at most `cap + 1` of them. An empty floor bounds nothing:
/// only `below` does.
///
/// DR-8: until something is taken, rows something relies on are passed, not
/// stopped at: a fork holding an older generation of a live name holds that
/// generation only, so the newer rows behind it in the same read are judged.
/// What was passed counts towards `cap`, and is what held the release when
/// nothing is taken. Taken rows stay one span: nothing is passed after the
/// first is taken.
fn judge(
    branch: &str,
    rows: &[Row],
    cap: usize,
    floor: &Floor,
    below: Below,
    rules: &Rules,
) -> (Vec<Version>, Vec<HeldBy>, bool) {
    let mut taken = Vec::new();
    let mut passed = Vec::new();
    for (index, row) in rows.iter().enumerate() {
        let at = row_version(row);
        if order(at) >= order(below.version) {
            break;
        }
        let held = held_at(branch, rows, index, floor, below, rules);
        if index == cap {
            // Rows passed fill the cap without a release: the next would read
            // and pass the same ones, so no more is asked for.
            let more = held.is_empty() && !taken.is_empty();
            let held_by = if taken.is_empty() { passed } else { Vec::new() };
            return (taken, held_by, more);
        }
        if !held.is_empty() {
            if taken.is_empty() && floor_holds(floor, at) {
                if passed.is_empty() {
                    passed = held;
                }
                continue;
            }
            return (taken, held, false);
        }
        taken.push(at);
    }
    let held_by = if taken.is_empty() { passed } else { Vec::new() };
    (taken, held_by, false)
}

/// Where a release reads its rows: from `from` towards `below`, at most
/// `limit` + 1 a read, and whether the name is live (DR-13).
#[derive(Debug, Clone, Copy)]
struct ReadFrom {
    from: Version,
    below: Below,
    limit: NonZeroUsize,
    live: bool,
}

/// What `judge` answers: the rows taken, what held the release, and more.
type Judged = (Vec<Version>, Vec<HeldBy>, bool);

/// DR-13: the start of the generation after the last of `rows`' first `cap`,
/// when the read filled its cap with rows `floor` relies on and that
/// generation is older than `below`'s; `None` otherwise.
fn past_held(rows: &[Row], cap: usize, floor: &Floor, below: Version) -> Option<Version> {
    if rows.len() <= cap {
        return None;
    }
    let passed = rows.get(..cap)?;
    if !passed
        .iter()
        .all(|row| floor_holds(floor, row_version(row)))
    {
        return None;
    }
    let generation = passed.last()?.key.generation;
    (generation < below.generation).then(|| Version {
        generation: generation.saturating_add(1),
        sequence: 0,
    })
}

/// Whether `floor` relies on `at` or a version before it in its generation.
fn floor_holds(floor: &Floor, at: Version) -> bool {
    let generation = at.generation;
    floor
        .oldest_in(generation)
        .is_some_and(|oldest| order(at) >= order(oldest))
}

/// The rows a decision left behind names, read no further than its count:
/// from its first version, at most `count` + 1 rows, those up to its last.
/// A number found other than its count refuses by name, and the decision
/// is left as it is: never guessed (HR4-fix2).
fn decided_rows(
    reader: &ReadOnlyDatabase,
    service: &BranchRefRecord,
    branch: &str,
    decision: &Decision,
    below: Below,
) -> Result<Vec<Version>, Failure> {
    let read = NonZeroUsize::new(decision.count.saturating_add(1)).unwrap_or(NonZeroUsize::MAX);
    let rows = history::oldest_rows(
        reader,
        service,
        branch,
        (decision.first, below.version),
        read,
    )?;
    let taken: Vec<Version> = rows
        .iter()
        .map(row_version)
        .filter(|at| order(*at) <= order(decision.last))
        .collect();
    if taken.len() != decision.count {
        return Err(Failure::new(
            "release_decision_mismatch",
            format!(
                "the release decided for {branch} names {} rows from {}.{} to {}.{}, and {} \
                 are there; the decision is left",
                decision.count,
                decision.first.generation,
                decision.first.sequence,
                decision.last.generation,
                decision.last.sequence,
                taken.len()
            ),
        ));
    }
    Ok(taken)
}

/// What keeps `rows[index]`: empty when it may go.
fn held_at(
    branch: &str,
    rows: &[Row],
    index: usize,
    floor: &Floor,
    below: Below,
    rules: &Rules,
) -> Vec<HeldBy> {
    let Some(row) = rows.get(index) else {
        return Vec::new();
    };
    let at = row_version(row);
    if let Some(oldest) = floor.oldest_in(at.generation)
        && order(at) >= order(oldest)
    {
        return floor
            .held_by
            .iter()
            .filter(|held| held.version == oldest)
            .cloned()
            .collect();
    }
    let Some(keep) = rules.keep else {
        return Vec::new();
    };
    let kept = |source| {
        vec![HeldBy {
            source,
            name: branch.to_owned(),
            version: at,
        }]
    };
    let Some(versions) = keep.versions else {
        return kept(Source::KeepVersions);
    };
    let Some(days) = keep.days else {
        return kept(Source::KeepDays);
    };
    let head = below.version;
    let newer = newer_than(rows, index, head);
    if newer < versions.get() {
        return kept(Source::KeepVersions);
    }
    // Superseded when the next version was written: the next row's
    // timestamp, or the head's for the last row. Unknown is kept.
    let superseded = match rows.get(index + 1) {
        Some(next) => Some(next.timestamp),
        None if order(at) < order(head) && rows.len() <= index + 1 => None,
        None => Some(below.written),
    };
    let old_enough = superseded.is_some_and(|superseded| {
        rules.now.saturating_sub(superseded) >= days.saturating_mul(NANOS_A_DAY)
    });
    if !old_enough {
        return kept(Source::KeepDays);
    }
    Vec::new()
}

/// When the row the keep days hold falls due: the next row's timestamp, which
/// is when it was superseded, plus the days. `None` when the days hold none,
/// or hold the last row read, whose superseding time is unknown.
fn due_at(rows: &[Row], held_by: &[HeldBy], rules: &Rules) -> Option<Timestamp> {
    let days = rules.keep?.days?;
    let held = held_by
        .iter()
        .find(|held| held.source == Source::KeepDays)?;
    let index = rows
        .iter()
        .position(|row| row_version(row) == held.version)?;
    let superseded = rows.get(index + 1)?.timestamp;
    Some(superseded.saturating_add(days.saturating_mul(NANOS_A_DAY)))
}

/// How many versions are newer than `rows[index]`, the head among them, as
/// far as the rows read show: never more than the truth, so a count short
/// of a keep setting keeps.
///
/// A generation's versions run from sequence 0 up by one: a record is
/// created at seq 0 (haematite `branch/lifecycle.rs`, `create_branch_record`;
/// `branch/fork_changes.rs` for a fork) and each advance writes
/// `expected_seq + 1` only over a record at `expected_seq`
/// (`branch/refstore.rs`, `advance_with_source`). So in the head's
/// generation exactly `head.sequence + 1` versions are newer than a row of
/// an older generation, and `head.sequence - at.sequence` than a row of its
/// own. Rows read after an older row are added only where their generation
/// is older than the head's: the head generation's are already in
/// `head.sequence + 1`. Versions of an older generation with no row read
/// are not counted.
fn newer_than(rows: &[Row], index: usize, head: Version) -> u64 {
    let Some(row) = rows.get(index) else {
        return 0;
    };
    let at = row_version(row);
    if at.generation == head.generation {
        return head.sequence.saturating_sub(at.sequence);
    }
    let older_after = rows
        .get(index + 1..)
        .unwrap_or_default()
        .iter()
        .filter(|later| later.key.generation < head.generation)
        .count();
    // Too few counted keeps; a count that cannot be made is none.
    head.sequence
        .saturating_add(1)
        .saturating_add(u64::try_from(older_after).unwrap_or(0))
}

/// DR-11: an operator's release of a generation discarded before DR-1.
///
/// Such a generation has rows and no marker, so no release finds it; naming
/// it writes its marker, dated by the request, and the ordinary rules then
/// take its rows.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UndatedGeneration {
    pub branch: String,
    /// The generation: its record's creation time, as its rows are keyed.
    pub generation: u64,
    /// Taken as the discard's time, so `keep.days` counts from it.
    pub at: Timestamp,
}

/// The rows of one generation a dating request reads at a time.
const UNDATED_PAGE: usize = 256;

/// A named point in a release's durable order where a test stops it, as a
/// crash would (STOR-003 HR3).
#[cfg(test)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Crash {
    /// The decision row is committed; nothing else is done.
    AfterDecision,
    /// Half the history files are removed.
    DuringFiles,
    /// Every history file is removed; the rows are not.
    AfterFiles,
    /// The rows, the mark and the decision are committed; no pass is asked.
    AfterCommit,
    /// The rows of the kept heads past their days are committed; the record
    /// still keeps them (DR-7).
    BeforeTrim,
}

#[cfg(test)]
thread_local! {
    static CRASH: std::cell::Cell<Option<Crash>> = const { std::cell::Cell::new(None) };
}

/// Stop this thread's next release at `point`.
#[cfg(test)]
pub fn crash_at(point: Crash) {
    CRASH.with(|crash| crash.set(Some(point)));
}

#[cfg(test)]
fn crashed(point: Crash) -> Result<(), Failure> {
    if CRASH.with(|crash| crash.get() == Some(point)) {
        CRASH.with(|crash| crash.set(None));
        return Err(Failure::new("release_crashed", format!("{point:?}")));
    }
    Ok(())
}

impl Storage {
    /// `history.release {branch, limit}` (STOR-003 HR3): takes the oldest
    /// versions of `branch` that nothing relies on and the keep settings let
    /// go, at most `limit`, in this durable order: one decision row; then the
    /// older history files, in one ref operation; then the rows, the mark and
    /// the decision's removal in one commit. A pass is then asked for. A
    /// decision a crash left is finished here first, taking only what may
    /// still go and saying what held the rest.
    ///
    /// Every version taken reads whole until that last commit: its row is
    /// still in the service tree, and every reclaim pass marks from the
    /// service's rows (STOR-003 R2), whether or not its older file is gone.
    ///
    /// No second release runs beside a decision's owner: this takes the
    /// storage mutably, the service holds it behind one lock, and another
    /// process is kept out by the writer lock. A call that meets a decision
    /// left behind finishes that decision as it was written, asking no keep
    /// setting, hold or limit again (HR4-fix2), and decides nothing new.
    ///
    /// # Errors
    /// `release_limit_exceeded` for a limit above `rules.max_rows`; whatever
    /// the floor, a read or a commit answers.
    pub fn release(
        &mut self,
        branch: &str,
        limit: Option<NonZeroUsize>,
        rules: &Rules,
        discards: &dyn PendingDiscards,
    ) -> Result<Answer, Failure> {
        let answer = self.release_once(branch, limit, rules, discards);
        // Ended, whatever it took and whether or not it failed: the driver's
        // count for the branch starts again, so neither a branch its settings
        // keep whole nor one whose release fails is asked again until
        // another batch of advances (STOR-003 HR4, HR4-fix).
        self.kept_rows.released(branch);
        answer
    }

    /// Every live branch, for the driver to ask once at start (STOR-003
    /// HR4-fix): the records in memory, no read.
    #[must_use]
    pub fn branch_names(&self) -> Vec<String> {
        self.refs
            .list_prefix(BRANCH_PREFIX)
            .filter_map(|record| record.name.strip_prefix(BRANCH_PREFIX))
            .map(str::to_owned)
            .collect()
    }

    /// The branches whose release a crash left decided and unfinished:
    /// one range read of the decision rows, never the history (STOR-003
    /// HR4). A decision whose branch has no live record is not named.
    ///
    /// # Errors
    /// Storage failures.
    pub fn decided_branches(&self) -> Result<Vec<String>, Failure> {
        let service = self
            .refs
            .get(SERVICE)
            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
        let decided = history::decided(&self.reader, service)?;
        Ok(self
            .refs
            .list_prefix(BRANCH_PREFIX)
            .filter_map(|record| record.name.strip_prefix(BRANCH_PREFIX))
            .filter(|name| decided.contains(&Hash::of(name.as_bytes())))
            .map(str::to_owned)
            .collect())
    }

    /// `branch`'s advances since its last release (STOR-003 HR4).
    #[must_use]
    pub fn advanced_since_release(&self, branch: &str) -> usize {
        self.kept_rows.advanced(branch)
    }

    /// The durable operations the store has made, for the release driver to
    /// count its own by (STOR-003 HR4).
    #[must_use]
    pub fn sync_counts(&self) -> SyncCounts {
        self.writer.sync_counts()
    }

    fn release_once(
        &mut self,
        branch: &str,
        limit: Option<NonZeroUsize>,
        rules: &Rules,
        discards: &dyn PendingDiscards,
    ) -> Result<Answer, Failure> {
        let limit = match limit {
            Some(limit) if limit > rules.max_rows => {
                return Err(Failure::new(
                    "release_limit_exceeded",
                    format!(
                        "limit {limit} is above retention.release_max_rows {}",
                        rules.max_rows
                    ),
                ));
            }
            Some(limit) => limit,
            None => rules.max_rows,
        };
        let service = self
            .refs
            .get(SERVICE)
            .cloned()
            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
        let live = self.refs.get(&branch_name(branch)).is_some();
        let pending = history::read_decision(&self.reader, &service, branch)?;
        // A decision left behind is the commit point: it is rolled forward
        // as written, and no keep setting, hold or limit is asked again
        // (Apollo 14:24, HR4-fix2). Its files may already be gone.
        let floor = match (&pending, live) {
            (Some(_decision), _) => None,
            (None, true) => Some(self.relied_on(branch, discards)?),
            (None, false) => Some(self.relied_on_by_others(branch, false, discards)?),
        };
        let (below, mark) =
            self.below(&service, branch, pending.as_ref(), floor.as_ref(), limit)?;
        let (taken, held_by, more, branches_read, due) = if let Some(decision) = &pending {
            let taken = decided_rows(&self.reader, &service, branch, decision, below)?;
            (taken, Vec::new(), false, 0, None)
        } else {
            let floor = floor.unwrap_or_default();
            let from = if live {
                LOWEST
            } else {
                past_mark(mark, below.version)
            };
            let read = ReadFrom {
                from,
                below,
                limit,
                live,
            };
            let (rows, (taken, held_by, more)) =
                self.judge_from(&service, branch, read, &floor, rules)?;
            let mut due = due_at(&rows, &held_by, rules);
            if live {
                let trim_due = self.trim_past_days(branch, rules)?;
                due = due.into_iter().chain(trim_due).min();
            }
            (taken, held_by, more, floor.branches_read, due)
        };
        let mut answer = Answer {
            branch: branch.to_owned(),
            released: None,
            marker: mark,
            held_by,
            more,
            branches_read,
            pass_asked: false,
            pass_refused: None,
            due_at: due,
        };
        let span = match (taken.first(), taken.last()) {
            (Some(first), Some(last)) => Some(Span {
                first: *first,
                last: *last,
            }),
            _ => None,
        };
        if pending.is_none() {
            let Some(span) = span else {
                return Ok(answer);
            };
            let decision = Decision {
                first: span.first,
                last: span.last,
                count: taken.len(),
                at: rules.now,
            };
            let bytes = serde_json::to_vec(&decision).map_err(Failure::storage)?;
            self.commit_service(&[(history::decision_key(branch), Some(bytes))])?;
            #[cfg(test)]
            crashed(Crash::AfterDecision)?;
        }
        let ended = dead::ended(&self.reader, &service, branch, &taken)?;
        self.carry_out(branch, &taken, span, rules.now, &ended, &mut answer)?;
        Ok(answer)
    }

    /// Judge `branch`'s oldest rows from `read.from`, at most `limit` + 1 a
    /// read (DR-13). When the first `limit` rows are all relied on by
    /// `floor`, `judge` passes them and takes nothing, and the next release
    /// would read the same ones. The rest of the last passed row's generation
    /// is relied on as well (`floor_holds` is by order within a generation),
    /// so the read starts again at the next generation. Rows are keyed by
    /// generation, then sequence, so that start is a key, not a walk: one
    /// more read for each generation the floor holds, and at most `limit`.
    ///
    /// # Errors
    /// As [`history::oldest_rows`] and [`Self::after_newest`].
    fn judge_from(
        &self,
        service: &BranchRefRecord,
        branch: &str,
        mut read: ReadFrom,
        floor: &Floor,
        rules: &Rules,
    ) -> Result<(Vec<Row>, Judged), Failure> {
        let cap = read.limit.get();
        let ask = NonZeroUsize::new(cap.saturating_add(1)).unwrap_or(NonZeroUsize::MAX);
        let mut skips = 0;
        loop {
            let span = (read.from, read.below.version);
            let mut rows = history::oldest_rows(&self.reader, service, branch, span, ask)?;
            if rows.len() <= cap {
                let newest = rows.last();
                rows.extend(self.after_newest(service, branch, newest, read.below, read.live)?);
            }
            let judged = judge(branch, &rows, cap, floor, read.below, rules);
            match past_held(&rows, cap, floor, read.below.version) {
                Some(from) if skips < cap => {
                    read.from = from;
                    skips += 1;
                }
                _ => return Ok((rows, judged)),
            }
        }
    }

    /// A read that reached `below` knows when its `newest` row was superseded:
    /// a discarded generation's at its discard, a live branch's when the
    /// record's next version was installed, the first it keeps after `newest`
    /// or else the head; no read
    /// (docs/design/HAEM-STALLED-BRANCH-NEWEST-ROW-2026-10-06.md). It is a row
    /// at `below`, which a release never takes. A live row of an older
    /// generation was superseded at that generation's discard, read from its
    /// marker, one entry on each shard (DR-6); with no marker it is unknown,
    /// so kept.
    ///
    /// # Errors
    /// As [`dead::marker`].
    fn after_newest(
        &self,
        service: &BranchRefRecord,
        branch: &str,
        newest: Option<&Row>,
        below: Below,
        live: bool,
    ) -> Result<Option<Row>, Failure> {
        let written = match newest {
            Some(newest) if live && newest.key.generation != below.version.generation => {
                let generation = newest.key.generation;
                match dead::marker(&self.reader, service, branch, generation)? {
                    Some(dead) => dead.discarded_at,
                    None => return Ok(None),
                }
            }
            Some(newest) if live => self
                .refs
                .kept(&branch_name(branch))
                .iter()
                .find(|kept| kept.seq > newest.key.sequence)
                .map_or(below.written, |kept| kept.timestamp),
            _ => below.written,
        };
        Ok(Some(dead::discard_row(branch, below.version, written)))
    }

    /// Where `branch`'s release stops, and the mark of that generation. A
    /// name with no record releases one discarded generation, the discard
    /// counted as the version that superseded its last (DR-6): of its
    /// markers, at most `limit` and oldest first, the first with rows left
    /// past its mark that `floor` does not hold; the oldest when there is
    /// none. A decision left behind keeps the generation it was made for.
    /// One or two point reads of a mark for each generation looked at.
    fn below(
        &self,
        service: &BranchRefRecord,
        branch: &str,
        pending: Option<&Decision>,
        floor: Option<&Floor>,
        limit: NonZeroUsize,
    ) -> Result<(Below, Option<Released>), Failure> {
        if let Some(record) = self.refs.get(&branch_name(branch)) {
            let below = Below {
                version: version(record),
                written: record.timestamp,
            };
            let generation = below.version.generation;
            let mark = history::read_released(&self.reader, service, branch, generation)?;
            return Ok((below, mark));
        }
        let markers = dead::of_name(&self.reader, service, branch, limit)?;
        if let (None, Some(floor)) = (pending, floor)
            && let Some(unheld) = self.first_unheld(service, branch, &markers, floor)?
        {
            return Ok(unheld);
        }
        let dead = pending
            .and_then(|decision| {
                markers
                    .iter()
                    .find(|dead| dead.last.generation == decision.last.generation)
            })
            .or_else(|| markers.first())
            .ok_or_else(|| Failure::new("branch_missing", branch))?;
        let generation = dead.last.generation;
        let mark = history::read_released(&self.reader, service, branch, generation)?;
        Ok((ending(dead), mark))
    }

    /// The first of `markers` with rows left past its generation's mark whose
    /// next row is below what `floor` relies on in that generation, or which
    /// nothing relies on, with its mark: one or two point reads each. A
    /// generation a mark from before DR-6 already covers has none left
    /// (DR-6b).
    fn first_unheld(
        &self,
        service: &BranchRefRecord,
        branch: &str,
        markers: &[dead::Dead],
        floor: &Floor,
    ) -> Result<Option<(Below, Option<Released>)>, Failure> {
        for dead in markers {
            let below = ending(dead);
            let generation = dead.last.generation;
            let mark = history::read_released(&self.reader, service, branch, generation)?;
            let from = past_mark(mark, below.version);
            if order(from) >= order(below.version) {
                continue;
            }
            let held = floor
                .oldest_in(generation)
                .is_some_and(|relied| order(relied) <= order(from));
            if !held {
                return Ok(Some((below, mark)));
            }
        }
        Ok(None)
    }

    /// DR-11: writes the marker `request` names in one service commit: its
    /// last version the generation's newest row, discarded at `request.at`.
    /// Refused by name: `branch_live` while the name has a record (its
    /// release reads every generation), `generation_marked` when the
    /// generation has its marker, `generation_empty` when it has no rows.
    /// Reads the generation's rows once, a page at a time: an operator's one
    /// request, never a release's wake. Answers the marker's last version.
    ///
    /// # Errors
    /// As above; `history_row_corrupt`, `dead_generation_corrupt`; storage
    /// failures.
    pub fn date_undated_generation(
        &mut self,
        request: &UndatedGeneration,
    ) -> Result<Version, Failure> {
        let branch = request.branch.as_str();
        if self.refs.get(&branch_name(branch)).is_some() {
            return Err(Failure::new(
                "branch_live",
                format!("{branch} has a record"),
            ));
        }
        let service = self
            .refs
            .get(SERVICE)
            .cloned()
            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
        let generation = request.generation;
        if dead::marker(&self.reader, &service, branch, generation)?.is_some() {
            return Err(Failure::new(
                "generation_marked",
                format!("{branch} generation {generation} has its marker"),
            ));
        }
        let newest = newest_row(&self.reader, &service, branch, generation)?;
        let Some(last) = newest else {
            return Err(Failure::new(
                "generation_empty",
                format!("{branch} generation {generation} has no rows"),
            ));
        };
        let dead = dead::Dead {
            branch: branch.to_owned(),
            last,
            discarded_at: request.at,
        };
        let bytes = serde_json::to_vec(&dead).map_err(Failure::storage)?;
        self.commit_service(&[(dead::dead_key(branch, generation), Some(bytes))])?;
        Ok(last)
    }

    /// DR-7 (candidate B, docs/design/HAEM-RELEASE-REMAINING-RULES-2026-10-06.md):
    /// the heads `branch`'s record keeps beyond its newest `keep.versions`
    /// whose next version was installed `keep.days` ago leave the record, in
    /// two durable operations in this order: their rows, in one service
    /// commit, as an advance writes them for the heads it drops; then one
    /// install of the record keeping the rest (`trim_kept`). A crash between
    /// the two leaves the rows and the heads still kept: the next release
    /// finds the rows, writes none, and installs the trim. The rows are then
    /// released by the same rules as any other, from the next release on.
    /// Nothing is written while no head is due, and nothing without both
    /// settings. Answers when the next kept head falls due, for the releaser
    /// to wake then: no advance wakes a branch that stopped.
    fn trim_past_days(
        &mut self,
        branch: &str,
        rules: &Rules,
    ) -> Result<Option<Timestamp>, Failure> {
        let Some((versions, days)) = rules.keep.and_then(|keep| keep.versions.zip(keep.days))
        else {
            return Ok(None);
        };
        let name = branch_name(branch);
        let Some(record) = self.refs.get(&name).cloned() else {
            return Ok(None);
        };
        let kept = self.refs.kept(&name);
        if let Some(keep_from) = past_days(&record, kept, versions, days, rules.now) {
            self.rows_before_trim(branch, &record, keep_from)?;
            #[cfg(test)]
            crashed(Crash::BeforeTrim)?;
            self.refs
                .trim_kept(&name, record.created, record.seq, keep_from)
                .map_err(|error| Failure::new("release_kept_untrimmed", error.to_string()))?;
            // The catalogue's readers see the record keep fewer from here on.
            self.publish_record(&name)?;
        }
        Ok(kept_due(&record, self.refs.kept(&name), versions, days))
    }

    /// A decided release's last two durable operations, in order: the files
    /// of `taken` (absent ones included), then its rows, mark and decision in
    /// one commit; then the reclaim pass asked for, a refusal named in
    /// `answer`, never a failure of the release.
    fn carry_out(
        &mut self,
        branch: &str,
        taken: &[Version],
        span: Option<Span>,
        now: Timestamp,
        ended: &[Vec<u8>],
        answer: &mut Answer,
    ) -> Result<(), Failure> {
        let files: Vec<String> = taken
            .iter()
            .map(|at| history::file_name(branch, *at))
            .collect();
        #[cfg(test)]
        if CRASH.with(|crash| crash.get() == Some(Crash::DuringFiles)) {
            let half = files.len() / 2;
            self.remove_files(files.get(..half).unwrap_or_default())?;
            crashed(Crash::DuringFiles)?;
        }
        self.remove_files(&files)?;
        #[cfg(test)]
        crashed(Crash::AfterFiles)?;
        let mut writes: Vec<(Vec<u8>, Option<Vec<u8>>)> = taken
            .iter()
            .map(|at| (history::row_key(branch, *at), None))
            .collect();
        writes.push((history::decision_key(branch), None));
        // A discarded generation's marker goes with its last row.
        writes.extend(ended.iter().map(|key| (key.clone(), None)));
        // DR-6: each generation taken is marked at its last version taken,
        // in the same commit as its rows; the name's old mark is not written.
        for last in last_of_each_generation(taken) {
            let released = Released {
                version: last,
                at: now,
            };
            let bytes = serde_json::to_vec(&released).map_err(Failure::storage)?;
            writes.push((
                history::released_key_at(branch, last.generation),
                Some(bytes),
            ));
        }
        if let Some(span) = span {
            answer.marker = Some(Released {
                version: span.last,
                at: now,
            });
        }
        self.commit_service(&writes)?;
        answer.released = span;
        #[cfg(test)]
        crashed(Crash::AfterCommit)?;
        // The release is durable by now: a pass refused is said by name in
        // the answer, never a failure of the release.
        match self.writer.request_reclaim() {
            Ok(()) => answer.pass_asked = true,
            Err(error) => answer.pass_refused = Some(error.to_string()),
        }
        Ok(())
    }

    /// Remove the older history files `names` in one ref operation, absent
    /// ones included.
    fn remove_files(&mut self, names: &[String]) -> Result<(), Failure> {
        let removed = self
            .refs
            .remove_set(names.iter().map(String::as_str))
            .map_err(|error| Failure::new("release_files_unremoved", error.to_string()))?;
        for record in removed.removed {
            self.visible.remove(&record.name)?;
        }
        Ok(())
    }

    /// One durable service commit of `writes`: a value puts, `None` deletes.
    /// A commit that failed may or may not have installed, so what the kept
    /// rows know is checked again, as for every service commit of rows.
    fn commit_service(&mut self, writes: &[(Vec<u8>, Option<Vec<u8>>)]) -> Result<(), Failure> {
        let committed = self.try_commit_service(writes);
        if committed.is_err() {
            self.kept_rows.mark_unsure();
        }
        committed
    }

    fn try_commit_service(&mut self, writes: &[(Vec<u8>, Option<Vec<u8>>)]) -> Result<(), Failure> {
        let at = self
            .refs
            .get(SERVICE)
            .map(version)
            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
        let service =
            open_branch(SERVICE, &self.refs, &self.registry).map_err(Failure::branch_commit)?;
        for (key, value) in writes {
            let shard = self.writer.shard_for(key);
            value
                .as_ref()
                .map_or_else(
                    || service.delete(shard, key),
                    |value| service.put(shard, key, value),
                )
                .map_err(Failure::branch)?;
        }
        self.writer
            .commit(
                &service,
                &self.registry,
                CommitRequest {
                    durability: CommitDurability::Durable {
                        refs: &mut self.refs,
                    },
                    extra_parents: &[],
                    timestamp: haematite::branch::current_timestamp(),
                },
            )
            .map_err(|error| {
                let installed = self.refs.get(SERVICE).is_some_and(|now| version(now) != at);
                Failure::new(
                    if installed {
                        "durability_unknown"
                    } else {
                        "write_refused"
                    },
                    error.to_string(),
                )
            })?;
        self.publish_record(SERVICE)
    }
}

/// Where a discarded generation's release starts, the release stopping at
/// `below` (DR-6): past its own mark, else at its first version. A mark from
/// before DR-6 at a later generation covers this one whole, so the answer is
/// `below` and the range is empty, never inverted (DR-6b).
const fn past_mark(mark: Option<Released>, below: Version) -> Version {
    match mark {
        Some(mark) if mark.version.generation > below.generation => below,
        Some(mark) => after(mark),
        None => Version {
            generation: below.generation,
            sequence: 0,
        },
    }
}

/// The newest row of `branch`'s `generation`, read oldest first a page at a
/// time (DR-11).
fn newest_row(
    reader: &ReadOnlyDatabase,
    service: &BranchRefRecord,
    branch: &str,
    generation: u64,
) -> Result<Option<Version>, Failure> {
    let page = NonZeroUsize::new(UNDATED_PAGE).unwrap_or(NonZeroUsize::MIN);
    let end = Version {
        generation,
        sequence: u64::MAX,
    };
    let mut from = Version {
        generation,
        sequence: 0,
    };
    let mut newest = None;
    loop {
        let rows = history::oldest_rows(reader, service, branch, (from, end), page)?;
        let Some(last) = rows.last().map(row_version) else {
            return Ok(newest);
        };
        newest = Some(last);
        if rows.len() < page.get() {
            return Ok(newest);
        }
        from = Version {
            generation,
            sequence: last.sequence.saturating_add(1),
        };
    }
}

/// The first seq `record` keeps on once DR-7's trim at `now` is done: the
/// kept heads below it are beyond the newest `versions`, the head counted,
/// and were superseded `days` ago or more, by the next kept head's install
/// or, for the newest, the head's. `None` when no head is.
fn past_days(
    record: &BranchRefRecord,
    kept: &[KeptHead],
    versions: NonZeroU64,
    days: u64,
    now: Timestamp,
) -> Option<u64> {
    let age = days.saturating_mul(NANOS_A_DAY);
    let superseded = kept
        .iter()
        .skip(1)
        .map(|next| next.timestamp)
        .chain(std::iter::once(record.timestamp));
    let leaving = kept
        .iter()
        .zip(superseded)
        .take_while(|(entry, superseded)| {
            record.seq.saturating_sub(entry.seq) >= versions.get()
                && now.saturating_sub(*superseded) >= age
        })
        .count();
    let keep_from = kept.get(leaving).map_or(record.seq, |entry| entry.seq);
    (leaving > 0).then_some(keep_from)
}

/// When the oldest head `record` keeps beyond its newest `versions` falls
/// due under `days`: its next version's install, plus the days.
fn kept_due(
    record: &BranchRefRecord,
    kept: &[KeptHead],
    versions: NonZeroU64,
    days: u64,
) -> Option<Timestamp> {
    let index = kept
        .iter()
        .position(|entry| record.seq.saturating_sub(entry.seq) >= versions.get())?;
    let superseded = kept
        .get(index + 1)
        .map_or(record.timestamp, |next| next.timestamp);
    Some(superseded.saturating_add(days.saturating_mul(NANOS_A_DAY)))
}

/// Where a release of `dead`'s generation stops: past its last version,
/// superseded when it was discarded.
const fn ending(dead: &dead::Dead) -> Below {
    Below {
        version: Version {
            generation: dead.last.generation,
            sequence: dead.last.sequence.saturating_add(1),
        },
        written: dead.discarded_at,
    }
}

/// The last version of each generation in `taken`, which runs oldest first.
fn last_of_each_generation(taken: &[Version]) -> Vec<Version> {
    let mut last: Vec<Version> = Vec::new();
    for at in taken {
        match last.last_mut() {
            Some(previous) if previous.generation == at.generation => *previous = *at,
            _ => last.push(*at),
        }
    }
    last
}

/// The version just after `released`, where the next release starts.
const fn after(released: Released) -> Version {
    Version {
        generation: released.version.generation,
        sequence: released.version.sequence.saturating_add(1),
    }
}

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