haematite-cli 0.35.1

//! STOR-050: lines (docs/design/storage/LINES.md).
//!
//! R1: a stage's lineage is one row under the reserved prefix `s\0`, keyed by
//! the version it is written at, in the same commit as the stage's change,
//! its receipt and its author row, so the commit's one device flush makes all
//! of them durable together or none of them (STOR-034). A merge carries none
//! of its source's stage rows: its own commit holds one row naming the
//! source and its version and listing, in order, every stage row the source
//! committed since its fork anchor, so the lineage outlives the source.
//!
//! R2: a line is a branch whose fork provenance says so
//! (serve/storage/branches.rs). A release takes nothing of a live line
//! (serve/storage/release.rs), so it keeps every version it commits. Its
//! discard gives every version up: its history rows are released in the
//! discard, as a discarded generation's rows are under the ruling of 08:05
//! (no keep setting asked), and the reclaim pass that release asks for prunes
//! the nodes only the line reached. A version a live fork still holds is
//! released when that fork is discarded.
//!
//! A merged commit's list is cut into pieces of at most [`ROW_BYTES`] when
//! it is longer than one row of that size: the merged row then names its
//! source and how many pieces follow, and each piece is a row of its own in
//! the same commit, keyed by the merged version and the piece's index. So a
//! spring-back of a line of any length writes rows of a bounded size, and
//! history answers the whole list (lane 27).
//!
//! A discard that has line versions to release writes one row in its own
//! commit saying so, under `line-release\0` in the service namespace; the
//! release's last round deletes it in that round's own commit. Opening the
//! store reads that prefix once, at most `UNFINISHED_PAGE` rows a shard,
//! and finishes every release it names, so a stop between the discard and
//! its last round never keeps the line's rows, nor the nodes they reach,
//! beyond the next open (Tom's pruning rule, 3 October 2026; lane 27).

use std::collections::BTreeMap;
use std::num::NonZeroUsize;

use haematite::branch::current_timestamp;
use haematite::db::ReadOnlyDatabase;
use haematite::tree::Hash;
use serde::{Deserialize, Serialize};

use super::branches::{self, ForkProvenance};
use super::mutations::PreparedWrite;
use super::release::{Answer, Rules};
use super::{
    BranchRefRecord, Captured, Failure, SERVICE, Storage, branch_name, dead, history, identifier,
    reads,
};
use crate::serve::protocol::{Change, Stage, Version};

/// The reserved prefix of stage rows, beside the receipts (`r\0`) and the
/// author rows (`a\0`).
pub const TAG: &[u8] = b"s\0";

const FORMAT: u32 = 1;

/// The most history rows one release of a discarded line names: a line of
/// more versions is released in more than one, each its own three durable
/// operations (serve/storage/release.rs).
const RELEASE_ROWS: usize = 4_096;

/// The most bytes a merged commit's stage row holds with its list inline,
/// and the most a piece of a longer list holds.
///
/// It is the bound the store already puts on a reserved row of its own (an
/// author row, serve/contracts/attribution.rs). A piece holds at least one listed stage,
/// so a single stage row longer than this (one the stage's own request
/// wrote) makes a piece of its own size.
pub const ROW_BYTES: usize = 65_536;

/// The prefix of the rows that say a discard's line release is not
/// finished, in the service namespace.
const UNFINISHED: &[u8] = b"line-release\0";

/// The most unfinished releases one read of each shard names at open.
const UNFINISHED_PAGE: usize = 64;

/// A branch and one of its versions.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct At {
    pub branch: String,
    pub version: Version,
}

/// One stage row a merge's source committed: where, and the stage.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Listed {
    pub branch: String,
    pub version: Version,
    pub stage: Stage,
}

/// What a merged commit's stage row says: the merge's source at the version
/// merged, and every stage row it committed since its fork anchor, in order.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Merged {
    pub from: At,
    pub stages: Vec<Listed>,
}

/// The row as stored: exactly one of `stage` and `merged`. A merged row
/// whose list is in pieces names how many, and lists none itself.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Row {
    format: u32,
    version: Version,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    stage: Option<Stage>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    merged: Option<Merged>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pieces: Option<u32>,
}

/// One piece of a merged commit's list, in order.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Piece {
    format: u32,
    version: Version,
    index: u32,
    stages: Vec<Listed>,
}

/// The key and value of a discard's unfinished row.
pub(super) type UnfinishedRow = (Vec<u8>, Vec<u8>);

/// A discard's line release not yet finished: the discarded name and its
/// generation.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Unfinished {
    branch: String,
    generation: u64,
}

/// What a version's stage row says, as history answers it.
#[derive(Debug, Default, PartialEq, Eq)]
pub struct Lineage {
    pub stage: Option<Stage>,
    pub merged: Option<Merged>,
}

/// The key of the stage row written at `version`: the tag, then the
/// generation and sequence big-endian, so rows sort in version order.
#[must_use]
pub fn key(version: Version) -> Vec<u8> {
    let mut key = Vec::with_capacity(TAG.len() + 16);
    key.extend_from_slice(TAG);
    key.extend_from_slice(&version.generation.to_be_bytes());
    key.extend_from_slice(&version.sequence.to_be_bytes());
    key
}

/// The key of piece `index` of the merged list written at `version`: the
/// merged row's key, then the index big-endian, so the pieces sort in order
/// just after their row.
#[must_use]
pub fn piece_key(version: Version, index: u32) -> Vec<u8> {
    let mut key = key(version);
    key.extend_from_slice(&index.to_be_bytes());
    key
}

/// The version a stage row's key names, and the piece index when it is a
/// piece's.
fn key_parts(key: &[u8]) -> Result<(Version, Option<u32>), Failure> {
    let corrupt = || Failure::new("stage_row_corrupt", "stage row key is not a version");
    let rest = key.strip_prefix(TAG).ok_or_else(corrupt)?;
    let (generation, rest) = rest.split_at_checked(8).ok_or_else(corrupt)?;
    let (sequence, piece) = rest.split_at_checked(8).ok_or_else(corrupt)?;
    let version = Version {
        generation: u64::from_be_bytes(generation.try_into().map_err(|_| corrupt())?),
        sequence: u64::from_be_bytes(sequence.try_into().map_err(|_| corrupt())?),
    };
    match piece.len() {
        0 => Ok((version, None)),
        4 => Ok((
            version,
            Some(u32::from_be_bytes(piece.try_into().map_err(|_| corrupt())?)),
        )),
        _ => Err(corrupt()),
    }
}

fn named(value: &str, path: String) -> Result<(), Failure> {
    identifier(value).map_err(|mut failure| {
        failure.message = format!("{path}: {}", failure.message);
        failure.path = Some(path);
        failure
    })
}

/// A stage beside `change`, checked before anything is read or written:
/// beside branch control it is refused `invalid_operation`, and a member
/// that is no name is refused `invalid_identifier` at its path.
///
/// # Errors
/// As above.
pub(super) fn validate(stage: &Stage, change: &Change) -> Result<(), Failure> {
    if matches!(
        change,
        Change::Fork(_) | Change::Discard(_) | Change::Merge(_)
    ) {
        return Err(Failure {
            code: "invalid_operation".to_owned(),
            message: "a stage is written by a data change: branch.fork, branch.discard and \
                      branch.merge take none"
                .to_owned(),
            path: Some("/stage".to_owned()),
            cause: None,
        });
    }
    named(&stage.line, "/stage/line".to_owned())?;
    named(&stage.run, "/stage/run".to_owned())?;
    named(&stage.stage, "/stage/stage".to_owned())?;
    named(&stage.subject, "/stage/subject".to_owned())?;
    for (index, input) in stage.inputs.iter().enumerate() {
        named(&input.branch, format!("/stage/inputs/{index}/branch"))?;
    }
    Ok(())
}

fn encode(row: &Row) -> Result<Vec<u8>, Failure> {
    serde_json::to_vec(row).map_err(Failure::storage)
}

/// The stage row of a change written at `next`.
///
/// # Errors
/// Serialization failures.
pub(super) fn stage_write(next: Version, stage: &Stage) -> Result<PreparedWrite, Failure> {
    let row = Row {
        format: FORMAT,
        version: next,
        stage: Some(stage.clone()),
        merged: None,
        pieces: None,
    };
    Ok((key(next), Some(encode(&row)?)))
}

fn format_known(format: u32) -> Result<(), Failure> {
    if format != FORMAT {
        return Err(Failure::new(
            "stage_row_format_unsupported",
            format!("stage row format {format} is not {FORMAT}"),
        ));
    }
    Ok(())
}

fn decode(version: Version, bytes: &[u8]) -> Result<Row, Failure> {
    let row: Row = serde_json::from_slice(bytes)
        .map_err(|error| Failure::new("stage_row_corrupt", error.to_string()))?;
    format_known(row.format)?;
    let misplaced = row.pieces.is_some()
        && !row
            .merged
            .as_ref()
            .is_some_and(|merged| merged.stages.is_empty());
    if row.version != version || row.stage.is_some() == row.merged.is_some() || misplaced {
        return Err(Failure::new(
            "stage_row_corrupt",
            "a stage row names its own version and holds a stage or a merge, not both, and \
             only a merge whose list is in pieces names them",
        ));
    }
    Ok(row)
}

fn decode_piece(version: Version, index: u32, bytes: &[u8]) -> Result<Vec<Listed>, Failure> {
    let piece: Piece = serde_json::from_slice(bytes)
        .map_err(|error| Failure::new("stage_row_corrupt", error.to_string()))?;
    format_known(piece.format)?;
    if piece.version != version || piece.index != index {
        return Err(Failure::new(
            "stage_row_corrupt",
            "a piece of a merged list names its own version and index",
        ));
    }
    Ok(piece.stages)
}

fn missing_piece(version: Version, index: u32) -> Failure {
    Failure::new(
        "stage_row_corrupt",
        format!(
            "the merged list at {}.{} has no piece {index}",
            version.generation, version.sequence
        ),
    )
}

/// The rows a merged commit at `next` writes for `merged`: one row holding
/// the list when that row is at most [`ROW_BYTES`], as a short list always
/// was; else a row naming the source and its pieces, and the pieces, each
/// at most [`ROW_BYTES`] unless one listed stage alone is longer.
fn merged_rows(next: Version, merged: Merged) -> Result<Vec<PreparedWrite>, Failure> {
    let mut whole = Row {
        format: FORMAT,
        version: next,
        stage: None,
        merged: Some(merged),
        pieces: None,
    };
    let bytes = encode(&whole)?;
    if bytes.len() <= ROW_BYTES {
        return Ok(vec![(key(next), Some(bytes))]);
    }
    let stages = whole
        .merged
        .as_mut()
        .map(|merged| std::mem::take(&mut merged.stages))
        .unwrap_or_default();
    // The longest a piece's own members can be, its index at its widest.
    let envelope = encode_piece(next, u32::MAX, Vec::new())?.len();
    let mut pieces: Vec<Vec<Listed>> = Vec::new();
    let mut listed: Vec<Listed> = Vec::new();
    let mut size = 0;
    for stage in stages {
        let one = serde_json::to_vec(&stage).map_err(Failure::storage)?.len();
        // A comma before every listed stage but the first.
        let grown = size + one + usize::from(!listed.is_empty());
        if !listed.is_empty() && envelope + grown > ROW_BYTES {
            pieces.push(std::mem::take(&mut listed));
            size = one;
        } else {
            size = grown;
        }
        listed.push(stage);
    }
    if !listed.is_empty() {
        pieces.push(listed);
    }
    whole.pieces = Some(u32::try_from(pieces.len()).map_err(Failure::storage)?);
    let mut writes = vec![(key(next), Some(encode(&whole)?))];
    for (index, stages) in (0_u32..).zip(pieces) {
        writes.push((
            piece_key(next, index),
            Some(encode_piece(next, index, stages)?),
        ));
    }
    Ok(writes)
}

fn encode_piece(version: Version, piece: u32, stages: Vec<Listed>) -> Result<Vec<u8>, Failure> {
    serde_json::to_vec(&Piece {
        format: FORMAT,
        version,
        index: piece,
        stages,
    })
    .map_err(Failure::storage)
}

/// What one stage row of a merge's source lists: its stage, or a merge it
/// took in, whose list is inline or in pieces found beside it.
#[derive(Debug)]
enum Entry {
    Listed(Vec<Listed>),
    Pieces(u32),
}

/// The stage rows a merge's source committed since its fork anchor, as the
/// merge's change pages found them, on any shard and in any order, with the
/// pieces of every merged list among them.
#[derive(Debug, Default)]
pub(super) struct Found {
    rows: Vec<(Version, Entry)>,
    pieces: BTreeMap<(u64, u64, u32), Vec<Listed>>,
}

impl Found {
    /// One stage row of `source`, or one piece of a merged list, found
    /// among its changes.
    ///
    /// # Errors
    /// A row that does not decode, by name.
    pub(super) fn add(&mut self, source: &str, key: &[u8], bytes: &[u8]) -> Result<(), Failure> {
        let (version, piece) = key_parts(key)?;
        if let Some(index) = piece {
            let stages = decode_piece(version, index, bytes)?;
            self.pieces
                .insert((version.generation, version.sequence, index), stages);
            return Ok(());
        }
        let row = decode(version, bytes)?;
        let entry = match (row.stage, row.merged, row.pieces) {
            (Some(stage), _, _) => Entry::Listed(vec![Listed {
                branch: source.to_owned(),
                version: row.version,
                stage,
            }]),
            (None, Some(_merged), Some(pieces)) => Entry::Pieces(pieces),
            // A merge the source took in lists its own source's stages: they
            // are listed here in its place, so nothing is nested.
            (None, Some(merged), None) => Entry::Listed(merged.stages),
            (None, None, _) => Entry::Listed(Vec::new()),
        };
        self.rows.push((row.version, entry));
        Ok(())
    }

    const fn is_empty(&self) -> bool {
        self.rows.is_empty()
    }

    /// The merged commit's stage rows, written at `next`: `from` at
    /// `version`, its stages in the order it committed them, in one row or
    /// in pieces (see `merged_rows`).
    ///
    /// # Errors
    /// `stage_row_corrupt` for a merged list missing a piece; serialization
    /// failures.
    fn write(
        mut self,
        next: Version,
        from: &str,
        version: Version,
    ) -> Result<Vec<PreparedWrite>, Failure> {
        self.rows
            .sort_unstable_by_key(|(at, _)| (at.generation, at.sequence));
        let mut stages = Vec::new();
        for (at, entry) in self.rows {
            match entry {
                Entry::Listed(listed) => stages.extend(listed),
                Entry::Pieces(count) => {
                    for index in 0..count {
                        let piece = self
                            .pieces
                            .remove(&(at.generation, at.sequence, index))
                            .ok_or_else(|| missing_piece(at, index))?;
                        stages.extend(piece);
                    }
                }
            }
        }
        let merged = Merged {
            from: At {
                branch: from.to_owned(),
                version,
            },
            stages,
        };
        merged_rows(next, merged)
    }
}

/// The stage rows a merged commit at `next` carries, when it carries any:
/// when the source committed a stage row since its fork, or is a line.
///
/// # Errors
/// Reading the source's provenance; `stage_row_corrupt`; serialization
/// failures.
pub(super) fn merged_write(
    reader: &ReadOnlyDatabase,
    source: &Captured,
    found: Found,
    next: Version,
) -> Result<Vec<PreparedWrite>, Failure> {
    if found.is_empty() && !is_line(reader, source)? {
        return Ok(Vec::new());
    }
    found.write(next, &source.branch, source.version)
}

/// What the stage row written at `at`'s own version says: `Lineage::default()`
/// for a version no stage and no merge of stages wrote.
///
/// One point read, and one more for each piece of a merged list in pieces.
///
/// # Errors
/// A row that does not decode, or a piece missing, by name; storage failures.
pub fn read(reader: &ReadOnlyDatabase, at: &Captured) -> Result<Lineage, Failure> {
    let key = key(at.version);
    let Some(bytes) = reads::get_bytes(reader, at, &key)? else {
        return Ok(Lineage::default());
    };
    let row = decode(at.version, &bytes)?;
    let mut merged = row.merged;
    if let (Some(merged), Some(count)) = (merged.as_mut(), row.pieces) {
        for index in 0..count {
            let piece = reads::get_bytes(reader, at, &piece_key(at.version, index))?
                .ok_or_else(|| missing_piece(at.version, index))?;
            merged
                .stages
                .extend(decode_piece(at.version, index, &piece)?);
        }
    }
    Ok(Lineage {
        stage: row.stage,
        merged,
    })
}

/// Whether the branch at `at` is a line: its fork provenance says so.
///
/// # Errors
/// As [`branches::provenance`].
pub(super) fn is_line(reader: &ReadOnlyDatabase, at: &Captured) -> Result<bool, Failure> {
    Ok(branches::origin(reader, at)?.is_some_and(|(_origin, line)| line))
}

/// The key of the row saying `branch`'s discarded generation `generation`
/// has line versions not yet released: the prefix, the name's hash, then
/// the generation big-endian.
fn unfinished_key(branch: &str, generation: u64) -> Vec<u8> {
    let mut key = Vec::with_capacity(UNFINISHED.len() + 32 + 8);
    key.extend_from_slice(UNFINISHED);
    key.extend_from_slice(Hash::of(branch.as_bytes()).as_bytes());
    key.extend_from_slice(&generation.to_be_bytes());
    key
}

/// The first key past every unfinished row.
fn unfinished_end() -> Vec<u8> {
    let mut end = UNFINISHED.to_vec();
    if let Some(last) = end.last_mut() {
        *last = last.saturating_add(1);
    }
    end
}

/// At most `UNFINISHED_PAGE` unfinished rows from each shard of `service`,
/// oldest key first: one range read a shard. Counted for the tests.
fn unfinished_page(
    reader: &ReadOnlyDatabase,
    service: &BranchRefRecord,
) -> Result<Vec<(Vec<u8>, Unfinished)>, Failure> {
    let limit = NonZeroUsize::new(UNFINISHED_PAGE).unwrap_or(NonZeroUsize::MIN);
    let end = unfinished_end();
    let mut found = Vec::new();
    for shard in &service.shards {
        let read = reader
            .range_page_at(shard.shard_id, shard.head, UNFINISHED, &end, limit)
            .map_err(Failure::observer)?;
        #[cfg(test)]
        unfinished_reads::count(read.entries.len());
        for (key, bytes) in read.entries {
            let row: Unfinished = serde_json::from_slice(&bytes)
                .map_err(|error| Failure::new("line_release_row_corrupt", error.to_string()))?;
            if unfinished_key(&row.branch, row.generation) != key {
                return Err(Failure::new(
                    "line_release_row_corrupt",
                    "an unfinished line release names another key",
                ));
            }
            found.push((key, row));
        }
    }
    found.sort_unstable_by(|(left, _), (right, _)| left.cmp(right));
    Ok(found)
}

impl Storage {
    /// The row a discard of `branch`'s generation `generation` writes in its
    /// own commit when it has line versions to release, read before
    /// anything is written: when the branch is a line, or was forked from a
    /// discarded line it still held. `origin` is the branch's fork
    /// provenance, as its head reads it. `None` for an ordinary discard
    /// whose origin is live or no line, which writes as before.
    ///
    /// # Errors
    /// Reading the origin's marker and last row; serialization failures.
    pub(super) fn unfinished_write(
        &self,
        branch: &str,
        generation: u64,
        origin: Option<&(ForkProvenance, bool)>,
    ) -> Result<Option<UnfinishedRow>, Failure> {
        let Some((origin, line)) = origin else {
            return Ok(None);
        };
        if !line && !self.origin_is_discarded_line(origin)? {
            return Ok(None);
        }
        let row = Unfinished {
            branch: branch.to_owned(),
            generation,
        };
        let bytes = serde_json::to_vec(&row).map_err(Failure::storage)?;
        Ok(Some((unfinished_key(branch, generation), bytes)))
    }

    /// Whether `origin` names a discarded generation of a line, its name
    /// with no record: what a discard of a fork of it releases.
    fn origin_is_discarded_line(&self, origin: &ForkProvenance) -> Result<bool, Failure> {
        Ok(self.refs.get(&branch_name(&origin.branch)).is_none()
            && matches!(
                self.discarded_origin(&origin.branch, origin.version.generation)?,
                Some((_, true))
            ))
    }

    /// After the discard of `branch`'s generation `generation` is durable:
    /// first the generation it was forked from, when that name is discarded
    /// and was a line, since this branch was what still held it; then this
    /// generation itself, when it is a line. Each is read from its discard
    /// marker and the history row of its last version, so a replay of the
    /// discard, or the next open, finds what is left and finishes it, and
    /// finds nothing once it is done. The discard's unfinished row goes in
    /// the commit of the round that releases its generation's last row, or
    /// in a commit of its own when no round did (what is left is held, or
    /// was released before). An ordinary discard whose origin is live reads
    /// four entries and writes nothing.
    ///
    /// # Errors
    /// `line_release_unfinished`, naming what stopped it: the discard itself
    /// stands, and a replay of it, or the next open, releases the rest.
    pub(super) fn release_after_discard(
        &mut self,
        branch: &str,
        generation: u64,
    ) -> Result<(), Failure> {
        self.finish_release(branch, generation).map_err(|failure| {
            Failure::new(
                "line_release_unfinished",
                format!(
                    "the discard of {branch} is durable; releasing its line versions stopped: \
                     {failure}"
                ),
            )
        })
    }

    fn finish_release(&mut self, branch: &str, generation: u64) -> Result<(), Failure> {
        let unfinished = unfinished_key(branch, generation);
        self.release_lines(branch, generation, &unfinished)?;
        let service = self
            .refs
            .get(SERVICE)
            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
        let shard_id = self.reader.shard_for(&unfinished);
        let root = service
            .shards
            .iter()
            .find(|shard| shard.shard_id == shard_id)
            .ok_or_else(|| Failure::new("incomplete_roots", "service row shard is missing"))?
            .head;
        let left = self
            .reader
            .get_at(shard_id, root, &unfinished)
            .map_err(Failure::observer)?;
        if left.is_some() {
            self.commit_service(&[(unfinished, None)])?;
        }
        Ok(())
    }

    /// At open, after the discards are reconciled: every line release a
    /// discard left unfinished, found by one read of the unfinished rows,
    /// at most `UNFINISHED_PAGE` a shard, and read again until none is
    /// left; each is finished as a replay of its discard would finish it,
    /// and a reclaim pass is then asked. A store with none reads that one
    /// page and writes nothing.
    ///
    /// # Errors
    /// `line_release_unfinished`, naming the discard and what stopped it.
    pub(super) fn finish_line_releases(&mut self) -> Result<(), Failure> {
        let mut finished = false;
        loop {
            let service = self
                .refs
                .get(SERVICE)
                .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
            let page = unfinished_page(&self.reader, service)?;
            if page.is_empty() {
                break;
            }
            for (_key, row) in page {
                self.release_after_discard(&row.branch, row.generation)?;
                finished = true;
            }
        }
        if finished {
            // Every round asked a pass after its commit; this one is for a
            // stop after a last round's commit and before its ask. A
            // refusal is said by name, as a discard's is.
            self.writer.request_reclaim().map_err(|error| {
                Failure::new(
                    "line_release_unfinished",
                    format!("the line releases are finished; asking a reclaim pass: {error}"),
                )
            })?;
        }
        Ok(())
    }

    fn release_lines(
        &mut self,
        branch: &str,
        generation: u64,
        unfinished: &[u8],
    ) -> Result<(), Failure> {
        let Some((origin, line)) = self.discarded_origin(branch, generation)? else {
            return Ok(());
        };
        let parent = origin.version.generation;
        if self.origin_is_discarded_line(&origin)? {
            // The unfinished row goes with the origin's last row only when
            // this generation has none of its own to release.
            self.release_line(&origin.branch, parent, (!line).then_some(unfinished))?;
        }
        if line {
            self.release_line(branch, generation, Some(unfinished))?;
        }
        Ok(())
    }

    /// The fork provenance of `branch`'s discarded generation `generation`,
    /// read at its last version's history row; `None` when its marker is
    /// gone (every row released) or it was never forked.
    fn discarded_origin(
        &self,
        branch: &str,
        generation: u64,
    ) -> Result<Option<(ForkProvenance, bool)>, Failure> {
        let service = self
            .refs
            .get(SERVICE)
            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
        let Some(marker) = dead::marker(&self.reader, service, branch, generation)? else {
            return Ok(None);
        };
        let Some(last) = history::capture_kept(&self.reader, service, branch, marker.last)? else {
            return Ok(None);
        };
        branches::origin(&self.reader, &last)
    }

    /// Every history row of `branch`'s discarded line generation `generation`
    /// that nothing else holds, released with no keep setting asked: a
    /// release at a time until none is left or what is left is held. The
    /// round that releases the generation's last row deletes `unfinished`
    /// in its own commit. A round is asked again until one takes nothing,
    /// so a release a stop left decided is rolled forward and the rest
    /// released after it.
    fn release_line(
        &mut self,
        branch: &str,
        generation: u64,
        unfinished: Option<&[u8]>,
    ) -> Result<(), Failure> {
        // A name made again since is released by its own rules, never here.
        if self.refs.get(&branch_name(branch)).is_some() {
            return Ok(());
        }
        let max_rows = NonZeroUsize::new(RELEASE_ROWS).unwrap_or(NonZeroUsize::MIN);
        loop {
            let service = self
                .refs
                .get(SERVICE)
                .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
            if dead::marker(&self.reader, service, branch, generation)?.is_none() {
                return Ok(());
            }
            let rules = Rules {
                keep: None,
                max_rows,
                now: current_timestamp(),
            };
            let answer: Answer = self.release_generation(branch, generation, &rules, unfinished)?;
            if answer.released.is_none() {
                return Ok(());
            }
        }
    }
}

/// How many unfinished rows the reads at open found, and how many reads
/// were made, on this thread (lane 27's clean-open count).
#[cfg(test)]
pub mod unfinished_reads {
    use std::cell::Cell;

    thread_local! {
        static READS: Cell<(usize, usize)> = const { Cell::new((0, 0)) };
    }

    pub(super) fn count(rows: usize) {
        READS.with(|reads| {
            let (made, found) = reads.get();
            reads.set((made + 1, found + rows));
        });
    }

    /// The reads made and the rows found since the last take, and starts
    /// both again at none.
    pub fn take() -> (usize, usize) {
        READS.with(|reads| reads.replace((0, 0)))
    }
}

/// A test-only crash between a change's staged writes and its commit, as a
/// kill would leave it: nothing is installed.
#[cfg(test)]
pub mod before_commit {
    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 writes were staged and before the commit",
            ));
        }
        Ok(())
    }
}

#[cfg(test)]
#[path = "lineage_pieces_tests.rs"]
mod pieces_tests;