haematite-cli 0.21.8

//! STOR-003 `MD5a`: one served write is one durable commit; history rows are
//! written only for versions about to leave the record's kept window, every
//! kept version without one in one service commit.
//!
//! Counts derived from the source before any run, one shard:
//! - a write that drops no version without a row: the data commit only, 1
//!   record install, 1 barrier device flush, 1 ref-file and 1 directory full
//!   flush;
//! - a write that does: first the rows' service commit, the same four (its
//!   rows all on the one shard), then the data commit;
//! - main starts with no kept version. Under K the first K writes drop
//!   nothing; write K + 1 drops version 0, which has no row, so it spills K
//!   rows; the next K - 1 writes drop versions that have theirs; write 2K + 1
//!   spills again. With K = 4, 12 writes spill at writes 5 and 9: 12 data
//!   installs and 2 row installs, 14 of each count.

use std::error::Error;

use haematite::SyncCounts;
use haematite::branch::KeptLimit;
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::json;

use super::{
    Captured, Hash, RETAIN_COMMITS, ROW_COMMITS, WINDOW_READS, after_rows, fence_fails, row_install,
};
use crate::serve::cli::Config;
use crate::serve::protocol::{
    BranchDiscard, Change, CollectionWrite, Fork, Mutation, Selection, Version,
};
use crate::serve::storage::{Storage, branch_name, version};

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

fn create(shard_count: usize, keep: usize) -> 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,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::reclaim(0),
        },
        NodeCacheBudget::Unlimited,
        KeptLimit::new(keep)?,
        fixture_long_wait_interval(),
    )?;
    Ok((directory, storage))
}

fn reopen(directory: &tempfile::TempDir, keep: usize) -> Result<Storage, Box<dyn Error>> {
    let path = directory.path().join("database");
    Ok(Storage::open_kept(
        &path,
        NodeCacheBudget::Unlimited,
        KeptLimit::new(keep)?,
        fixture_long_wait_interval(),
    )?)
}

fn head(storage: &Storage) -> Result<Version, Box<dyn Error>> {
    Ok(storage
        .refs
        .get(&branch_name("main"))
        .map(version)
        .ok_or("no main")?)
}

/// Write `index` on `main` at its head: a collection named for it.
fn mutation(storage: &Storage, index: usize) -> Result<Mutation, Box<dyn Error>> {
    Ok(Mutation {
        request_id: format!("declare-{index}"),
        branch: "main".to_owned(),
        expected: head(storage)?,
        change: Change::Collection(CollectionWrite {
            collection: format!("c{index}"),
            schema: None,
        }),
    })
}

fn write(storage: &mut Storage, index: usize) -> TestResult {
    let mutation = mutation(storage, index)?;
    storage.mutate(&mutation)?;
    Ok(())
}

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

fn reads() -> usize {
    WINDOW_READS.with(std::cell::Cell::get)
}

fn row_commits() -> usize {
    ROW_COMMITS.with(std::cell::Cell::get)
}

fn retain_commits() -> usize {
    RETAIN_COMMITS.with(std::cell::Cell::get)
}

/// A fork named `name` from `main` at `from`, asked at `main`'s head.
fn fork(storage: &Storage, name: &str, from: Version) -> Result<Mutation, Box<dyn Error>> {
    Ok(Mutation {
        request_id: format!("fork-{name}"),
        branch: "main".to_owned(),
        expected: head(storage)?,
        change: Change::Fork(Fork {
            name: name.to_owned(),
            from,
        }),
    })
}

/// A store at the default K whose `main` has moved twice; the version before
/// the head, kept by the record with no row.
fn moved_twice() -> Result<(tempfile::TempDir, Storage, Version), Box<dyn Error>> {
    let directory = tempfile::tempdir()?;
    let mut storage = Storage::create(
        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,
        fixture_long_wait_interval(),
    )?;
    write(&mut storage, 0)?;
    let previous = head(&storage)?;
    write(&mut storage, 1)?;
    Ok((directory, storage, previous))
}

/// `count` of each: installs, barrier, ref-file and directory flushes.
const fn each(count: u64) -> SyncCounts {
    SyncCounts {
        record_installs: count,
        barrier_device_flushes: count,
        ref_file_full_flushes: count,
        directory_full_flushes: count,
    }
}

fn captured_versions(storage: &Storage) -> Result<Vec<Version>, Box<dyn Error>> {
    Ok(storage
        .catalogue()
        .history("main", None, 100)?
        .into_iter()
        .map(|captured| captured.version)
        .collect())
}

/// Every version from 0 to the head is captured as itself, and the list
/// runs 0 to the head with no gap.
fn all_readable(storage: &Storage) -> TestResult {
    let head = head(storage)?;
    let expected: Vec<Version> = (0..=head.sequence)
        .map(|sequence| Version {
            generation: head.generation,
            sequence,
        })
        .collect();
    for at in &expected {
        let selection = Selection {
            branch: "main".to_owned(),
            version: Some(*at),
        };
        assert_eq!(storage.capture(&selection)?.version, *at);
    }
    assert_eq!(captured_versions(storage)?, expected);
    Ok(())
}

#[test]
fn twelve_writes_under_k_four_make_twelve_data_commits_and_two_row_commits() -> TestResult {
    let (_directory, mut storage) = create(1, 4)?;
    let before = counts(&storage);
    let read_before = reads();
    for index in 0..12 {
        write(&mut storage, index)?;
    }
    assert_eq!(counts(&storage).since(before), each(14));
    assert_eq!(reads() - read_before, 1);
    Ok(())
}

#[test]
fn one_write_dropping_a_version_with_no_row_commits_the_rows_first() -> TestResult {
    let (_directory, mut storage) = create(1, 4)?;
    for index in 0..4 {
        write(&mut storage, index)?;
    }
    let before = counts(&storage);
    write(&mut storage, 4)?;
    assert_eq!(counts(&storage).since(before), each(2));
    let before = counts(&storage);
    write(&mut storage, 5)?;
    assert_eq!(counts(&storage).since(before), each(1));
    Ok(())
}

#[test]
fn every_version_is_readable_at_every_step_over_two_shards() -> TestResult {
    let (_directory, mut storage) = create(2, 4)?;
    all_readable(&storage)?;
    for index in 0..12 {
        write(&mut storage, index)?;
        all_readable(&storage)?;
    }
    Ok(())
}

#[test]
fn four_shards_spill_in_one_install() -> TestResult {
    let (_directory, mut storage) = create(4, 4)?;
    let before = counts(&storage);
    for index in 0..12 {
        write(&mut storage, index)?;
    }
    assert_eq!(counts(&storage).since(before).record_installs, 14);
    all_readable(&storage)
}

/// A crash after the rows and before the data commit: the retry, in the
/// same process or after a reopen, finds the rows there and equal and makes
/// only the data commit.
#[test]
fn a_crash_between_the_rows_and_the_data_commit_writes_no_rows_again() -> TestResult {
    for reopened in [false, true] {
        let (directory, mut storage) = create(1, 4)?;
        for index in 0..4 {
            write(&mut storage, index)?;
        }
        let retry = mutation(&storage, 4)?;
        after_rows::arm();
        let before = counts(&storage);
        let crashed = storage
            .mutate(&retry)
            .err()
            .ok_or("the crash was not injected")?;
        assert_eq!(crashed.code, "injected_crash");
        assert_eq!(counts(&storage).since(before), each(1));
        if reopened {
            drop(storage);
            storage = reopen(&directory, 4)?;
        }
        let read_before = reads();
        let before = counts(&storage);
        storage.mutate(&retry)?;
        assert_eq!(counts(&storage).since(before), each(1));
        assert_eq!(reads() - read_before, usize::from(reopened));
        all_readable(&storage)?;
    }
    Ok(())
}

/// After the spill at write 5, writes 6 to 8 drop versions 1 to 3, which
/// have their rows: once reopened, they make no row commit, and what was
/// known is rebuilt by one range read.
#[test]
fn after_a_reopen_the_spilled_window_makes_no_row_commit() -> TestResult {
    let (directory, mut storage) = create(1, 4)?;
    for index in 0..5 {
        write(&mut storage, index)?;
    }
    drop(storage);
    let mut storage = reopen(&directory, 4)?;
    let read_before = reads();
    let before = counts(&storage);
    for index in 5..8 {
        write(&mut storage, index)?;
    }
    assert_eq!(counts(&storage).since(before), each(3));
    assert_eq!(reads() - read_before, 1);
    all_readable(&storage)
}

/// A store whose every version has its row, as each write left it before
/// `MD5a`, reopened under a smaller K: the first write drops versions that all
/// have rows, writes none, and every version stays readable.
#[test]
fn a_store_with_a_row_for_every_version_reopens_under_k_and_loses_none() -> TestResult {
    let (directory, mut storage) = create(1, 8)?;
    for index in 0..6 {
        let record = storage
            .refs
            .get(&branch_name("main"))
            .cloned()
            .ok_or("no main")?;
        storage.retain_history_row("main", &record)?;
        write(&mut storage, index)?;
    }
    drop(storage);
    let mut storage = reopen(&directory, 2)?;
    let before = counts(&storage);
    write(&mut storage, 6)?;
    assert_eq!(counts(&storage).since(before), each(1));
    all_readable(&storage)
}

fn config(kept: Option<u64>) -> Result<Config, serde_json::Error> {
    let mut value = json!({
        "data_dir": "/unused/database",
        "socket": "/unused/socket",
        "branch_cache_bytes_per_shard": 1,
        "max_frame": 1,
        "max_clients": 1,
        "max_page": 1,
        "timeout_ms": 1,
    });
    if let (Some(kept), Some(object)) = (kept, value.as_object_mut()) {
        object.insert("kept_versions".to_owned(), json!(kept));
    }
    serde_json::from_value(value)
}

#[test]
fn kept_versions_outside_one_to_the_most_is_refused_by_name_at_load() -> TestResult {
    for refused in [0, 4_097] {
        let error = config(Some(refused)).err().ok_or("loaded")?;
        assert!(error.to_string().contains("kept_versions"), "{error}");
    }
    assert_eq!(config(Some(4_096))?.kept_versions, 4_096);
    assert_eq!(config(None)?.kept_versions, super::KEPT_VERSIONS_DEFAULT);
    Ok(())
}

/// Fails while a fork reads a kept version's row before anyone wrote it: at
/// the default K, a fork from the version before the head writes that row in
/// one service commit, then forks.
#[test]
fn a_fork_from_the_version_before_the_head_writes_its_row_first() -> TestResult {
    let (_directory, mut storage, previous) = moved_twice()?;
    let commits_before = row_commits();
    let mutation = fork(&storage, "work", previous)?;
    storage.mutate(&mutation)?;
    assert_eq!(row_commits() - commits_before, 1);
    let work = Selection {
        branch: "work".to_owned(),
        version: None,
    };
    let at = storage.capture(&work)?;
    assert_eq!(storage.fork_provenance(&at)?.version, previous);
    Ok(())
}

/// A crash after the fork's row commit and before the fork: the retry, in
/// the same process or after a reopen, finds the row and commits no row again.
#[test]
fn a_crash_between_the_forks_row_and_the_fork_writes_no_row_again() -> TestResult {
    for reopened in [false, true] {
        let (directory, mut storage, previous) = moved_twice()?;
        let mutation = fork(&storage, "work", previous)?;
        after_rows::arm();
        let commits_before = row_commits();
        let crashed = storage
            .mutate(&mutation)
            .err()
            .ok_or("the crash was not injected")?;
        assert_eq!(crashed.code, "injected_crash");
        assert_eq!(row_commits() - commits_before, 1);
        if reopened {
            drop(storage);
            let path = directory.path().join("database");
            storage = Storage::open(
                &path,
                NodeCacheBudget::Unlimited,
                fixture_long_wait_interval(),
            )?;
        }
        let commits_before = row_commits();
        storage.mutate(&mutation)?;
        assert_eq!(row_commits() - commits_before, 0);
    }
    Ok(())
}

fn on(branch: &str, version: Option<Version>) -> Selection {
    Selection {
        branch: branch.to_owned(),
        version,
    }
}

/// Request `id` on `branch` at its head.
fn asked(
    storage: &Storage,
    id: &str,
    branch: &str,
    change: Change,
) -> Result<Mutation, Box<dyn Error>> {
    Ok(Mutation {
        request_id: id.to_owned(),
        branch: branch.to_owned(),
        expected: storage.capture(&on(branch, None))?.version,
        change,
    })
}

/// `work` forked from `main`'s head and written twice; its versions, oldest
/// first, and the discard of it.
fn work_to_discard(storage: &mut Storage) -> Result<(Vec<Version>, Mutation), Box<dyn Error>> {
    let change = Change::Fork(Fork {
        name: "work".to_owned(),
        from: head(storage)?,
    });
    let forked = asked(storage, "fork-work", "main", change)?;
    let mut versions = vec![storage.mutate(&forked)?.version];
    for index in 0..2 {
        let change = Change::Collection(CollectionWrite {
            collection: format!("w{index}"),
            schema: None,
        });
        let write = asked(storage, &format!("work-{index}"), "work", change)?;
        versions.push(storage.mutate(&write)?.version);
    }
    let change = Change::Discard(BranchDiscard {
        name: "work".to_owned(),
    });
    let discard = asked(storage, "discard-work", "work", change)?;
    Ok((versions, discard))
}

/// Fails while a discard loses the versions its record keeps: at the default
/// K, every version of the discarded generation is still captured, from its
/// row, after the name is made again.
#[test]
fn a_discard_writes_the_rows_of_the_versions_its_record_keeps() -> TestResult {
    let (_directory, mut storage, _previous) = moved_twice()?;
    let (versions, discard) = work_to_discard(&mut storage)?;
    storage.mutate(&discard)?;
    let change = Change::Fork(Fork {
        name: "work".to_owned(),
        from: head(&storage)?,
    });
    let again = asked(&storage, "fork-again", "main", change)?;
    storage.mutate(&again)?;
    for at in versions {
        assert_eq!(storage.capture(&on("work", Some(at)))?.version, at);
    }
    Ok(())
}

/// Each shard's id and head: what a version is, however it is captured.
fn roots(captured: &Captured) -> Vec<(usize, Hash)> {
    captured
        .shards
        .iter()
        .map(|shard| (shard.shard_id, shard.head))
        .collect()
}

/// Row commits of either kind made on this thread.
fn all_row_commits() -> usize {
    row_commits() + retain_commits()
}

/// A crash after the discard's one row commit: the retry commits no row
/// again (STOR-003 `MD5c`).
#[test]
fn a_crash_between_the_discards_rows_and_the_discard_writes_no_row_again() -> TestResult {
    let (_directory, mut storage, _previous) = moved_twice()?;
    let (_versions, discard) = work_to_discard(&mut storage)?;
    after_rows::arm();
    let commits_before = all_row_commits();
    let crashed = storage
        .mutate(&discard)
        .err()
        .ok_or("the crash was not injected")?;
    assert_eq!(crashed.code, "injected_crash");
    assert_eq!(all_row_commits() - commits_before, 1);
    let commits_before = all_row_commits();
    storage.mutate(&discard)?;
    assert_eq!(all_row_commits() - commits_before, 0);
    Ok(())
}

/// STOR-003 `MD5c`: fails while a discard makes two row commits. At K = 2,
/// `work`'s record keeps versions 0 and 1 with no rows and its head 2 has
/// none: the discard writes all three rows in one service commit, and every
/// version of the ended generation is captured from its row, exactly as it
/// was, after the name is made again.
#[test]
fn a_discard_with_a_full_window_and_no_rows_makes_one_row_commit() -> TestResult {
    let (_directory, mut storage) = create(1, 2)?;
    let (versions, discard) = work_to_discard(&mut storage)?;
    let mut before = Vec::new();
    for at in &versions {
        before.push(storage.capture(&on("work", Some(*at)))?);
    }
    let commits_before = all_row_commits();
    storage.mutate(&discard)?;
    assert_eq!(all_row_commits() - commits_before, 1);
    let change = Change::Fork(Fork {
        name: "work".to_owned(),
        from: head(&storage)?,
    });
    let again = asked(&storage, "fork-again", "main", change)?;
    storage.mutate(&again)?;
    for was in before {
        let captured = storage.capture(&on("work", Some(was.version)))?;
        assert_eq!(captured.version, was.version);
        assert_eq!(roots(&captured), roots(&was));
    }
    Ok(())
}

/// A full window with no rows (4 writes at K = 4), reopened at K = 1: the next
/// write drops all four kept versions, so it first writes every kept row in
/// one service commit, then its data commit, and nothing is lost.
#[test]
fn a_full_window_with_no_rows_reopened_at_k_one_writes_them_all_first() -> TestResult {
    let (directory, mut storage) = create(1, 4)?;
    for index in 0..4 {
        write(&mut storage, index)?;
    }
    drop(storage);
    let mut storage = reopen(&directory, 1)?;
    let commits_before = row_commits();
    let before = counts(&storage);
    write(&mut storage, 4)?;
    assert_eq!(counts(&storage).since(before), each(2));
    assert_eq!(row_commits() - commits_before, 1);
    all_readable(&storage)
}

/// Fails while only the oldest leaving version is asked about: after the
/// spill at write 5 (K = 4) the oldest kept version has its row and the
/// newest does not; reopened at K = 1 the next write drops all four, so the
/// newest must get its row first.
#[test]
fn a_smaller_k_asks_after_every_version_it_drops() -> TestResult {
    let (directory, mut storage) = create(1, 4)?;
    for index in 0..5 {
        write(&mut storage, index)?;
    }
    drop(storage);
    let mut storage = reopen(&directory, 1)?;
    let commits_before = row_commits();
    write(&mut storage, 5)?;
    assert_eq!(row_commits() - commits_before, 1);
    all_readable(&storage)
}

/// What a spill writes for a version is, byte for byte, what
/// `retain_history_row` wrote for it when it was the head, read back from the
/// store.
#[test]
fn a_spilled_row_is_the_row_retain_wrote_byte_for_byte() -> TestResult {
    let (_directory, mut storage) = create(1, 1)?;
    let first = storage
        .refs
        .get(&branch_name("main"))
        .cloned()
        .ok_or("no main")?;
    write(&mut storage, 0)?;
    write(&mut storage, 1)?;
    let made = super::kept_row("main", &first);
    let key = made.key.encode();
    let shard = storage.writer.shard_for(&key);
    let service = storage.refs.get(super::SERVICE).ok_or("no service")?;
    let root = service
        .shards
        .iter()
        .find(|held| held.shard_id == shard)
        .ok_or("no shard")?
        .head;
    let stored = storage.reader.get_at(shard, root, &key)?.ok_or("no row")?;
    assert_eq!(stored, made.encode_value()?);
    Ok(())
}

/// STOR-003 `MD5b`: fails while the row a fork from the head wrote is not
/// known. K = 1 and three writes (the spills at writes 2 and 3 load what is
/// known); a fork from the head, version 3, writes its row; then write 4
/// spills version 2 and write 5 drops version 3, which has its row: 1 row
/// commit, not the 2 made before `MD5b`.
#[test]
fn a_head_forked_from_is_not_written_again_when_it_leaves() -> TestResult {
    let (_directory, mut storage) = create(1, 1)?;
    for index in 0..3 {
        write(&mut storage, index)?;
    }
    let forked = fork(&storage, "work", head(&storage)?)?;
    storage.mutate(&forked)?;
    let commits_before = row_commits();
    for index in 3..5 {
        write(&mut storage, index)?;
    }
    assert_eq!(row_commits() - commits_before, 1);
    all_readable(&storage)
}

/// STOR-003 `MD5b`: a fork from a version other than the head writes no row
/// for the head; the head, once it leaves the window, is captured and forked
/// from as before.
#[test]
fn a_fork_from_another_version_writes_no_head_row() -> TestResult {
    let (_directory, mut storage) = create(1, 1)?;
    for index in 0..2 {
        write(&mut storage, index)?;
    }
    let then = head(&storage)?;
    let previous = Version {
        sequence: then.sequence - 1,
        ..then
    };
    let retains = retain_commits();
    let forked = fork(&storage, "work", previous)?;
    storage.mutate(&forked)?;
    assert_eq!(retain_commits() - retains, 0);
    for index in 2..4 {
        write(&mut storage, index)?;
    }
    all_readable(&storage)?;
    let again = fork(&storage, "again", then)?;
    storage.mutate(&again)?;
    Ok(())
}

/// How the fork from `work`'s head and the first discard of it go.
#[derive(Clone, Copy, PartialEq, Eq)]
enum Retained {
    /// The head's row commits and is fenced.
    Clean,
    /// The head's row commit installs, then fails as if its flush did.
    Unknown,
    /// As `Unknown`, and the first discard's fence then fails.
    FenceFails,
}

/// The directory flushes of the discard of `work` that succeeds, after a
/// fork from its head wrote the head's row as `retained` says.
fn discard_after_head_retained(retained: Retained) -> Result<u64, Box<dyn Error>> {
    let (_directory, mut storage) = create(1, 2)?;
    let (versions, _discard) = work_to_discard(&mut storage)?;
    let change = Change::Fork(Fork {
        name: "child".to_owned(),
        from: *versions.last().ok_or("no versions")?,
    });
    let child = asked(&storage, "fork-child", "work", change)?;
    if retained == Retained::Clean {
        storage.mutate(&child)?;
    } else {
        row_install::arm();
        let refused = storage
            .mutate(&child)
            .err()
            .ok_or("the failure was not injected")?;
        assert_eq!(refused.code, "history_row_durability_unknown");
    }
    let change = Change::Discard(BranchDiscard {
        name: "work".to_owned(),
    });
    let discard = asked(&storage, "discard-work", "work", change)?;
    if retained == Retained::FenceFails {
        fence_fails::arm();
        let refused = storage
            .mutate(&discard)
            .err()
            .ok_or("the fence failure was not injected")?;
        assert_eq!(refused.code, "history_row_durability_unknown");
        assert_eq!(
            storage.capture(&on("work", None))?.version,
            discard.expected
        );
    }
    let before = counts(&storage);
    storage.mutate(&discard)?;
    Ok(counts(&storage).since(before).directory_full_flushes)
}

/// STOR-003 `MD5c-r2`: fails while a discard relies on a head row whose commit
/// installed with its durability unknown. The discard fences the service
/// record before it reads that row: one directory flush more than the same
/// discard after a retain that did not fail.
#[test]
fn a_discard_after_a_retain_of_unknown_durability_fences_first() -> TestResult {
    let clean = discard_after_head_retained(Retained::Clean)?;
    let unknown = discard_after_head_retained(Retained::Unknown)?;
    assert_eq!(unknown, clean + 1);
    Ok(())
}

/// STOR-003 `MD5c-r3`: fails while a failed fence is forgotten. The first
/// discard refuses by name and removes nothing; the second fences again
/// before it relies on the head's row.
#[test]
fn a_fence_that_fails_is_owed_by_the_next_discard() -> TestResult {
    let clean = discard_after_head_retained(Retained::Clean)?;
    let owed = discard_after_head_retained(Retained::FenceFails)?;
    assert_eq!(owed, clean + 1);
    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
}