haematite-cli 0.21.8

//! Predecessor rows are durable before the live reference can advance.

use haematite::branch::history::{Key, Row, ShardRoot};
use haematite::branch::{CommitDurability, CommitRequest, KeptHead, KeptLimit, Timestamp};
use haematite::db::ReadOnlyDatabase;
use haematite::tree::Hash;
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet};
use std::num::NonZeroUsize;

use super::{
    BranchKind, BranchRefRecord, BranchRefStore, BranchShardRef, Captured, Failure, SERVICE,
    Storage, current_timestamp, open_branch,
};
use crate::serve::protocol::Version;

/// STOR-003 `MD5a`: how many previous versions each branch record keeps when
/// the serve configuration names no number, and the most it may name.
pub const KEPT_VERSIONS_DEFAULT: usize = 64;
pub const KEPT_VERSIONS_MOST: usize = 4_096;

/// The limit a store keeps when the configuration names none.
///
/// # Errors
/// Never in practice: the default is not zero.
pub fn default_kept() -> Result<KeptLimit, Failure> {
    KeptLimit::new(KEPT_VERSIONS_DEFAULT).map_err(Failure::branch_ref)
}

/// STOR-003 `MD5a`: the kept window's limit, and which kept versions of each
/// branch generation are known to have their history row. A generation is
/// read once, the first time an advance of it would drop a version, by one
/// range read of at most the window's rows; after that this is kept as rows
/// are written.
#[derive(Debug)]
pub struct KeptRows {
    limit: KeptLimit,
    known: BTreeMap<(String, u64), BTreeSet<u64>>,
    /// A row commit failed with its outcome unknown: the next advance fences
    /// the service record before it relies on a row it reads.
    unsure: bool,
    /// For each branch, its advances since its last release (STOR-003 HR4):
    /// each moves one version towards the window's edge, where it gets its
    /// row, so this is what the release driver's batch is measured in. Kept
    /// in memory only: an open starts each branch at none.
    advanced: BTreeMap<String, usize>,
}

impl KeptRows {
    #[must_use]
    pub const fn new(limit: KeptLimit) -> Self {
        Self {
            limit,
            known: BTreeMap::new(),
            unsure: false,
            advanced: BTreeMap::new(),
        }
    }

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

    /// A release of `branch` ran to its end: its count starts again.
    pub fn released(&mut self, branch: &str) {
        self.advanced.remove(branch);
    }

    /// A service commit failed with its outcome unknown, a release's as a
    /// row's (STOR-003 HR3): the next advance fences before relying on a row.
    pub const fn mark_unsure(&mut self) {
        self.unsure = true;
    }
}

/// Where a branch's released mark is kept in the service namespace, followed
/// by the branch name's hash. Outside `h\0`, so no pass reads it as a row.
const RELEASED_PREFIX: &[u8] = b"released\0";

/// The last version of a branch whose history was released, and when
/// (STOR-003 HR2): every version at or below it, by (generation, sequence),
/// is gone. Since DR-6 one for each generation of a name, so a generation
/// something holds no longer holds the newer ones back; the one name mark
/// written before DR-6 is still read for every generation up to its own.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Released {
    pub version: Version,
    /// When the release was committed, in nanoseconds since the epoch.
    pub at: Timestamp,
}

impl Released {
    /// Whether `version` is at or below the mark, generation first.
    pub const fn covers(&self, version: Version) -> bool {
        version.generation < self.version.generation
            || (version.generation == self.version.generation
                && version.sequence <= self.version.sequence)
    }

    /// `version_released`, naming the version asked for and the mark.
    pub fn refusal(&self, branch: &str, version: Version) -> Failure {
        Failure::new(
            "version_released",
            format!(
                "version {}.{} of branch {branch} was released: history up to {}.{} was \
                 released at {}",
                version.generation,
                version.sequence,
                self.version.generation,
                self.version.sequence,
                self.at
            ),
        )
    }
}

/// Where an unfinished release's decision is kept, followed by the branch
/// name's hash (STOR-003 HR3). Outside `h\0`, as the mark is.
const DECISION_PREFIX: &[u8] = b"release-decision\0";

/// A release decided and not yet finished: the oldest `count` versions of a
/// branch from `first` to `last`, decided at `at` (STOR-003 HR3).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Decision {
    pub first: Version,
    pub last: Version,
    pub count: usize,
    pub at: Timestamp,
}

/// The branch-name hashes of every unfinished release decision at
/// `service`'s roots: one range read of the decision prefix on each shard,
/// in proportion to the decisions left, never to the history (STOR-003 HR4).
///
/// # Errors
/// Storage failures.
pub(super) fn decided(
    reader: &ReadOnlyDatabase,
    service: &BranchRefRecord,
) -> Result<BTreeSet<Hash>, Failure> {
    let mut end = DECISION_PREFIX.to_vec();
    if let Some(last) = end.last_mut() {
        *last = last.saturating_add(1);
    }
    let page = NonZeroUsize::new(256).unwrap_or(NonZeroUsize::MIN);
    let mut found = BTreeSet::new();
    for shard in &service.shards {
        let mut from = DECISION_PREFIX.to_vec();
        loop {
            let read = reader
                .range_page_at(shard.shard_id, shard.head, &from, &end, page)
                .map_err(Failure::observer)?;
            for (key, _value) in read.entries {
                let named = key.get(DECISION_PREFIX.len()..).unwrap_or_default();
                let bytes: [u8; 32] = named
                    .try_into()
                    .map_err(|_| Failure::new("release_decision_corrupt", "decision key"))?;
                found.insert(Hash::from_bytes(bytes));
            }
            let Some(next) = read.next_from else {
                break;
            };
            from = next;
        }
    }
    Ok(found)
}

pub(super) fn decision_key(branch: &str) -> Vec<u8> {
    let mut key = Vec::with_capacity(DECISION_PREFIX.len() + 32);
    key.extend_from_slice(DECISION_PREFIX);
    key.extend_from_slice(Hash::of(branch.as_bytes()).as_bytes());
    key
}

/// `branch`'s unfinished release decision at `service`'s roots, if any.
///
/// # Errors
/// `release_decision_corrupt` for one that does not decode; storage failures.
pub(super) fn read_decision(
    reader: &ReadOnlyDatabase,
    service: &BranchRefRecord,
    branch: &str,
) -> Result<Option<Decision>, Failure> {
    let key = decision_key(branch);
    let shard_id = reader.shard_for(&key);
    let root = service
        .shards
        .get(shard_id)
        .ok_or_else(|| Failure::new("incomplete_roots", "service row shard is missing"))?
        .head;
    reader
        .get_at(shard_id, root, &key)
        .map_err(Failure::observer)?
        .map(|bytes| {
            serde_json::from_slice(&bytes)
                .map_err(|error| Failure::new("release_decision_corrupt", error.to_string()))
        })
        .transpose()
}

/// The key of `branch`'s history row for `version`.
pub(super) fn row_key(branch: &str, version: Version) -> Vec<u8> {
    Key {
        branch: Hash::of(branch.as_bytes()),
        generation: version.generation,
        sequence: version.sequence,
    }
    .encode()
}

/// The name of the older history file for `branch` at `version`, as
/// [`require_file_rows`] reads it.
pub(super) fn file_name(branch: &str, version: Version) -> String {
    format!(
        "@haem/history/{}/{}/{:020}",
        Hash::of(branch.as_bytes()),
        version.generation,
        version.sequence
    )
}

/// `branch`'s oldest history rows from `from` up to, not including, `below`,
/// in (generation, sequence) order, at most `limit`: at most `limit` from
/// each shard, so in proportion to `limit` and never to the history.
///
/// # Errors
/// `history_row_corrupt` for a row that does not decode; storage failures.
pub(super) fn oldest_rows(
    reader: &ReadOnlyDatabase,
    service: &BranchRefRecord,
    branch: &str,
    (from, below): (Version, Version),
    limit: NonZeroUsize,
) -> Result<Vec<Row>, Failure> {
    validate_service(reader, service)?;
    #[cfg(test)]
    ROW_READS.with(|reads| reads.set(reads.get() + 1));
    let (start, end) = (row_key(branch, from), row_key(branch, below));
    let mut rows = Vec::new();
    for shard in &service.shards {
        let page = reader
            .range_page_at(shard.shard_id, shard.head, &start, &end, limit)
            .map_err(Failure::observer)?;
        for (key, bytes) in page.entries {
            let row = Row::decode(&key, &bytes)
                .map_err(|error| Failure::new("history_row_corrupt", error.to_string()))?;
            rows.push(row);
        }
    }
    rows.sort_unstable_by_key(|row| (row.key.generation, row.key.sequence));
    rows.truncate(limit.get());
    Ok(rows)
}

/// The name's mark from before DR-6; no release writes it now.
pub(super) fn released_key(branch: &str) -> Vec<u8> {
    let mut key = Vec::with_capacity(RELEASED_PREFIX.len() + 32 + 8);
    key.extend_from_slice(RELEASED_PREFIX);
    key.extend_from_slice(Hash::of(branch.as_bytes()).as_bytes());
    key
}

/// The mark of one generation of `branch` (DR-6): the name's key, then the
/// generation big-endian.
pub(super) fn released_key_at(branch: &str, generation: u64) -> Vec<u8> {
    let mut key = released_key(branch);
    key.extend_from_slice(&generation.to_be_bytes());
    key
}

/// `branch`'s released mark for `generation` at `service`'s roots: the
/// generation's own mark, else the name's mark from before DR-6 when it
/// reaches that generation. One point read, two when the generation has no
/// mark of its own.
///
/// # Errors
/// `history_marker_corrupt` for a mark that does not decode; storage failures.
pub(super) fn read_released(
    reader: &ReadOnlyDatabase,
    service: &BranchRefRecord,
    branch: &str,
    generation: u64,
) -> Result<Option<Released>, Failure> {
    if let Some(mark) = read_mark(reader, service, &released_key_at(branch, generation))? {
        return Ok(Some(mark));
    }
    let legacy = read_mark(reader, service, &released_key(branch))?;
    Ok(legacy.filter(|mark| mark.version.generation >= generation))
}

/// The mark under `key`, if any was written.
fn read_mark(
    reader: &ReadOnlyDatabase,
    service: &BranchRefRecord,
    key: &[u8],
) -> Result<Option<Released>, Failure> {
    #[cfg(test)]
    MARK_READS.with(|reads| reads.set(reads.get() + 1));
    let shard_id = reader.shard_for(key);
    let root = service
        .shards
        .get(shard_id)
        .ok_or_else(|| Failure::new("incomplete_roots", "service row shard is missing"))?
        .head;
    reader
        .get_at(shard_id, root, key)
        .map_err(Failure::observer)?
        .map(|bytes| {
            serde_json::from_slice(&bytes)
                .map_err(|error| Failure::new("history_marker_corrupt", error.to_string()))
        })
        .transpose()
}

fn validate_service(reader: &ReadOnlyDatabase, service: &BranchRefRecord) -> Result<(), Failure> {
    if service.kind != BranchKind::Namespace {
        return Err(Failure::new(
            "history_service_kind",
            "history service must be a namespace",
        ));
    }
    if service.shards.len() != reader.shard_count()
        || service
            .shards
            .iter()
            .enumerate()
            .any(|(id, shard)| shard.shard_id != id)
    {
        return Err(Failure::new(
            "incomplete_roots",
            "service does not name every physical shard",
        ));
    }
    Ok(())
}

fn capture_row(reader: &ReadOnlyDatabase, branch: &str, row: Row) -> Result<Captured, Failure> {
    if row.shards.len() != reader.shard_count()
        || row
            .shards
            .iter()
            .enumerate()
            .any(|(id, shard)| shard.shard_id != id)
    {
        return Err(Failure::new(
            "history_row_corrupt",
            "history row does not name every physical shard",
        ));
    }
    Ok(Captured {
        branch: branch.to_owned(),
        version: Version {
            generation: row.key.generation,
            sequence: row.key.sequence,
        },
        shards: row
            .shards
            .into_iter()
            .map(|shard| BranchShardRef {
                shard_id: shard.shard_id,
                // A kept reference diverges at the head it preserves.
                fork_anchor: shard.root,
                head: shard.root,
            })
            .collect(),
    })
}

pub(super) fn capture_kept(
    reader: &ReadOnlyDatabase,
    service: &BranchRefRecord,
    branch: &str,
    requested: Version,
) -> Result<Option<Captured>, Failure> {
    validate_service(reader, service)?;
    let key = Key {
        branch: Hash::of(branch.as_bytes()),
        generation: requested.generation,
        sequence: requested.sequence,
    }
    .encode();
    let shard_id = reader.shard_for(&key);
    let root = service
        .shards
        .get(shard_id)
        .ok_or_else(|| Failure::new("incomplete_roots", "service row shard is missing"))?
        .head;
    reader
        .get_at(shard_id, root, &key)
        .map_err(Failure::observer)?
        .map(|bytes| {
            let row = Row::decode(&key, &bytes)
                .map_err(|error| Failure::new("history_row_corrupt", error.to_string()))?;
            capture_row(reader, branch, row)
        })
        .transpose()
}

/// `requested` from the heads `current`'s record keeps, captured as a row
/// of it would be (anchor and head the kept root on each shard), or `None`
/// when the record does not keep it (STOR-003 MD4).
pub(super) fn capture_from_kept(
    current: &Captured,
    kept: &[KeptHead],
    requested: Version,
) -> Option<Captured> {
    if requested.generation != current.version.generation {
        return None;
    }
    kept.iter()
        .find(|entry| entry.seq == requested.sequence)
        .map(|entry| kept_capture(current, entry))
}

fn kept_capture(current: &Captured, entry: &KeptHead) -> Captured {
    Captured {
        branch: current.branch.clone(),
        version: Version {
            generation: current.version.generation,
            sequence: entry.seq,
        },
        shards: current
            .shards
            .iter()
            .zip(&entry.heads)
            .map(|(shard, head)| BranchShardRef {
                shard_id: shard.shard_id,
                fork_anchor: *head,
                head: *head,
            })
            .collect(),
    }
}

/// The refusal for a version of a generation that has ended: the branch was
/// discarded, or recreated since (STOR-003 MD4). Never `version_missing`, so
/// a caller can tell an ended generation from a version that never was.
pub(super) fn generation_ended(branch: &str, requested: Version) -> Failure {
    Failure::new(
        "version_generation_ended",
        format!(
            "version {}.{} of branch {branch} belongs to a generation that has ended",
            requested.generation, requested.sequence
        ),
    )
}

/// The branch's versions after `after`, oldest first: the record's kept heads
/// for the newest below its head, rows below them, then the head itself.
pub(super) fn list_kept(
    reader: &ReadOnlyDatabase,
    service: &BranchRefRecord,
    current: &Captured,
    kept: &[KeptHead],
    after: Option<Version>,
    limit: NonZeroUsize,
) -> Result<Vec<Captured>, Failure> {
    validate_service(reader, service)?;
    if after.is_some_and(|at| at.sequence >= current.version.sequence) {
        return Ok(Vec::new());
    }
    // What was released is no gap: the list starts after the mark when the
    // mark is in this generation (STOR-003 HR2).
    let generation = current.version.generation;
    let first_kept = match read_released(reader, service, &current.branch, generation)? {
        Some(released) if released.version.generation == current.version.generation => {
            let sequence = released.version.sequence;
            sequence
                .checked_add(1)
                .ok_or_else(|| Failure::new("history_marker_corrupt", "mark cannot advance"))?
        }
        _ => 0,
    };
    let mut expected = match after {
        Some(at) => at.sequence.checked_add(1).ok_or_else(|| {
            Failure::new("invalid_continuation", "history sequence cannot advance")
        })?,
        None => 0,
    }
    .max(first_kept);
    // STOR-003 MD4: the record's kept heads name the newest versions below
    // the head, so rows are read only below the first of them.
    let window: Vec<Captured> = kept
        .iter()
        .filter(|entry| entry.seq >= expected && entry.seq < current.version.sequence)
        .map(|entry| kept_capture(current, entry))
        .collect();
    let rows_end = window
        .first()
        .map_or(current.version.sequence, |first| first.version.sequence);
    let mut rows = Vec::new();
    if expected < rows_end {
        let from = Key {
            branch: Hash::of(current.branch.as_bytes()),
            generation: current.version.generation,
            sequence: expected,
        };
        let to = Key {
            sequence: rows_end,
            ..from
        }
        .encode();
        let from = from.encode();
        for shard in &service.shards {
            let page = reader
                .range_page_at(shard.shard_id, shard.head, &from, &to, limit)
                .map_err(Failure::observer)?;
            for (key, bytes) in page.entries {
                if reader.shard_for(&key) != shard.shard_id {
                    return Err(Failure::new(
                        "history_row_corrupt",
                        "history row is in the wrong shard",
                    ));
                }
                let row = Row::decode(&key, &bytes)
                    .map_err(|error| Failure::new("history_row_corrupt", error.to_string()))?;
                rows.push(capture_row(reader, &current.branch, row)?);
            }
        }
    }
    rows.sort_unstable_by_key(|row| row.version.sequence);
    rows.truncate(limit.get());
    let room = limit.get().saturating_sub(rows.len());
    rows.extend(window.into_iter().take(room));
    for row in &rows {
        if row.version.sequence != expected {
            return Err(Failure::new("version_missing", "history row is missing"));
        }
        expected = expected.checked_add(1).ok_or_else(|| {
            Failure::new(
                "history_row_corrupt",
                "retained history sequence cannot advance",
            )
        })?;
    }
    if rows.len() < limit.get() {
        if expected != current.version.sequence {
            return Err(Failure::new("version_missing", "history row is missing"));
        }
        rows.push(current.clone());
    }
    Ok(rows)
}

pub(super) fn require_file_rows(
    reader: &ReadOnlyDatabase,
    refs: &BranchRefStore,
) -> Result<(), Failure> {
    let service = refs
        .get(SERVICE)
        .ok_or_else(|| Failure::new("not_served", "missing service namespace"))?;
    validate_service(reader, service)?;
    let mut missing = 0_usize;
    for record in refs.list_prefix("@haem/history/") {
        let key = file_key(&record.name)?.encode();
        let shard_id = reader.shard_for(&key);
        let root = service
            .shards
            .get(shard_id)
            .ok_or_else(|| Failure::new("incomplete_roots", "service row shard is missing"))?
            .head;
        // Presence is checked here; reads decode only the rows they select.
        if reader
            .get_at(shard_id, root, &key)
            .map_err(Failure::observer)?
            .is_none()
        {
            missing = missing.checked_add(1).ok_or_else(|| {
                Failure::new(
                    "history_rows_count_overflow",
                    "missing row count cannot advance",
                )
            })?;
        }
    }
    if missing != 0 {
        return Err(Failure::new(
            "history_rows_incomplete",
            format!("{missing} history files have no matching rows"),
        ));
    }
    Ok(())
}

fn file_key(name: &str) -> Result<Key, Failure> {
    haematite::branch::history::file_key(name).ok_or_else(|| {
        Failure::new(
            "history_ref_corrupt",
            format!("invalid history reference {name}"),
        )
    })
}

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

#[cfg(test)]
#[path = "history_kept_tests.rs"]
mod kept_tests;

#[cfg(test)]
#[path = "history_spill_tests.rs"]
mod spill_tests;

fn kept_row(branch: &str, record: &BranchRefRecord) -> Row {
    Row {
        key: Key {
            branch: Hash::of(branch.as_bytes()),
            generation: record.created,
            sequence: record.seq,
        },
        timestamp: record.timestamp,
        shards: record
            .shards
            .iter()
            .map(|shard| ShardRoot {
                shard_id: shard.shard_id,
                root: shard.head,
            })
            .collect(),
    }
}

#[cfg(test)]
thread_local! {
    /// Range reads of a kept window made on this thread (STOR-003 `MD5a`).
    pub static WINDOW_READS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
    /// Service commits of kept rows made on this thread (STOR-003 `MD5a`).
    pub static ROW_COMMITS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
    /// Service commits of a head's row by `retain_history_row` on this thread
    /// (STOR-003 `MD5b`).
    pub static RETAIN_COMMITS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
    /// Point reads of a released mark made on this thread (DR-6).
    pub static MARK_READS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
    /// Reads of one generation's discard marker made on this thread (DR-6).
    pub static MARKER_READS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
    /// Reads of a branch's oldest rows made on this thread (DR-8).
    pub static ROW_READS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}

/// STOR-003 `MD5a`: a crash between a write's row commit and its data commit,
/// injected once on this thread.
#[cfg(test)]
pub mod after_rows {
    use std::cell::Cell;

    use super::Failure;

    thread_local! {
        static ARMED: Cell<bool> = const { Cell::new(false) };
    }

    pub fn arm() {
        ARMED.with(|armed| armed.set(true));
    }

    pub fn reached() -> Result<(), Failure> {
        if ARMED.with(|armed| armed.replace(false)) {
            return Err(Failure::new(
                "injected_crash",
                "stopped after the rows and before the data commit",
            ));
        }
        Ok(())
    }
}

/// A test-only point at which a row commit that installed fails as if its
/// flush did, its durability unknown (STOR-003 `MD5c-r2`).
#[cfg(test)]
pub mod row_install {
    use std::cell::Cell;

    use super::Failure;

    thread_local! {
        static ARMED: Cell<bool> = const { Cell::new(false) };
    }

    pub fn arm() {
        ARMED.with(|armed| armed.set(true));
    }

    pub fn reached() -> Result<(), Failure> {
        if ARMED.with(|armed| armed.replace(false)) {
            return Err(Failure::new(
                "history_row_durability_unknown",
                "installed, then failed as if its flush did",
            ));
        }
        Ok(())
    }
}

/// A test-only point at which the fence of the service record fails
/// (STOR-003 `MD5c-r3`).
#[cfg(test)]
pub mod fence_fails {
    use std::cell::Cell;

    use super::Failure;

    thread_local! {
        static ARMED: Cell<bool> = const { Cell::new(false) };
    }

    pub fn arm() {
        ARMED.with(|armed| armed.set(true));
    }

    pub fn reached() -> Result<(), Failure> {
        if ARMED.with(|armed| armed.replace(false)) {
            return Err(Failure::new(
                "history_row_durability_unknown",
                "the fence of the service record failed",
            ));
        }
        Ok(())
    }
}

/// Whether a version already has its row: an existing row with the same
/// roots on the same shards, in the same order, is that row whatever its
/// timestamp; any other is `history_row_conflict`. Every site that asks uses
/// this one rule (STOR-003 `MD5a`).
fn same_roots(existing: &Row, made: &Row) -> Result<(), Failure> {
    if existing.shards == made.shards {
        return Ok(());
    }
    Err(Failure::new(
        "history_row_conflict",
        "kept version row has different shard roots",
    ))
}

/// The row `retain_history_row` wrote for version `entry` of `record`'s
/// generation when it was the head: its heads on `record`'s shards, in order,
/// at its own timestamp.
fn kept_head_row(branch: &str, record: &BranchRefRecord, entry: &KeptHead) -> Row {
    Row {
        key: Key {
            branch: Hash::of(branch.as_bytes()),
            generation: record.created,
            sequence: entry.seq,
        },
        timestamp: entry.timestamp,
        shards: record
            .shards
            .iter()
            .zip(&entry.heads)
            .map(|(shard, head)| ShardRoot {
                shard_id: shard.shard_id,
                root: *head,
            })
            .collect(),
    }
}

impl Storage {
    /// STOR-003 `MD5a`: what an advance of `branch` from `record` needs first.
    /// The advance moves `record`'s head into its kept window and drops the
    /// oldest kept versions past the limit. When one of those has no history
    /// row, every kept version without one is given its row in one service
    /// commit, so the next advances drop versions that have theirs; otherwise
    /// nothing is written.
    ///
    /// # Errors
    /// `history_row_conflict` when a row in the window differs from the one
    /// its kept head makes; as `retain_history_row` for the commit.
    pub(super) fn rows_before_advance(
        &mut self,
        branch: &str,
        record: &BranchRefRecord,
    ) -> Result<(), Failure> {
        let advanced = self
            .kept_rows
            .advanced
            .entry(branch.to_owned())
            .or_insert(0);
        *advanced = advanced.saturating_add(1);
        let kept = self.refs.kept(&record.name);
        let dropping = (kept.len() + 1).saturating_sub(self.kept_rows.limit.get());
        let leaving: Vec<u64> = kept.iter().take(dropping).map(|entry| entry.seq).collect();
        self.rows_before_leaving(branch, record, &leaving)
    }

    /// DR-7: before the heads `record` keeps below seq `keep_from` leave it
    /// with no advance, as [`Self::rows_before_advance`] before the heads an
    /// advance drops: every kept version without a row is given its row in
    /// one service commit when one leaving has none; otherwise nothing is
    /// written.
    ///
    /// # Errors
    /// As [`Self::rows_before_advance`].
    pub(super) fn rows_before_trim(
        &mut self,
        branch: &str,
        record: &BranchRefRecord,
        keep_from: u64,
    ) -> Result<(), Failure> {
        let leaving: Vec<u64> = self
            .refs
            .kept(&record.name)
            .iter()
            .map(|entry| entry.seq)
            .filter(|seq| *seq < keep_from)
            .collect();
        self.rows_before_leaving(branch, record, &leaving)
    }

    /// The kept versions `leaving` given their rows, as the two above ask.
    fn rows_before_leaving(
        &mut self,
        branch: &str,
        record: &BranchRefRecord,
        leaving: &[u64],
    ) -> Result<(), Failure> {
        if leaving.is_empty() {
            return Ok(());
        }
        let kept = self.refs.kept(&record.name).to_vec();
        self.fence_if_unsure()?;
        let generation = (branch.to_owned(), record.created);
        let known = match self.kept_rows.known.remove(&generation) {
            Some(known) => known,
            None => self.rows_in_window(branch, record, &kept, false)?,
        };
        if leaving.iter().all(|seq| known.contains(seq)) {
            self.kept_rows.known.insert(generation, known);
            return Ok(());
        }
        let rows: Vec<Row> = kept
            .iter()
            .filter(|entry| !known.contains(&entry.seq))
            .map(|entry| kept_head_row(branch, record, entry))
            .collect();
        self.commit_rows(&rows)?;
        // Every kept version now has its row; what was known below the window
        // is let go and what was learned within it is kept (STOR-003 MD5b).
        let oldest = kept.first().map_or(0, |entry| entry.seq);
        let mut written = known;
        written.retain(|seq| *seq >= oldest);
        written.extend(kept.iter().map(|entry| entry.seq));
        self.kept_rows.known.insert(generation, written);
        Ok(())
    }

    /// STOR-003 `MD5a`: before a fork from version `at` of `branch`, whose
    /// current record is `record`: when `at` is a version the record keeps
    /// and has no row, its row is written first, in one service commit, as
    /// `retain_history_row` wrote it when `at` was the head. A fork from the
    /// head, or from a version older than the window, writes nothing here.
    ///
    /// # Errors
    /// As [`Self::rows_before_advance`].
    pub(super) fn row_for_kept(
        &mut self,
        branch: &str,
        record: &BranchRefRecord,
        at: Version,
    ) -> Result<(), Failure> {
        if at.generation != record.created {
            return Ok(());
        }
        let kept = self.refs.kept(&record.name).to_vec();
        let Some(entry) = kept.iter().find(|entry| entry.seq == at.sequence) else {
            return Ok(());
        };
        self.fence_if_unsure()?;
        let generation = (branch.to_owned(), record.created);
        let mut known = match self.kept_rows.known.remove(&generation) {
            Some(known) => known,
            None => self.rows_in_window(branch, record, &kept, false)?,
        };
        if known.insert(entry.seq) {
            self.commit_rows(&[kept_head_row(branch, record, entry)])?;
        }
        self.kept_rows.known.insert(generation, known);
        Ok(())
    }

    /// STOR-003 `MD5c`: before `branch`'s record is discarded, with it the
    /// heads it keeps: the head's row when it has none, then the row of every
    /// kept version without one, all in one service commit, so no version of
    /// the ended generation is lost. Nothing is written when each has its row.
    /// A row found has the roots its version makes (`same_roots`), or the
    /// discard refuses.
    ///
    /// # Errors
    /// As [`Self::rows_before_advance`].
    pub(super) fn rows_before_discard(
        &mut self,
        branch: &str,
        record: &BranchRefRecord,
    ) -> Result<(), Failure> {
        let kept = self.refs.kept(&record.name).to_vec();
        self.fence_if_unsure()?;
        let generation = (branch.to_owned(), record.created);
        let known = match self.kept_rows.known.remove(&generation) {
            Some(known) => known,
            None => self.rows_in_window(branch, record, &kept, true)?,
        };
        let head = (!known.contains(&record.seq)).then(|| kept_row(branch, record));
        let rows: Vec<Row> = head
            .into_iter()
            .chain(
                kept.iter()
                    .filter(|entry| !known.contains(&entry.seq))
                    .map(|entry| kept_head_row(branch, record, entry)),
            )
            .collect();
        if !rows.is_empty() {
            self.commit_rows(&rows)?;
        }
        Ok(())
    }

    /// After a row commit whose outcome is unknown, the service record is
    /// fenced before any row read from it is relied on.
    /// The flag is cleared only once the fence holds: a fence that fails is
    /// still owed by the next call (STOR-003 `MD5c-r3`).
    fn fence_if_unsure(&mut self) -> Result<(), Failure> {
        if !self.kept_rows.unsure {
            return Ok(());
        }
        #[cfg(test)]
        fence_fails::reached()?;
        let service = self
            .refs
            .get(SERVICE)
            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
        let identity = (service.created, service.seq);
        self.refs
            .confirm_version(SERVICE, identity.0, identity.1)
            .map_err(|error| Failure::new("history_row_durability_unknown", error.to_string()))?;
        self.kept_rows.unsure = false;
        Ok(())
    }

    /// The versions of `record`'s generation whose rows are there: one range
    /// read of the service rows in [oldest kept, head), or through the head
    /// when `with_head` (STOR-003 `MD5c`), at most one key per version on each
    /// shard. A row there has the roots its version makes (`same_roots`), or
    /// the window refuses.
    fn rows_in_window(
        &self,
        branch: &str,
        record: &BranchRefRecord,
        kept: &[KeptHead],
        with_head: bool,
    ) -> Result<BTreeSet<u64>, Failure> {
        let mut found = BTreeSet::new();
        let oldest = kept.first().map_or(record.seq, |entry| entry.seq);
        let Some(limit) = NonZeroUsize::new(kept.len() + usize::from(with_head)) else {
            return Ok(found);
        };
        let end = if with_head {
            record
                .seq
                .checked_add(1)
                .ok_or_else(|| Failure::new("sequence_exhausted", "branch sequence exhausted"))?
        } else {
            record.seq
        };
        let service = self
            .refs
            .get(SERVICE)
            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
        let from = Key {
            branch: Hash::of(branch.as_bytes()),
            generation: record.created,
            sequence: oldest,
        };
        let to = Key {
            sequence: end,
            ..from
        }
        .encode();
        let from = from.encode();
        for shard in &service.shards {
            let page = self
                .reader
                .range_page_at(shard.shard_id, shard.head, &from, &to, limit)
                .map_err(Failure::observer)?;
            for (key, bytes) in page.entries {
                let row = Row::decode(&key, &bytes)
                    .map_err(|error| Failure::new("history_row_corrupt", error.to_string()))?;
                let made = if with_head && row.key.sequence == record.seq {
                    kept_row(branch, record)
                } else {
                    let Some(entry) = kept.iter().find(|entry| entry.seq == row.key.sequence)
                    else {
                        continue;
                    };
                    kept_head_row(branch, record, entry)
                };
                same_roots(&row, &made)?;
                found.insert(row.key.sequence);
            }
        }
        #[cfg(test)]
        WINDOW_READS.with(|reads| reads.set(reads.get() + 1));
        Ok(found)
    }

    /// `rows` in one durable service commit.
    fn commit_rows(&mut self, rows: &[Row]) -> Result<(), Failure> {
        self.commit_service_rows(rows)?;
        #[cfg(test)]
        ROW_COMMITS.with(|commits| commits.set(commits.get() + 1));
        self.publish_record(SERVICE)
    }

    /// The one service commit every history row is written by. A commit that
    /// fails, refused or installed with its durability unknown, leaves
    /// `unsure` set, so the next row read fences the service record before it
    /// relies on any row found there (STOR-003 `MD5c-r2`).
    fn commit_service_rows(&mut self, rows: &[Row]) -> Result<(), Failure> {
        let committed = self.try_commit_service_rows(rows);
        if committed.is_err() {
            self.kept_rows.unsure = true;
        }
        committed
    }

    fn try_commit_service_rows(&mut self, rows: &[Row]) -> Result<(), Failure> {
        let service = self
            .refs
            .get(SERVICE)
            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
        if service.kind != BranchKind::Namespace {
            return Err(Failure::new(
                "history_service_kind",
                "history service branch must be a namespace",
            ));
        }
        let identity = (service.created, service.seq);
        let branch =
            open_branch(SERVICE, &self.refs, &self.registry).map_err(Failure::branch_commit)?;
        for row in rows {
            let key = row.key.encode();
            let value = row
                .encode_value()
                .map_err(|error| Failure::new("history_row_corrupt", error.to_string()))?;
            branch
                .put(self.writer.shard_for(&key), key, value)
                .map_err(Failure::branch)?;
        }
        self.writer
            .commit(
                &branch,
                &self.registry,
                CommitRequest {
                    durability: CommitDurability::Durable {
                        refs: &mut self.refs,
                    },
                    extra_parents: &[],
                    timestamp: current_timestamp(),
                },
            )
            .map_err(|error| {
                let installed = self
                    .refs
                    .get(SERVICE)
                    .is_some_and(|now| (now.created, now.seq) != identity);
                Failure::new(
                    if installed {
                        "history_row_durability_unknown"
                    } else {
                        "history_row_write_refused"
                    },
                    error.to_string(),
                )
            })?;
        #[cfg(test)]
        row_install::reached()?;
        Ok(())
    }
}

impl Storage {
    pub(super) fn retain_history_row(
        &mut self,
        branch: &str,
        record: &BranchRefRecord,
    ) -> Result<(), Failure> {
        let row = kept_row(branch, record);
        let key = row.key.encode();
        let shard_id = self.writer.shard_for(&key);
        let service = self
            .refs
            .get(SERVICE)
            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
        if service.kind != BranchKind::Namespace {
            return Err(Failure::new(
                "history_service_kind",
                "history service branch must be a namespace",
            ));
        }
        let root = service
            .shards
            .iter()
            .find(|shard| shard.shard_id == shard_id)
            .ok_or_else(|| {
                Failure::new(
                    "incomplete_roots",
                    "service reference does not name the row's shard",
                )
            })?
            .head;
        let identity = (service.created, service.seq);
        if let Some(bytes) = self
            .reader
            .get_at(shard_id, root, &key)
            .map_err(Failure::observer)?
        {
            let existing = Row::decode(&key, &bytes)
                .map_err(|error| Failure::new("history_row_corrupt", error.to_string()))?;
            same_roots(&existing, &row)?;
            // An installed row may be the result of a failed reference entry fence.
            self.refs
                .confirm_version(SERVICE, identity.0, identity.1)
                .map_err(|error| {
                    Failure::new("history_row_durability_unknown", error.to_string())
                })?;
            self.learn_row(branch, record);
            return self.publish_record(SERVICE);
        }
        self.commit_service_rows(std::slice::from_ref(&row))?;
        #[cfg(test)]
        RETAIN_COMMITS.with(|commits| commits.set(commits.get() + 1));
        self.learn_row(branch, record);
        self.publish_record(SERVICE)
    }

    /// `record`'s version has its row: a loaded set of known rows learns it,
    /// so no spill writes it again (STOR-003 `MD5b`); a set not loaded yet finds
    /// it on its first read.
    fn learn_row(&mut self, branch: &str, record: &BranchRefRecord) {
        let generation = (branch.to_owned(), record.created);
        if let Some(known) = self.kept_rows.known.get_mut(&generation) {
            known.insert(record.seq);
        }
    }
}