haematite-cli 0.21.8

//! STOR-003 HR1 and V4: the versions of a branch something relies on, from
//! each of the four sources the design names and the embedding job's
//! position, with the pending discards stood in for.

use std::error::Error;
use std::num::{NonZeroU64, NonZeroUsize};
use std::sync::{Arc, Mutex};

use haematite::SyncCounts;
use haematite::branch::history::{Key, Row};
use haematite::branch::{ForkWithChanges, KeptLimit, current_timestamp, open as open_branch};
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use haematite::tree::Hash;

use super::{
    Answer, Below, Crash, DiscardsUnread, Floor, HeldBy, Keep, LOWEST, NANOS_A_DAY,
    PendingDiscards, Rules, Source, UndatedGeneration, crash_at, judge,
};
use crate::serve::protocol::{
    BranchDiscard, Change, CollectionWrite, Fork, Metric, Mutation, Selection, Version,
};
use crate::serve::releaser::{Releaser, Settings};
use crate::serve::storage::dead;
use crate::serve::storage::embed_job::{Page, Reach, field_key};
use crate::serve::storage::embedded::{self, Declaration, ModelName, Position, Toward, VectorKind};
use crate::serve::storage::history::{MARK_READS, MARKER_READS, ROW_COMMITS, ROW_READS};
use crate::serve::storage::{
    Captured, Catalogue, Failure, MergePin, SERVICE, Storage, branch_name, history,
};

type TestResult = Result<(), Box<dyn Error>>;

/// The pending discards as a test sets them.
struct Pending(Vec<(String, Version)>);

impl PendingDiscards for Pending {
    fn pending(&self, _storage: &Storage) -> Result<Vec<(String, Version)>, Failure> {
        Ok(self.0.clone())
    }
}

fn none() -> Pending {
    Pending(Vec::new())
}

fn create() -> Result<(tempfile::TempDir, Storage), Box<dyn Error>> {
    let directory = tempfile::tempdir()?;
    let storage = Storage::create(
        DatabaseConfig {
            data_dir: directory.path().join("database"),
            shard_count: 2,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::reclaim(0),
        },
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    Ok((directory, storage))
}

fn head(storage: &Storage, branch: &str) -> Result<Version, Failure> {
    Ok(storage
        .capture(&Selection {
            branch: branch.to_owned(),
            version: None,
        })?
        .version)
}

/// `branch` moved on by one write; its new version.
fn advance(storage: &mut Storage, branch: &str, tag: &str) -> Result<Version, Failure> {
    let mutation = Mutation {
        request_id: format!("{branch}-{tag}"),
        branch: branch.to_owned(),
        expected: head(storage, branch)?,
        change: Change::Collection(CollectionWrite {
            collection: format!("c-{tag}"),
            schema: None,
        }),
    };
    Ok(storage.mutate(&mutation)?.version)
}

/// `name` forked from `from` at its head; the version forked from.
fn fork(storage: &mut Storage, from: &str, name: &str) -> Result<Version, Failure> {
    let at = head(storage, from)?;
    storage.mutate(&Mutation {
        request_id: format!("fork-{name}"),
        branch: from.to_owned(),
        expected: at,
        change: Change::Fork(Fork {
            name: name.to_owned(),
            from: at,
        }),
    })?;
    Ok(at)
}

fn held(source: Source, name: &str, version: Version) -> HeldBy {
    HeldBy {
        source,
        name: name.to_owned(),
        version,
    }
}

/// With nothing else, only the head is relied on, and no branch is read.
#[test]
fn the_head_alone_is_relied_on_with_nothing_else() -> TestResult {
    let (_directory, mut storage) = create()?;
    let now = advance(&mut storage, "main", "a")?;
    let floor = storage.relied_on("main", &none())?;
    assert_eq!(floor.held_by, [held(Source::Head, "main", now)]);
    assert_eq!(floor.branches_read, 0);
    Ok(())
}

/// Fails while a fork's origin is not read: a branch forked from `main`
/// relies on the version it was taken from, older than `main`'s head, and
/// that is the floor. The one other branch is read once.
#[test]
fn a_fork_relies_on_the_version_it_was_taken_from() -> TestResult {
    let (_directory, mut storage) = create()?;
    advance(&mut storage, "main", "a")?;
    let taken = fork(&mut storage, "main", "feature")?;
    advance(&mut storage, "main", "b")?;
    let now = advance(&mut storage, "main", "c")?;
    let floor = storage.relied_on("main", &none())?;
    assert_eq!(
        floor.held_by,
        [
            held(Source::Fork, "feature", taken),
            held(Source::Head, "main", now),
        ]
    );
    assert_eq!(floor.oldest(), Some(&held(Source::Fork, "feature", taken)));
    assert_eq!(floor.branches_read, 1);
    Ok(())
}

/// Fails while an open merge's moments are not read, or an abandoned one's
/// still are: open, its origin and target on `main` are relied on; once
/// abandoned, they are not.
#[test]
fn an_open_merge_relies_on_its_moments_and_an_abandoned_one_on_none() -> TestResult {
    let (_directory, mut storage) = create()?;
    let taken = fork(&mut storage, "main", "feature")?;
    let source = MergePin {
        branch: "feature".to_owned(),
        version: advance(&mut storage, "feature", "f")?,
    };
    let target = MergePin {
        branch: "main".to_owned(),
        version: head(&storage, "main")?,
    };
    storage.merge_open("review", &source, &target)?;
    let now = advance(&mut storage, "main", "a")?;
    let floor = storage.relied_on("main", &none())?;
    let mut merges: Vec<&HeldBy> = floor
        .held_by
        .iter()
        .filter(|held| held.source == Source::Merge)
        .collect();
    merges.dedup();
    assert_eq!(merges, [&held(Source::Merge, "review", taken)]);
    assert_eq!(floor.held_by.len(), 4, "{:?}", floor.held_by);
    storage.merge_abandon("review")?;
    let floor = storage.relied_on("main", &none())?;
    assert_eq!(
        floor.held_by,
        [
            held(Source::Fork, "feature", taken),
            held(Source::Head, "main", now),
        ]
    );
    Ok(())
}

/// Fails while the pending discards are not asked: an unfinished discard of
/// this branch relies on its version; one of another branch does not.
#[test]
fn an_unfinished_discard_relies_on_its_version() -> TestResult {
    let (_directory, mut storage) = create()?;
    let asked = advance(&mut storage, "main", "a")?;
    let now = advance(&mut storage, "main", "b")?;
    let pending = Pending(vec![
        ("main".to_owned(), asked),
        ("elsewhere".to_owned(), asked),
    ]);
    let floor = storage.relied_on("main", &pending)?;
    assert_eq!(
        floor.held_by,
        [
            held(Source::Discard, "main", asked),
            held(Source::Head, "main", now),
        ]
    );
    Ok(())
}

/// Fails while a floor is given without the pending discards: until they can
/// be read, the floor is refused by name.
#[test]
fn no_floor_is_given_while_the_pending_discards_cannot_be_read() -> TestResult {
    let (_directory, storage) = create()?;
    let refused = storage
        .relied_on("main", &DiscardsUnread)
        .err()
        .ok_or("a floor was given without the pending discards")?;
    assert_eq!(
        refused.code, "release_discards_unread",
        "{}",
        refused.message
    );
    Ok(())
}

/// Fails while the floor needs the branch's own ref: with `feature`'s ref
/// gone, as a discard being finished leaves it, the floor still names the
/// fork taken from it and the open merge pinning it, at the version they
/// rely on, and no head.
#[test]
fn a_floor_for_a_branch_with_no_ref_names_its_forks_and_merges_and_no_head() -> TestResult {
    let (_directory, mut storage) = create()?;
    fork(&mut storage, "main", "feature")?;
    let taken = fork(&mut storage, "feature", "child")?;
    let source = MergePin {
        branch: "child".to_owned(),
        version: advance(&mut storage, "child", "c")?,
    };
    let target = MergePin {
        branch: "feature".to_owned(),
        version: head(&storage, "feature")?,
    };
    storage.merge_open("review", &source, &target)?;
    storage.refs.remove(&branch_name("feature"))?;
    let floor = storage.relied_on_by_others("feature", false, &none())?;
    let mut sources: Vec<Source> = floor.held_by.iter().map(|held| held.source).collect();
    sources.dedup();
    assert_eq!(
        sources,
        [Source::Fork, Source::Merge],
        "{:?}",
        floor.held_by
    );
    assert!(floor.held_by.iter().all(|held| held.version == taken));
    assert_eq!(floor.oldest().map(|held| held.version), Some(taken));
    Ok(())
}

/// Fails while a branch's own pending discard is counted against its own
/// finishing: skipped, it is absent; asked without the skip, it is there;
/// another branch's pending discard is untouched by the skip.
#[test]
fn its_own_pending_discard_is_skipped_and_another_branchs_is_not() -> TestResult {
    let (_directory, mut storage) = create()?;
    let asked = advance(&mut storage, "main", "a")?;
    let pending = Pending(vec![
        ("feature".to_owned(), asked),
        ("elsewhere".to_owned(), asked),
    ]);
    let skipped = storage.relied_on_by_others("feature", true, &pending)?;
    assert_eq!(skipped.held_by, []);
    let counted = storage.relied_on_by_others("feature", false, &pending)?;
    assert_eq!(counted.held_by, [held(Source::Discard, "feature", asked)]);
    let other = storage.relied_on_by_others("elsewhere", false, &pending)?;
    assert_eq!(other.held_by, [held(Source::Discard, "elsewhere", asked)]);
    Ok(())
}

/// Fails while `relied_on` answers for a branch that does not exist: it
/// still refuses by name.
#[test]
fn relied_on_still_refuses_a_branch_with_no_ref() -> TestResult {
    let (_directory, storage) = create()?;
    let refused = storage
        .relied_on("gone", &none())
        .err()
        .ok_or("a floor was given for a missing branch")?;
    assert_eq!(refused.code, "branch_missing", "{}", refused.message);
    Ok(())
}

/// An empty floor means nothing relies on any version: `oldest` is `None`,
/// and every version may go.
#[test]
fn a_floor_nothing_relies_on_is_empty() -> TestResult {
    let (_directory, storage) = create()?;
    let floor = storage.relied_on_by_others("nobody", false, &none())?;
    assert_eq!(floor.held_by, []);
    assert_eq!(floor.oldest(), None);
    assert_eq!(floor.branches_read, 1);
    Ok(())
}

/// `notes.body_vec` declared embedded on main, written directly as V6's verb
/// will write it.
fn declare(storage: &mut Storage) -> Result<(), Box<dyn Error>> {
    let declaration = Declaration {
        kind: VectorKind::Dense,
        source: "body".to_owned(),
        model: ModelName {
            name: "stand-in".to_owned(),
            revision: "1".to_owned(),
            manifest_sha256: "abc".to_owned(),
        },
        dimensions: NonZeroUsize::new(2).ok_or("two is not zero")?,
        distance: Metric::Cosine,
    };
    let bytes = embedded::encode_declaration(&declaration)?;
    let key = field_key("notes", "body_vec");
    storage.commit_namespace(&branch_name("main"), &[(key, Some(bytes))])?;
    Ok(())
}

/// An embedding page with no records, moving the job's position.
fn embed(storage: &mut Storage, toward: Version, reach: Reach) -> Result<(), Box<dyn Error>> {
    let page = Page {
        collection: "notes".to_owned(),
        field: "body_vec".to_owned(),
        model: ModelName {
            name: "stand-in".to_owned(),
            revision: "1".to_owned(),
            manifest_sha256: "abc".to_owned(),
        },
        made: Vec::new(),
        removed: Vec::new(),
        cleared: Vec::new(),
        windowed: Vec::new(),
        toward,
        reach,
    };
    storage.embed_commit("main", &page, NonZeroUsize::MIN)?;
    Ok(())
}

/// Fails while the embedding job's position is not read: the version its
/// finished walk reached is relied on, so its next walk can read from it.
#[test]
fn the_embedding_jobs_position_is_relied_on() -> TestResult {
    let (_directory, mut storage) = create()?;
    declare(&mut storage)?;
    let reached = head(&storage, "main")?;
    embed(&mut storage, reached, Reach::End)?;
    advance(&mut storage, "main", "a")?;
    let now = advance(&mut storage, "main", "b")?;
    let floor = storage.relied_on("main", &none())?;
    assert_eq!(
        floor.held_by,
        [
            held(Source::Embed, "notes.body_vec", reached),
            held(Source::Head, "main", now),
        ]
    );
    Ok(())
}

/// Fails while an open walk's target is not held: both the finished
/// `through` and the target the walk is reading are relied on.
#[test]
fn an_open_walk_holds_its_start_and_its_target() -> TestResult {
    let (_directory, mut storage) = create()?;
    declare(&mut storage)?;
    let first = head(&storage, "main")?;
    embed(&mut storage, first, Reach::End)?;
    let second = advance(&mut storage, "main", "a")?;
    embed(&mut storage, second, Reach::Upto("a".to_owned()))?;
    let now = advance(&mut storage, "main", "b")?;
    let floor = storage.relied_on("main", &none())?;
    assert_eq!(
        floor.held_by,
        [
            held(Source::Embed, "notes.body_vec", first),
            held(Source::Embed, "notes.body_vec", second),
            held(Source::Head, "main", now),
        ]
    );
    Ok(())
}

/// A position in `owner`'s companion, for `notes.body_vec`, naming
/// `branch` at `through` and, when given, an open walk toward `toward`.
fn inherited(
    storage: &mut Storage,
    owner: &str,
    branch: &str,
    through: Version,
    toward: Option<Version>,
) -> Result<(), Failure> {
    let position = Position {
        branch: branch.to_owned(),
        through: Some(through),
        toward: toward.map(|version| Toward {
            version,
            after: "a".to_owned(),
        }),
    };
    storage.position_for_tests(owner, "notes", "body_vec", &position)
}

/// Fails while another branch's companion position naming this branch is
/// not read: main's floor holds the old version a fork's inherited position
/// names, under its own kind and the fork's name.
#[test]
fn a_position_in_another_companion_holds_this_branchs_version() -> TestResult {
    let (_directory, mut storage) = create()?;
    let old = advance(&mut storage, "main", "a")?;
    inherited(&mut storage, "feature", "main", old, None)?;
    let now = advance(&mut storage, "main", "b")?;
    let floor = storage.relied_on("main", &none())?;
    assert_eq!(
        floor.held_by,
        [
            held(Source::EmbedOf, "feature:notes.body_vec", old),
            held(Source::Head, "main", now),
        ]
    );
    Ok(())
}

/// Fails while a position naming another branch holds this one's version:
/// it is that branch's to hold.
#[test]
fn a_position_naming_another_branch_does_not_hold_this_one() -> TestResult {
    let (_directory, mut storage) = create()?;
    let old = advance(&mut storage, "main", "a")?;
    inherited(&mut storage, "feature", "elsewhere", old, None)?;
    let now = advance(&mut storage, "main", "b")?;
    let floor = storage.relied_on("main", &none())?;
    assert_eq!(floor.held_by, [held(Source::Head, "main", now)]);
    Ok(())
}

/// Fails while a version is held on a branch it is not a version of: a fork
/// mid-walk from its inherited position holds the parent's version on the
/// parent, and its own walk's target on itself, never the other way.
#[test]
fn an_inherited_open_walk_holds_each_version_on_its_own_branch() -> TestResult {
    let (_directory, mut storage) = create()?;
    let forked = fork(&mut storage, "main", "feature")?;
    let target = advance(&mut storage, "feature", "t")?;
    inherited(&mut storage, "feature", "main", forked, Some(target))?;
    let main_now = advance(&mut storage, "main", "b")?;
    let feature_now = advance(&mut storage, "feature", "u")?;
    let main_floor = storage.relied_on("main", &none())?;
    assert_eq!(
        main_floor.held_by,
        [
            held(Source::Fork, "feature", forked),
            held(Source::EmbedOf, "feature:notes.body_vec", forked),
            held(Source::Head, "main", main_now),
        ]
    );
    let feature_floor = storage.relied_on("feature", &none())?;
    assert_eq!(
        feature_floor.held_by,
        [
            held(Source::Embed, "notes.body_vec", target),
            held(Source::Head, "feature", feature_now),
        ]
    );
    Ok(())
}

/// Fails while a corrupt position key is guessed at: a companion holding a
/// key whose field part is not UTF-8 makes the floor refuse by name.
#[test]
fn a_corrupt_position_key_refuses_the_floor() -> TestResult {
    let (_directory, mut storage) = create()?;
    let now = advance(&mut storage, "main", "a")?;
    inherited(&mut storage, "main", "main", now, None)?;
    let position = Position {
        branch: "main".to_owned(),
        through: Some(now),
        toward: None,
    };
    let bytes = embedded::encode_position(&position)?;
    let key = b"e\0notes\0\xff".to_vec();
    let companion = embedded::companion_name("main");
    storage.commit_namespace(&companion, &[(key, Some(bytes))])?;
    let refused = storage.relied_on("main", &none()).err();
    let code = refused.map(|failure| failure.code);
    assert_eq!(code.as_deref(), Some("embed_position_key_corrupt"));
    Ok(())
}

/// A one-shard store keeping one version, `main` moved on 14 times, so its
/// oldest versions have left the window as rows, each given its older
/// history file, as a store from before rows has them; the versions with a
/// row, oldest first.
fn with_rows() -> Result<(tempfile::TempDir, Storage, Vec<Version>), Box<dyn Error>> {
    with_rows_after(14)
}

fn with_rows_after(
    advances: usize,
) -> Result<(tempfile::TempDir, Storage, Vec<Version>), Box<dyn Error>> {
    let directory = tempfile::tempdir()?;
    let mut storage = Storage::create_kept(
        DatabaseConfig {
            data_dir: directory.path().join("database"),
            shard_count: 1,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::reclaim(0),
        },
        NodeCacheBudget::Unlimited,
        KeptLimit::new(1)?,
        fixture_long_wait_interval(),
    )?;
    for tag in 0..advances {
        advance(&mut storage, "main", &format!("r{tag}"))?;
    }
    let rows = rows_below_head(&storage)?;
    let service = open_branch(SERVICE, &storage.refs, &storage.registry)?;
    for at in &rows {
        let key = Key {
            branch: Hash::of(b"main"),
            generation: at.generation,
            sequence: at.sequence,
        };
        let name = history::file_name("main", *at);
        storage.writer.fork_kept(
            &service,
            key,
            &mut storage.refs,
            &storage.registry,
            ForkWithChanges {
                name: &name,
                timestamp: current_timestamp(),
                writes: &[],
            },
        )?;
    }
    Ok((directory, storage, rows))
}

/// The versions of `main` with a row below its head, oldest first.
fn rows_below_head(storage: &Storage) -> Result<Vec<Version>, Box<dyn Error>> {
    let service = storage.refs.get(SERVICE).ok_or("no service record")?;
    let below = head(storage, "main")?;
    let rows = history::oldest_rows(
        &storage.reader,
        service,
        "main",
        (LOWEST, below),
        NonZeroUsize::new(1_000).ok_or("zero page")?,
    )?;
    Ok(rows
        .iter()
        .map(|row| Version {
            generation: row.key.generation,
            sequence: row.key.sequence,
        })
        .collect())
}

/// Keep the newest version and nothing for its age: everything below the
/// floor may go.
fn rules() -> Result<Rules, Box<dyn Error>> {
    Ok(Rules {
        keep: Some(Keep {
            versions: NonZeroU64::new(1),
            days: Some(0),
        }),
        max_rows: NonZeroUsize::new(64).ok_or("zero bound")?,
        now: current_timestamp(),
    })
}

/// As [`rules`], keeping the newest three versions. Of four advances' three
/// rows, the newest is then kept by `keep.versions`, its superseding time
/// known since DR-3, and the two below it go. Under a kept limit of 1 the
/// kept head is never trimmed (DR-7), so a test counting a release's durable
/// operations or listing its rows sees the release alone.
fn three_kept() -> Result<Rules, Box<dyn Error>> {
    let mut rules = rules()?;
    rules.keep = Some(Keep {
        versions: NonZeroU64::new(3),
        days: Some(0),
    });
    Ok(rules)
}

fn counts(storage: &Storage) -> SyncCounts {
    storage.writer.sync_counts()
}

/// `commits` service commits on this one shard, each one install, one
/// barrier, one ref-file and one directory flush (as `write_cost_tests`'s
/// `ONE_WRITE`), and `directories` directory flushes in all.
const fn ops(commits: u64, directories: u64) -> SyncCounts {
    SyncCounts {
        record_installs: commits,
        barrier_device_flushes: commits,
        ref_file_full_flushes: commits,
        directory_full_flushes: directories,
    }
}

/// Every node of `main` at `at`, read through its capture to the end.
fn whole(storage: &Storage, at: Version) -> TestResult {
    let captured = storage.capture(&Selection {
        branch: "main".to_owned(),
        version: Some(at),
    })?;
    let page = NonZeroUsize::new(256).ok_or("zero page")?;
    for shard in &captured.shards {
        let mut from = Vec::new();
        loop {
            let read = storage.reader.range_page_at(
                shard.shard_id,
                shard.head,
                &from,
                &[0xff; 8],
                page,
            )?;
            let Some(next) = read.next_from else {
                break;
            };
            from = next;
        }
    }
    Ok(())
}

/// Tom's rule 10, STOR-003 HR3: one release of ten rows makes three durable
/// operations, whatever the count: the decision's commit, the ten older
/// files' removal in one directory flush, and the rows' commit, so two
/// commits and three directory flushes.
#[test]
fn one_release_of_ten_rows_makes_three_durable_operations() -> TestResult {
    let (_directory, mut storage, rows) = with_rows()?;
    assert!(rows.len() > 10, "{rows:?}");
    let before = counts(&storage);
    let answer = storage.release("main", NonZeroUsize::new(10), &three_kept()?, &none())?;
    assert_eq!(counts(&storage).since(before), ops(2, 3));
    let span = answer.released.ok_or("nothing released")?;
    assert_eq!(Some(&span.first), rows.first());
    assert_eq!(Some(&span.last), rows.get(9));
    assert_eq!(
        rows_below_head(&storage)?.as_slice(),
        rows.get(10..).ok_or("too few rows")?
    );
    assert_eq!(
        storage.refs.list_prefix("@haem/history/").count(),
        rows.len() - 10
    );
    Ok(())
}

/// A stopped release resumes only the work its decision has not completed.
fn finish_stopped_release(point: Crash, finishing: SyncCounts) -> TestResult {
    let (_directory, mut storage, rows) = with_rows_after(4)?;
    // The newest row is kept by `keep.versions`: its successor's time is
    // known from the kept window since DR-3, so `keep.days` 0 would take it.
    let remaining = rows.last().copied().ok_or("no history rows")?;
    assert_eq!(
        rows.len(),
        3,
        "the partial-file crash needs two released rows"
    );
    crash_at(point);
    let stopped = storage.release("main", None, &three_kept()?, &none()).err();
    assert!(
        stopped
            .as_ref()
            .is_some_and(|failure| failure.code == "release_crashed"),
        "{point:?}: {stopped:?}"
    );
    let before = counts(&storage);
    storage.release("main", None, &three_kept()?, &none())?;
    assert_eq!(counts(&storage).since(before), finishing, "{point:?}");
    assert_eq!(rows_below_head(&storage)?, [remaining], "{point:?}");
    assert_eq!(
        storage.refs.list_prefix("@haem/history/").count(),
        1,
        "{point:?}"
    );
    assert!(
        storage
            .refs
            .get(&history::file_name("main", remaining))
            .is_some(),
        "{point:?}: the retained row's history file is missing"
    );
    let service = storage.refs.get(SERVICE).ok_or("no service record")?;
    assert_eq!(
        history::read_decision(&storage.reader, service, "main")?,
        None
    );
    Ok(())
}

#[test]
fn a_release_stopped_after_its_decision_is_finished_by_the_next_call() -> TestResult {
    finish_stopped_release(Crash::AfterDecision, ops(1, 2))
}

#[test]
fn a_release_stopped_during_its_files_is_finished_by_the_next_call() -> TestResult {
    finish_stopped_release(Crash::DuringFiles, ops(1, 2))
}

#[test]
fn a_release_stopped_after_its_files_is_finished_by_the_next_call() -> TestResult {
    finish_stopped_release(Crash::AfterFiles, ops(1, 1))
}

#[test]
fn a_release_stopped_after_its_commit_is_finished_by_the_next_call() -> TestResult {
    finish_stopped_release(Crash::AfterCommit, ops(0, 0))
}

/// STOR-003 HR3 and R2: after the older files are gone and before the rows'
/// commit, the oldest version taken reads whole after a reclaim pass run to
/// rest. Only its row keeps it: the store keeps one version and no past
/// roots.
#[test]
fn a_version_taken_reads_whole_after_a_pass_before_its_row_goes() -> TestResult {
    let (_directory, mut storage, rows) = with_rows()?;
    let first = *rows.first().ok_or("no rows")?;
    crash_at(Crash::AfterFiles);
    assert!(
        storage
            .release("main", None, &three_kept()?, &none())
            .is_err()
    );
    storage.writer.request_reclaim()?;
    storage.activate_reclaim()?;
    storage.quiesce_reclaim()?;
    whole(&storage, first)
}

/// STOR-003 HR3: a call that meets a decision left behind finishes that
/// decision, whatever its own limit, and decides nothing new beside it.
#[test]
fn a_call_meeting_a_left_decision_finishes_it_and_decides_nothing_new() -> TestResult {
    let (_directory, mut storage, rows) = with_rows()?;
    crash_at(Crash::AfterDecision);
    let left = storage.release("main", NonZeroUsize::new(4), &three_kept()?, &none());
    assert!(left.is_err());
    let answer = storage.release("main", Some(NonZeroUsize::MIN), &three_kept()?, &none())?;
    let span = answer.released.ok_or("the decision was not finished")?;
    assert_eq!(Some(&span.first), rows.first());
    assert_eq!(Some(&span.last), rows.get(3));
    Ok(())
}

/// A row of `main` in `generation` at `sequence`, written at `timestamp`.
fn row(generation: u64, sequence: u64, timestamp: u64) -> Row {
    Row {
        key: Key {
            branch: Hash::of(b"main"),
            generation,
            sequence,
        },
        timestamp,
        shards: Vec::new(),
    }
}

const fn at(generation: u64, sequence: u64) -> Version {
    Version {
        generation,
        sequence,
    }
}

/// What `judge` takes of `rows` below a head at `head`, written at
/// `written`, with nothing but the head relied on and room for every row.
fn judged(
    rows: &[Row],
    head: Version,
    written: u64,
    keep: Option<Keep>,
    now: u64,
) -> Result<(Vec<Version>, Vec<HeldBy>), Box<dyn Error>> {
    let floor = Floor {
        held_by: vec![HeldBy {
            source: Source::Head,
            name: "main".to_owned(),
            version: head,
        }],
        branches_read: 0,
    };
    let rules = Rules {
        keep,
        max_rows: NonZeroUsize::new(64).ok_or("zero bound")?,
        now,
    };
    let below = Below {
        version: head,
        written,
    };
    let (taken, holders, _more) = judge("main", rows, 64, &floor, below, &rules);
    Ok((taken, holders))
}

fn kept(source: Source, version: Version) -> Vec<HeldBy> {
    vec![HeldBy {
        source,
        name: "main".to_owned(),
        version,
    }]
}

/// Keep newest `versions` and nothing for age.
const fn newest(versions: u64) -> Keep {
    Keep {
        versions: NonZeroU64::new(versions),
        days: Some(0),
    }
}

/// STOR-003 HR3: keep.versions 5 keeps exactly the newest five, the head
/// among them: with the head at sequence 12, rows 8 to 11 stay and 0 to 7 go.
#[test]
fn keep_versions_keeps_exactly_the_newest_within_one_generation() -> TestResult {
    let rows: Vec<Row> = (0..12).map(|sequence| row(1, sequence, 1)).collect();
    let (taken, held) = judged(&rows, at(1, 12), 1, Some(newest(5)), u64::MAX)?;
    assert_eq!(
        taken,
        (0..8).map(|sequence| at(1, sequence)).collect::<Vec<_>>()
    );
    assert_eq!(held, kept(Source::KeepVersions, at(1, 8)));
    Ok(())
}

/// Fails while the head generation is counted twice (the red for HR3-fix):
/// keep.versions 10, the head at generation 2 sequence 5, generation 1's
/// rows 0 to 9 and generation 2's rows 0 to 4. The newest ten are
/// generation 2's six and generation 1's 6 to 9, so 6 to 9 stay: newer than
/// generation 1's sequence 9 are six versions, not eleven.
#[test]
fn keep_versions_counts_the_head_generation_once_across_a_generation() -> TestResult {
    let mut rows: Vec<Row> = (0..10).map(|sequence| row(1, sequence, 1)).collect();
    rows.extend((0..5).map(|sequence| row(2, sequence, 1)));
    let (taken, held) = judged(&rows, at(2, 5), 1, Some(newest(10)), u64::MAX)?;
    assert_eq!(
        taken,
        (0..6).map(|sequence| at(1, sequence)).collect::<Vec<_>>()
    );
    assert_eq!(held, kept(Source::KeepVersions, at(1, 6)));
    Ok(())
}

/// STOR-003 HR3: keep.days 1 keeps a version superseded less than a day
/// ago, by one nanosecond, and lets one superseded exactly a day ago go.
#[test]
fn keep_days_holds_to_the_nanosecond() -> TestResult {
    let rows = [
        row(1, 0, 1_000),
        row(1, 1, 2_000),
        row(1, 2, 3_000),
        row(1, 3, 3_001),
    ];
    let days = Some(Keep {
        versions: NonZeroU64::new(1),
        days: Some(1),
    });
    let now = 3_000 + NANOS_A_DAY;
    let (taken, held) = judged(&rows, at(1, 4), 4_000, days, now)?;
    assert_eq!(taken, [at(1, 0), at(1, 1)]);
    assert_eq!(held, kept(Source::KeepDays, at(1, 2)));
    let (taken, held) = judged(&rows, at(1, 4), 4_000, days, now - 1)?;
    assert_eq!(taken, [at(1, 0)]);
    assert_eq!(held, kept(Source::KeepDays, at(1, 1)));
    Ok(())
}

/// STOR-003 HR3: a keep setting left unset keeps every version, and the
/// release says which setting held it.
#[test]
fn an_unset_keep_setting_keeps_all_and_is_named() -> TestResult {
    let rows = [row(1, 0, 1), row(1, 1, 2)];
    let versions_unset = Some(Keep {
        versions: None,
        days: Some(0),
    });
    let (taken, held) = judged(&rows, at(1, 2), 3, versions_unset, u64::MAX)?;
    assert!(taken.is_empty());
    assert_eq!(held, kept(Source::KeepVersions, at(1, 0)));
    let days_unset = Some(Keep {
        versions: NonZeroU64::new(1),
        days: None,
    });
    let (taken, held) = judged(&rows, at(1, 2), 3, days_unset, u64::MAX)?;
    assert!(taken.is_empty());
    assert_eq!(held, kept(Source::KeepDays, at(1, 0)));
    Ok(())
}

/// STOR-003 HR3: `keep: None`, as a finishing discard asks, judges no keep
/// setting: every row below the floor goes.
#[test]
fn no_keep_rules_judge_nothing_but_the_floor() -> TestResult {
    let rows = [row(1, 0, 1), row(1, 1, 2), row(1, 2, 3)];
    let (taken, held) = judged(&rows, at(1, 3), 4, None, 0)?;
    assert_eq!(taken, [at(1, 0), at(1, 1), at(1, 2)]);
    assert!(held.is_empty());
    Ok(())
}

/// STOR-003 HR3: a row whose superseding time is unknown, the last one read
/// with versions between it and the head unread, is kept whatever its age.
#[test]
fn a_row_superseded_at_an_unknown_time_is_kept() -> TestResult {
    let rows = [row(1, 0, 1), row(1, 1, 2)];
    let (taken, held) = judged(&rows, at(1, 5), 6, Some(newest(1)), u64::MAX)?;
    assert_eq!(taken, [at(1, 0)]);
    assert_eq!(held, kept(Source::KeepDays, at(1, 1)));
    Ok(())
}

/// Release settings that keep every version: nothing new may be taken.
fn keep_all() -> Result<Settings, Box<dyn Error>> {
    Ok(Settings {
        max_rows: NonZeroUsize::new(64).ok_or("zero rows")?,
        batch: NonZeroUsize::MIN,
        keep: Keep {
            versions: None,
            days: None,
        },
    })
}

/// The store `with_rows` made, opened again as a restart does.
fn reopened(directory: &tempfile::TempDir) -> Result<Storage, Box<dyn Error>> {
    Ok(Storage::open_kept(
        &directory.path().join("database"),
        NodeCacheBudget::Unlimited,
        KeptLimit::new(1)?,
        fixture_long_wait_interval(),
    )?)
}

/// Start the release driver over `storage`, as the service does, and give
/// the storage back once it has stopped.
fn started(storage: Storage) -> Result<(Storage, Option<Answer>), Box<dyn Error>> {
    let storage = Arc::new(Mutex::new(storage));
    let releaser = Releaser::start(Arc::clone(&storage), keep_all()?)?;
    let last = releaser.last().map(|last| last.answer);
    drop(releaser);
    let storage = Arc::try_unwrap(storage)
        .map_err(|_| "the storage is still shared")?
        .into_inner()
        .map_err(|_| "the storage lock was poisoned")?;
    Ok((storage, last))
}

/// STOR-003 HR4-fix2 (Apollo 14:24): a decision is the commit point. Left
/// after its files with keep set, it is rolled forward as written at the
/// next start with keep unset: its span is released, its four rows go, and
/// every version above the span reads whole. Red while the decision is
/// judged again: nothing released and the rows unchanged.
#[test]
fn a_left_decision_is_rolled_forward_as_written_at_start() -> TestResult {
    let (directory, mut storage, rows) = with_rows()?;
    crash_at(Crash::AfterFiles);
    let left = storage.release("main", NonZeroUsize::new(4), &three_kept()?, &none());
    assert!(left.is_err());
    drop(storage);
    let (storage, last) = started(reopened(&directory)?)?;
    let span = last
        .and_then(|answer| answer.released)
        .ok_or("the decision was not rolled forward")?;
    assert_eq!(Some(&span.first), rows.first());
    assert_eq!(Some(&span.last), rows.get(3));
    assert!(storage.decided_branches()?.is_empty());
    assert_eq!(
        rows_below_head(&storage)?,
        rows.get(4..).unwrap_or_default()
    );
    storage.writer.request_reclaim()?;
    storage.activate_reclaim()?;
    storage.quiesce_reclaim()?;
    for at in rows.get(4..).unwrap_or_default() {
        whole(&storage, *at)?;
    }
    Ok(())
}

/// STOR-003 HR4-fix2: start finishes a decision before it returns, so no
/// hold can be made on a decided version first; a pass afterwards leaves
/// every version the store still lists readable.
#[test]
fn start_finishes_a_left_decision_before_it_returns() -> TestResult {
    let (directory, mut storage, rows) = with_rows()?;
    crash_at(Crash::AfterDecision);
    let left = storage.release("main", NonZeroUsize::new(4), &three_kept()?, &none());
    assert!(left.is_err());
    drop(storage);
    let storage = Arc::new(Mutex::new(reopened(&directory)?));
    let releaser = Releaser::start(Arc::clone(&storage), keep_all()?)?;
    let decided = storage
        .lock()
        .map_err(|_| "the storage lock was poisoned")?
        .decided_branches()?;
    assert!(decided.is_empty());
    drop(releaser);
    let storage = Arc::try_unwrap(storage)
        .map_err(|_| "the storage is still shared")?
        .into_inner()
        .map_err(|_| "the storage lock was poisoned")?;
    storage.writer.request_reclaim()?;
    storage.activate_reclaim()?;
    storage.quiesce_reclaim()?;
    let listed = rows_below_head(&storage)?;
    assert_eq!(listed, rows.get(4..).unwrap_or_default());
    for at in &listed {
        whole(&storage, *at)?;
    }
    whole(&storage, head(&storage, "main")?)
}

/// STOR-003 HR4-fix2: a decision whose span no longer holds its count of
/// rows is refused by name and left; start fails by name, never guessing.
#[test]
fn a_decision_whose_rows_differ_from_its_count_is_refused() -> TestResult {
    let (directory, mut storage, rows) = with_rows_after(4)?;
    crash_at(Crash::AfterDecision);
    let left = storage.release("main", NonZeroUsize::new(4), &three_kept()?, &none());
    assert!(
        left.as_ref()
            .is_err_and(|failure| failure.code == "release_crashed"),
        "{left:?}"
    );
    let second = *rows.get(1).ok_or("too few rows")?;
    let service = storage.refs.get(SERVICE).ok_or("no service record")?;
    let decision = history::read_decision(&storage.reader, service, "main")?
        .ok_or("the crash left no decision")?;
    assert_eq!(decision.count, 2);
    assert_eq!(decision.last, second);
    storage.commit_service(&[(history::row_key("main", second), None)])?;
    let refused = storage
        .release("main", None, &three_kept()?, &none())
        .err()
        .ok_or("a decision short of its rows was finished")?;
    assert_eq!(refused.code, "release_decision_mismatch");
    assert_eq!(storage.decided_branches()?, ["main"]);
    // RC2 (6C2): the row's history file goes with its row, as a release
    // removes both; a file left without its row is refused at open as
    // history_rows_incomplete (history.rs require_file_rows) before start
    // could name the decision.
    storage.refs.remove(&history::file_name("main", second))?;
    drop(storage);
    let storage = Arc::new(Mutex::new(reopened(&directory)?));
    let failed = Releaser::start(Arc::clone(&storage), keep_all()?)
        .err()
        .ok_or("start finished a decision short of its rows")?;
    assert_eq!(failed.code, "release_decision_unfinished");
    assert!(failed.message.contains("release_decision_mismatch"));
    Ok(())
}

/// DR-3: a branch that stops advancing still releases its newest row once
/// its days pass, counted from when the record's next version was installed
/// (docs/design/HAEM-STALLED-BRANCH-NEWEST-ROW-2026-10-06.md). One version a
/// record keeps, so three advances leave rows for the first two versions;
/// the newest row's next version is the one the record keeps.
#[test]
fn a_stalled_branch_releases_its_newest_row_once_its_days_pass() -> TestResult {
    let directory = tempfile::tempdir()?;
    let mut storage = Storage::create_kept(
        DatabaseConfig {
            data_dir: directory.path().join("database"),
            shard_count: 1,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::reclaim(0),
        },
        NodeCacheBudget::Unlimited,
        KeptLimit::new(1)?,
        fixture_long_wait_interval(),
    )?;
    let newest = advance(&mut storage, "main", "one")?;
    advance(&mut storage, "main", "two")?;
    advance(&mut storage, "main", "three")?;
    let installed = storage
        .refs
        .kept(&branch_name("main"))
        .first()
        .map(|entry| entry.timestamp)
        .ok_or("the record keeps no version")?;
    let young = Rules {
        keep: Some(Keep {
            versions: NonZeroU64::new(1),
            days: Some(1),
        }),
        max_rows: NonZeroUsize::new(64).ok_or("zero bound")?,
        now: installed + NANOS_A_DAY - 1,
    };
    let held = storage.release("main", None, &young, &none())?;
    assert_eq!(held.held_by, kept(Source::KeepDays, newest), "{held:?}");
    assert_eq!(
        held.due_at,
        Some(installed + NANOS_A_DAY),
        "its day is known"
    );

    let old = Rules {
        now: installed + NANOS_A_DAY,
        ..young
    };
    let answer = storage.release("main", None, &old, &none())?;
    assert_eq!(
        answer.released.map(|span| span.last),
        Some(newest),
        "{answer:?}"
    );
    Ok(())
}

/// `work` advanced twice and discarded, then made again from `main` and left:
/// its newest row is the old generation's last, whose version this returns.
fn recreated(storage: &mut Storage) -> Result<Version, Box<dyn Error>> {
    fork(storage, "main", "work")?;
    advance(storage, "work", "a")?;
    let last = advance(storage, "work", "b")?;
    storage.mutate(&Mutation {
        request_id: "discard-work".to_owned(),
        branch: "work".to_owned(),
        expected: last,
        change: Change::Discard(BranchDiscard {
            name: "work".to_owned(),
        }),
    })?;
    let at = head(storage, "main")?;
    storage.mutate(&Mutation {
        request_id: "recreate-work".to_owned(),
        branch: "main".to_owned(),
        expected: at,
        change: Change::Fork(Fork {
            name: "work".to_owned(),
            from: at,
        }),
    })?;
    Ok(last)
}

/// Keep one version and one day, at `now`.
fn a_day_at(now: u64) -> Result<Rules, Box<dyn Error>> {
    Ok(Rules {
        keep: Some(Keep {
            versions: NonZeroU64::new(1),
            days: Some(1),
        }),
        max_rows: NonZeroUsize::new(64).ok_or("zero bound")?,
        now,
    })
}

/// Reads of marks and markers on this thread since the last call.
fn reads_since() -> (usize, usize) {
    (
        MARK_READS.with(|reads| reads.replace(0)),
        MARKER_READS.with(|reads| reads.replace(0)),
    )
}

/// DR-6: a recreated name's newest row is its old generation's last, and it
/// was superseded when that generation was discarded, read from the
/// generation's marker: held until a day after, then released.
#[test]
fn a_recreated_names_older_generation_falls_due_after_its_days() -> TestResult {
    let (_directory, mut storage) = create()?;
    let last = recreated(&mut storage)?;
    let service = storage.refs.get(SERVICE).cloned().ok_or("no service")?;
    let marker = dead::marker(&storage.reader, &service, "work", last.generation)?;
    let due = marker.ok_or("no marker")?.discarded_at + NANOS_A_DAY;
    reads_since();

    let young = storage.release("work", None, &a_day_at(due - 1)?, &none())?;
    assert_eq!(
        young.held_by,
        [held(Source::KeepDays, "work", last)],
        "{young:?}"
    );
    assert_eq!(young.due_at, Some(due), "its day is the discard's");
    // The current generation's mark, its own key then the name's old one;
    // the old generation's marker for its date, then as rows before it go.
    assert_eq!(reads_since(), (2, 2), "mark and marker reads");

    let old = storage.release("work", None, &a_day_at(due)?, &none())?;
    assert_eq!(old.released.map(|span| span.last), Some(last), "{old:?}");
    assert_eq!(reads_since(), (2, 2), "mark and marker reads");
    Ok(())
}

/// The control: an old generation with no marker, as one discarded before
/// DR-1 has, stays unknown: held with no due time, as before.
#[test]
fn an_older_generation_with_no_marker_stays_held() -> TestResult {
    let (_directory, mut storage) = create()?;
    let last = recreated(&mut storage)?;
    storage.commit_service(&[(dead::dead_key("work", last.generation), None)])?;
    let answer = storage.release("work", None, &a_day_at(u64::MAX)?, &none())?;
    assert_eq!(
        answer.held_by,
        [held(Source::KeepDays, "work", last)],
        "{answer:?}"
    );
    assert_eq!(answer.due_at, None, "no marker, no date");
    Ok(())
}

/// A store 0.15.0 wrote marks `branch` released to `version` under the
/// name's own key, as 0.15.0's release wrote it, through the service writer.
fn seed_old_mark(storage: &mut Storage, branch: &str, version: Version) -> TestResult {
    let mark = history::Released { version, at: 1 };
    let bytes = serde_json::to_vec(&mark)?;
    storage.commit_service(&[(history::released_key(branch), Some(bytes))])?;
    Ok(())
}

/// The versions of `branch`'s rows in `generation`, oldest first.
fn rows_of(
    storage: &Storage,
    branch: &str,
    generation: u64,
) -> Result<Vec<Version>, Box<dyn Error>> {
    let service = storage.refs.get(SERVICE).cloned().ok_or("no service")?;
    let span = (
        Version {
            generation,
            sequence: 0,
        },
        Version {
            generation,
            sequence: u64::MAX,
        },
    );
    let limit = NonZeroUsize::new(64).ok_or("zero bound")?;
    let rows = history::oldest_rows(&storage.reader, &service, branch, span, limit)?;
    Ok(rows.iter().map(super::row_version).collect())
}

/// `branch`'s mark for `generation`, read as a release reads it.
fn mark_of(
    storage: &Storage,
    branch: &str,
    generation: u64,
) -> Result<Option<Version>, Box<dyn Error>> {
    let service = storage.refs.get(SERVICE).cloned().ok_or("no service")?;
    let mark = history::read_released(&storage.reader, &service, branch, generation)?;
    Ok(mark.map(|mark| mark.version))
}

/// DR-6b: a 0.15.0 store marked `work` at a version of its second
/// generation, its first generation's rows still there. A release takes
/// nothing of generation 1, which the old mark covers whole, starts
/// generation 2 past the mark, and marks generation 2 under its own key.
#[test]
fn an_old_mark_at_a_later_generation_covers_the_earlier_one_whole() -> TestResult {
    let (_directory, mut storage) = create()?;
    let old = recreated(&mut storage)?;
    let marked = advance(&mut storage, "work", "c")?;
    let next = advance(&mut storage, "work", "d")?;
    let last = advance(&mut storage, "work", "e")?;
    storage.mutate(&Mutation {
        request_id: "discard-work-again".to_owned(),
        branch: "work".to_owned(),
        expected: last,
        change: Change::Discard(BranchDiscard {
            name: "work".to_owned(),
        }),
    })?;
    seed_old_mark(&mut storage, "work", marked)?;
    let first_generation = rows_of(&storage, "work", old.generation)?;
    assert!(!first_generation.is_empty(), "generation 1 has rows");
    reads_since();

    let rules = Rules {
        keep: None,
        max_rows: NonZeroUsize::new(64).ok_or("zero bound")?,
        now: u64::MAX,
    };
    let answer = storage.release("work", None, &rules, &none())?;
    let span = answer.released.map(|span| (span.first, span.last));
    assert_eq!(span, Some((next, last)), "{answer:?}");
    // Each generation's own key, absent, then the name's old key.
    assert_eq!(reads_since().0, 4, "mark reads");
    assert_eq!(
        rows_of(&storage, "work", old.generation)?,
        first_generation,
        "generation 1 left"
    );
    assert_eq!(
        mark_of(&storage, "work", old.generation)?,
        Some(marked),
        "the old key as seeded"
    );
    assert_eq!(
        mark_of(&storage, "work", marked.generation)?,
        Some(last),
        "generation 2's own key"
    );
    Ok(())
}

/// DR-6b: a 0.15.0 store marked `work` in its first generation, its rows to
/// the mark gone, then `work` was made again. The old mark is not the new
/// generation's: a release reports no mark, reads from the oldest row, and
/// takes only the rows past the mark, once they are due.
#[test]
fn a_live_names_old_mark_is_not_its_new_generations() -> TestResult {
    let (_directory, mut storage) = create()?;
    let last = recreated(&mut storage)?;
    let rows = rows_of(&storage, "work", last.generation)?;
    let (marked, left) = rows.split_first().ok_or("no rows")?;
    seed_old_mark(&mut storage, "work", *marked)?;
    storage.commit_service(&[(history::row_key("work", *marked), None)])?;
    reads_since();

    let young = storage.release("work", None, &a_day_at(current_timestamp())?, &none())?;
    assert!(young.released.is_none(), "{young:?}");
    assert_eq!(
        young.marker, None,
        "generation 1's old mark is not generation 2's"
    );
    assert_eq!(reads_since().0, 2, "its own key, then the name's");

    let old = storage.release("work", None, &a_day_at(u64::MAX)?, &none())?;
    let span = old.released.map(|span| (span.first, span.last));
    let first_left = *left.first().ok_or("no row left")?;
    assert_eq!(span, Some((first_left, last)), "{old:?}");
    assert_eq!(
        mark_of(&storage, "work", last.generation)?,
        Some(last),
        "generation 1's own key"
    );
    Ok(())
}

/// DR-6b: with only the name's old key, a generation's mark costs two point
/// reads, its own key and then the name's; once its own is written, one.
#[test]
fn an_old_mark_alone_costs_two_point_reads() -> TestResult {
    let (_directory, mut storage) = create()?;
    let at = advance(&mut storage, "main", "a")?;
    seed_old_mark(&mut storage, "main", at)?;
    reads_since();
    assert_eq!(mark_of(&storage, "main", at.generation)?, Some(at));
    assert_eq!(reads_since().0, 2, "its own key, then the name's");

    let mark = history::Released { version: at, at: 2 };
    let key = history::released_key_at("main", at.generation);
    storage.commit_service(&[(key, Some(serde_json::to_vec(&mark)?))])?;
    assert_eq!(mark_of(&storage, "main", at.generation)?, Some(at));
    assert_eq!(reads_since().0, 1, "its own key only");
    Ok(())
}

/// `main` advanced five times under the default window, so its record keeps
/// five heads and no rows; the oldest version and the time of the head.
fn stalled(storage: &mut Storage) -> Result<(Version, Version, u64), Box<dyn Error>> {
    let first = head(storage, "main")?;
    let mut heads = Vec::new();
    for tag in ["a", "b", "c", "d", "e"] {
        heads.push(advance(storage, "main", tag)?);
    }
    let last_kept = *heads.get(3).ok_or("five advances")?;
    let record = storage.refs.get(&branch_name("main")).ok_or("no main")?;
    Ok((first, last_kept, record.timestamp))
}

/// Kept heads `main`'s record holds now.
fn kept_count(storage: &Storage) -> usize {
    storage.refs.kept(&branch_name("main")).len()
}

/// Service commits of kept rows on this thread since the last call.
fn row_commits_since() -> usize {
    ROW_COMMITS.with(|commits| commits.replace(0))
}

/// The heads a capture reads, one per shard.
fn heads_of(captured: &Captured) -> Vec<Hash> {
    captured.shards.iter().map(|shard| shard.head).collect()
}

/// DR-7 (candidate B): a stalled branch's release names when its oldest kept
/// head falls due; a release a day after the head trims its five kept heads,
/// their rows written first in one commit, and takes nothing; the next
/// release takes the five rows.
#[test]
fn a_stalled_branchs_kept_heads_leave_after_their_days() -> TestResult {
    let (_directory, mut storage) = create()?;
    let (first, last_kept, written) = stalled(&mut storage)?;
    assert_eq!(kept_count(&storage), 5, "five heads kept");
    let kept = storage.refs.kept(&branch_name("main"));
    let superseded = kept.get(1).ok_or("five kept")?.timestamp;
    let young = storage.release("main", None, &a_day_at(written)?, &none())?;
    assert_eq!(young.due_at, Some(superseded + NANOS_A_DAY), "{young:?}");
    assert_eq!(kept_count(&storage), 5, "none due yet");
    let rules = a_day_at(written + NANOS_A_DAY)?;
    row_commits_since();

    let trimmed = storage.release("main", None, &rules, &none())?;
    assert!(trimmed.released.is_none(), "{trimmed:?}");
    assert_eq!(kept_count(&storage), 0, "the kept heads left");
    assert_eq!(row_commits_since(), 1, "their rows in one commit");

    let taken = storage.release("main", None, &rules, &none())?;
    let span = taken.released.map(|span| (span.first, span.last));
    assert_eq!(span, Some((first, last_kept)), "{taken:?}");
    assert_eq!(row_commits_since(), 0, "no row written again");
    Ok(())
}

/// DR-7: a crash between the rows commit and the trim leaves the rows and
/// the heads still kept. The next release trims, writes no row again, and
/// takes each row once.
#[test]
fn a_crash_between_the_rows_and_the_trim_is_finished_by_the_next_release() -> TestResult {
    let (_directory, mut storage) = create()?;
    let (first, last_kept, written) = stalled(&mut storage)?;
    let rules = a_day_at(written + NANOS_A_DAY)?;
    row_commits_since();

    crash_at(Crash::BeforeTrim);
    let stopped = storage.release("main", None, &rules, &none()).err();
    assert!(
        stopped
            .as_ref()
            .is_some_and(|failure| failure.code == "release_crashed"),
        "{stopped:?}"
    );
    assert_eq!(kept_count(&storage), 5, "the heads are still kept");
    assert_eq!(row_commits_since(), 1, "the rows are there");

    let finished = storage.release("main", None, &rules, &none())?;
    assert_eq!(kept_count(&storage), 0, "trimmed");
    assert_eq!(row_commits_since(), 0, "no row written twice");
    let span = finished.released.map(|span| (span.first, span.last));
    assert_eq!(span, Some((first, last_kept)), "{finished:?}");
    let again = storage.release("main", None, &rules, &none())?;
    assert!(again.released.is_none(), "nothing taken twice: {again:?}");
    Ok(())
}

/// The versions `reader` lists in `main`'s history.
fn listed(reader: &Catalogue) -> Result<Vec<Version>, Failure> {
    let history = reader.history("main", None, 100)?;
    Ok(history.iter().map(|captured| captured.version).collect())
}

/// DR-7: a reader beside the writer reads the oldest version from the kept
/// head, then from its row once the head has left the record, the same
/// heads, and the history it lists is the same; once the row is released it
/// refuses the version.
#[test]
fn a_reader_beside_the_writer_reads_a_trimmed_head_from_its_row() -> TestResult {
    let (_directory, mut storage) = create()?;
    let (first, _last_kept, written) = stalled(&mut storage)?;
    let rules = a_day_at(written + NANOS_A_DAY)?;
    let reader = storage.catalogue();
    let oldest = Selection {
        branch: "main".to_owned(),
        version: Some(first),
    };
    let before = reader.capture(&oldest)?;
    let listed_before = listed(&reader)?;

    storage.release("main", None, &rules, &none())?;
    assert_eq!(kept_count(&storage), 0, "the kept heads left");
    let between = reader.capture(&oldest)?;
    assert_eq!(heads_of(&between), heads_of(&before), "read from its row");
    assert_eq!(listed(&reader)?, listed_before, "the same history");

    storage.release("main", None, &rules, &none())?;
    let after = reader.capture(&oldest);
    assert!(
        matches!(&after, Err(failure) if failure.code == "version_released"),
        "{after:?}"
    );
    Ok(())
}

/// A store keeping one previous version, so rows are written as heads leave.
fn kept_one() -> Result<(tempfile::TempDir, Storage), Box<dyn Error>> {
    let directory = tempfile::tempdir()?;
    let storage = Storage::create_kept(
        DatabaseConfig {
            data_dir: directory.path().join("database"),
            shard_count: 1,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::reclaim(0),
        },
        NodeCacheBudget::Unlimited,
        KeptLimit::new(1)?,
        fixture_long_wait_interval(),
    )?;
    Ok((directory, storage))
}

/// `branch` writes `change` at its head under `request_id`.
fn mutate(
    storage: &mut Storage,
    branch: &str,
    request_id: &str,
    change: Change,
) -> Result<(), Box<dyn Error>> {
    let expected = head(storage, branch)?;
    storage.mutate(&Mutation {
        request_id: request_id.to_owned(),
        branch: branch.to_owned(),
        expected,
        change,
    })?;
    Ok(())
}

/// DR-8: `work`'s first generation, forked to `child` at its first version,
/// is discarded; `work` is made again from `main` and advanced three times.
/// The child holds all of generation 1's rows. A release of `work` reads its
/// oldest rows once, passes generation 1's, and takes generation 2's behind
/// them; the child still reads.
#[test]
fn a_fork_on_an_older_generation_does_not_block_a_live_names_newer_rows() -> TestResult {
    let (_directory, mut storage) = kept_one()?;
    fork(&mut storage, "main", "work")?;
    let first = head(&storage, "work")?;
    advance(&mut storage, "work", "a")?;
    advance(&mut storage, "work", "b")?;
    let child = Fork {
        name: "child".to_owned(),
        from: first,
    };
    mutate(&mut storage, "work", "fork-child", Change::Fork(child))?;
    let discard = BranchDiscard {
        name: "work".to_owned(),
    };
    mutate(
        &mut storage,
        "work",
        "discard-work",
        Change::Discard(discard),
    )?;
    let at = head(&storage, "main")?;
    let again = Fork {
        name: "work".to_owned(),
        from: at,
    };
    mutate(&mut storage, "main", "recreate-work", Change::Fork(again))?;
    let made = head(&storage, "work")?;
    let second = advance(&mut storage, "work", "c")?;
    advance(&mut storage, "work", "d")?;
    advance(&mut storage, "work", "e")?;
    ROW_READS.with(|reads| reads.set(0));

    // Keep one version, none for age: generation 2's rows below its kept
    // head go, the made version and `c`.
    let answer = storage.release("work", None, &rules()?, &none())?;
    let span = answer.released.map(|span| (span.first, span.last));
    assert_eq!(span, Some((made, second)), "{answer:?}");
    assert_eq!(ROW_READS.with(std::cell::Cell::get), 1, "one read of rows");
    let held = storage.capture(&Selection {
        branch: "child".to_owned(),
        version: None,
    })?;
    assert_eq!(held.branch, "child", "the child still reads");
    let old = storage.capture(&Selection {
        branch: "work".to_owned(),
        version: Some(first),
    });
    assert!(old.is_ok(), "generation 1 is not marked released: {old:?}");
    Ok(())
}

/// DR-13, DR-8 FOUND 1: as above, with generation 1 longer than a read. A
/// release limited to two rows reads generation 1's first three, all held by
/// the child. Today it passes them, takes nothing, and every release after
/// reads the same three. Now it reads again from generation 2 and takes
/// generation 2's rows behind its kept head.
#[test]
fn a_held_generation_longer_than_a_read_does_not_block_a_live_names_newer_rows() -> TestResult {
    let (_directory, mut storage) = kept_one()?;
    fork(&mut storage, "main", "work")?;
    let first = head(&storage, "work")?;
    for value in ["a", "b", "c"] {
        advance(&mut storage, "work", value)?;
    }
    let child = Fork {
        name: "child".to_owned(),
        from: first,
    };
    mutate(&mut storage, "work", "fork-child", Change::Fork(child))?;
    let discard = BranchDiscard {
        name: "work".to_owned(),
    };
    mutate(
        &mut storage,
        "work",
        "discard-work",
        Change::Discard(discard),
    )?;
    let at = head(&storage, "main")?;
    let again = Fork {
        name: "work".to_owned(),
        from: at,
    };
    mutate(&mut storage, "main", "recreate-work", Change::Fork(again))?;
    let made = head(&storage, "work")?;
    let second = advance(&mut storage, "work", "d")?;
    advance(&mut storage, "work", "e")?;
    advance(&mut storage, "work", "f")?;
    ROW_READS.with(|reads| reads.set(0));

    let two = NonZeroUsize::new(2).ok_or("two is not zero")?;
    let answer = storage.release("work", Some(two), &rules()?, &none())?;
    let span = answer.released.map(|span| (span.first, span.last));
    assert_eq!(span, Some((made, second)), "{answer:?}");
    assert_eq!(
        ROW_READS.with(std::cell::Cell::get),
        2,
        "one read past generation 1"
    );
    let old = storage.capture(&Selection {
        branch: "work".to_owned(),
        version: Some(first),
    });
    assert!(old.is_ok(), "generation 1 is not marked released: {old:?}");
    Ok(())
}

/// `work` forked from `main`, advanced twice and discarded, its marker then
/// taken away, as a store from before DR-1 never wrote one; its last
/// version.
fn undated(storage: &mut Storage) -> Result<Version, Box<dyn Error>> {
    fork(storage, "main", "work")?;
    advance(storage, "work", "a")?;
    let last = advance(storage, "work", "b")?;
    let discard = BranchDiscard {
        name: "work".to_owned(),
    };
    mutate(storage, "work", "discard-work", Change::Discard(discard))?;
    storage.commit_service(&[(dead::dead_key("work", last.generation), None)])?;
    Ok(last)
}

/// The code a refused dating request names.
fn refusal(storage: &mut Storage, request: &UndatedGeneration) -> Option<String> {
    storage
        .date_undated_generation(request)
        .err()
        .map(|failure| failure.code)
}

/// DR-11: a generation discarded before DR-1 is found by no release. An
/// operator's request dates it in one read of its rows; it falls due a day
/// after that time and its last row goes. A second request, one for a live
/// name and one for a generation with no rows are refused by name.
#[test]
fn an_undated_generation_is_released_once_an_operator_dates_it() -> TestResult {
    let (_directory, mut storage) = create()?;
    let last = undated(&mut storage)?;
    let missing = storage.release("work", None, &a_day_at(u64::MAX)?, &none());
    assert!(
        missing
            .as_ref()
            .is_err_and(|failure| failure.code == "branch_missing"),
        "{missing:?}"
    );

    let at = current_timestamp();
    let request = UndatedGeneration {
        branch: "work".to_owned(),
        generation: last.generation,
        at,
    };
    ROW_READS.with(|reads| reads.set(0));
    assert_eq!(storage.date_undated_generation(&request)?, last);
    assert_eq!(
        ROW_READS.with(std::cell::Cell::get),
        1,
        "one read of its rows"
    );
    let again = refusal(&mut storage, &request);
    assert_eq!(again.as_deref(), Some("generation_marked"));
    let live = UndatedGeneration {
        branch: "main".to_owned(),
        ..request.clone()
    };
    assert_eq!(refusal(&mut storage, &live).as_deref(), Some("branch_live"));
    let empty = UndatedGeneration {
        generation: last.generation + 1,
        ..request.clone()
    };
    let none_there = refusal(&mut storage, &empty);
    assert_eq!(none_there.as_deref(), Some("generation_empty"));

    let due = at + NANOS_A_DAY;
    let young = storage.release("work", None, &a_day_at(due - 1)?, &none())?;
    assert_eq!(young.due_at, Some(due), "{young:?}");
    let old = storage.release("work", None, &a_day_at(due)?, &none())?;
    assert_eq!(old.released.map(|span| span.last), Some(last), "{old:?}");
    Ok(())
}

const fn fixture_long_wait_interval() -> std::time::Duration {
    const LONG_WAIT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
    LONG_WAIT_INTERVAL
}