haematite-cli 0.21.8

//! STOR-003 MD4: a version the branch record keeps is captured from its kept
//! heads with no history row, and the history list runs across the kept
//! window and the rows below it with no gap.
//!
//! Since STOR-003 `MD5a` a mutation writes no row of its own: the rows of the
//! kept versions are written in one service commit when the oldest of them
//! would leave the window without one. A test that needs a version with no
//! row takes the row away in one service commit, as `history_tests`'
//! release does.

use std::error::Error;

use haematite::branch::KeptLimit;
use haematite::branch::history::Key;
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;

use super::{
    Captured, CommitDurability, CommitRequest, Failure, Hash, SERVICE, Storage, capture_kept,
    current_timestamp, open_branch,
};
use crate::serve::protocol::{
    BranchDiscard, Change, CollectionWrite, Fork, Mutation, Selection, Version,
};

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

fn main_at(version: Option<Version>) -> Selection {
    at("main", version)
}

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

/// Request `id`: a fork named `name` from `branch` at `from`, asked on
/// `branch`'s head.
fn fork(
    storage: &Storage,
    id: &str,
    branch: &str,
    name: &str,
    from: Version,
) -> Result<Mutation, Failure> {
    Ok(Mutation {
        request_id: id.to_owned(),
        branch: branch.to_owned(),
        expected: storage.capture(&at(branch, None))?.version,
        change: Change::Fork(Fork {
            name: name.to_owned(),
            from,
        }),
    })
}

/// A store keeping `keep` previous versions whose `main` has moved `writes`
/// times; every version as it was captured while it was the head, oldest
/// first, the head last.
fn moved(
    keep: usize,
    writes: usize,
) -> Result<(tempfile::TempDir, Storage, Vec<Captured>), Box<dyn Error>> {
    let directory = tempfile::tempdir()?;
    // The storage's own limit, so the rows a version needs before it leaves
    // the window are written as the service writes them (STOR-003 `MD5a`).
    let mut storage = Storage::create_kept(
        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,
        KeptLimit::new(keep)?,
        fixture_long_wait_interval(),
    )?;
    let mut heads = vec![storage.capture(&main_at(None))?];
    for index in 0..writes {
        let mutation = Mutation {
            request_id: format!("declare-{index}"),
            branch: "main".to_owned(),
            expected: storage.capture(&main_at(None))?.version,
            change: Change::Collection(CollectionWrite {
                collection: format!("c{index}"),
                schema: None,
            }),
        };
        storage.mutate(&mutation)?;
        heads.push(storage.capture(&main_at(None))?);
    }
    Ok((directory, storage, heads))
}

/// One service commit that takes `main`'s rows for `taken`.
fn take_rows(storage: &mut Storage, taken: &[&Captured]) -> TestResult {
    let versions: Vec<Version> = taken.iter().map(|captured| captured.version).collect();
    take_rows_of(storage, "main", &versions)
}

/// One service commit that takes `branch`'s rows for `versions`.
fn take_rows_of(storage: &mut Storage, branch: &str, versions: &[Version]) -> TestResult {
    let service =
        open_branch(SERVICE, &storage.refs, &storage.registry).map_err(Failure::branch_commit)?;
    for version in versions {
        let key = Key {
            branch: Hash::of(branch.as_bytes()),
            generation: version.generation,
            sequence: version.sequence,
        }
        .encode();
        service
            .delete(storage.writer.shard_for(&key), key)
            .map_err(Failure::branch)?;
    }
    storage
        .writer
        .commit(
            &service,
            &storage.registry,
            CommitRequest {
                durability: CommitDurability::Durable {
                    refs: &mut storage.refs,
                },
                extra_parents: &[],
                timestamp: current_timestamp(),
            },
        )
        .map_err(Failure::branch_commit)?;
    storage.publish_record(SERVICE)?;
    Ok(())
}

/// Each shard's id, anchor and head, so two captures compare whole.
fn roots(captured: &Captured) -> Vec<(usize, Hash, Hash)> {
    captured
        .shards
        .iter()
        .map(|shard| (shard.shard_id, shard.fork_anchor, shard.head))
        .collect()
}

/// As a row of `head` captures it: anchor and head its root on each shard.
fn as_kept(head: &Captured) -> Vec<(usize, Hash, Hash)> {
    head.shards
        .iter()
        .map(|shard| (shard.shard_id, shard.head, shard.head))
        .collect()
}

#[test]
fn the_previous_version_is_captured_from_kept_with_no_row() -> TestResult {
    let (_directory, mut storage, heads) = moved(3, 2)?;
    let previous = heads.get(1).ok_or("too few versions")?;
    take_rows(&mut storage, &[previous])?;

    let captured = storage.capture(&main_at(Some(previous.version)))?;
    assert_eq!(captured.version, previous.version);
    assert_eq!(roots(&captured), as_kept(previous));
    Ok(())
}

/// K = 2 and three writes: the third would drop version 0 with no row, so
/// versions 0 and 1 get their rows first; version 1 is then both kept and a
/// row (STOR-003 `MD5a`).
#[test]
fn a_version_both_kept_and_a_row_answers_the_same_either_way() -> TestResult {
    let (_directory, storage, heads) = moved(2, 3)?;
    let previous = heads.get(1).ok_or("too few versions")?;
    let service = storage.refs.get(SERVICE).ok_or("no service record")?;

    let from_kept = storage.capture(&main_at(Some(previous.version)))?;
    let from_row =
        capture_kept(&storage.reader, service, "main", previous.version)?.ok_or("no row")?;
    assert_eq!(from_kept.version, from_row.version);
    assert_eq!(from_kept.branch, from_row.branch);
    assert_eq!(roots(&from_kept), roots(&from_row));
    Ok(())
}

/// K = 2 keeps versions 3 and 4 with their rows taken away; versions 0, 1
/// and 2 have only their rows. The list runs 0 to the head 5 in order.
#[test]
fn history_across_k_kept_and_three_rows_is_contiguous_and_in_order() -> TestResult {
    let (_directory, mut storage, heads) = moved(2, 5)?;
    let window: Vec<&Captured> = heads.iter().skip(3).take(2).collect();
    take_rows(&mut storage, &window)?;

    let listed: Vec<Version> = storage
        .catalogue()
        .history("main", None, 100)?
        .into_iter()
        .map(|captured| captured.version)
        .collect();
    let expected: Vec<Version> = heads.iter().map(|head| head.version).collect();
    assert_eq!(listed, expected);
    let paged: Vec<Version> = storage
        .catalogue()
        .history("main", heads.get(2).map(|head| head.version), 2)?
        .into_iter()
        .map(|captured| captured.version)
        .collect();
    assert_eq!(
        paged,
        expected.get(3..5).ok_or("too few versions")?.to_vec()
    );
    Ok(())
}

/// `work` is forked, discarded (its discard writes a row for its last
/// version) and forked again. With that row taken away, the discarded
/// generation's version has ended: capture and a fork from it both refuse
/// `version_generation_ended`, never `version_missing`.
#[test]
fn an_ended_generation_with_no_row_is_refused_by_that_name() -> TestResult {
    let (_directory, mut storage, heads) = moved(2, 1)?;
    let head = heads.last().ok_or("no versions")?.version;
    storage.mutate(&fork(&storage, "fork-work", "main", "work", head)?)?;
    let ended = storage.capture(&at("work", None))?.version;
    storage.mutate(&Mutation {
        request_id: "discard-work".to_owned(),
        branch: "work".to_owned(),
        expected: ended,
        change: Change::Discard(BranchDiscard {
            name: "work".to_owned(),
        }),
    })?;
    storage.mutate(&fork(&storage, "recreate-work", "main", "work", head)?)?;
    take_rows_of(&mut storage, "work", &[ended])?;

    let captured = storage
        .capture(&at("work", Some(ended)))
        .err()
        .ok_or("an ended version with no row was captured")?;
    assert_eq!(
        captured.code, "version_generation_ended",
        "{}",
        captured.message
    );
    let forked = storage
        .mutate(&fork(&storage, "fork-ended", "work", "again", ended)?)
        .err()
        .ok_or("a fork from an ended version with no row was made")?;
    assert_eq!(
        forked.code, "version_generation_ended",
        "{}",
        forked.message
    );
    Ok(())
}

/// STOR-003 MD4-b: K = 2 keeps versions 2 and 3 with their rows taken away.
/// Dropped and opened again, the storage captures each from the heads the
/// record keeps, and lists 0 to the head 4 in order.
#[test]
fn kept_versions_are_captured_and_listed_after_the_storage_is_opened_again() -> TestResult {
    let (directory, mut storage, heads) = moved(2, 4)?;
    let window: Vec<&Captured> = heads.iter().skip(2).take(2).collect();
    take_rows(&mut storage, &window)?;
    drop(storage);

    let path = directory.path().join("database");
    let reopened = Storage::open(
        &path,
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    for kept in &window {
        let captured = reopened.capture(&main_at(Some(kept.version)))?;
        assert_eq!(captured.version, kept.version);
        assert_eq!(roots(&captured), as_kept(kept));
    }
    let listed: Vec<Version> = reopened
        .catalogue()
        .history("main", None, 100)?
        .into_iter()
        .map(|captured| captured.version)
        .collect();
    let expected: Vec<Version> = heads.iter().map(|head| head.version).collect();
    assert_eq!(listed, expected);
    Ok(())
}

/// STOR-003 `MD5a-t`: K = 4 and five writes spill rows for versions 0 to 3 and
/// keep 1 to 4. Version 3, inside the window, has its row taken away and is
/// served the same as with it.
#[test]
fn a_kept_version_whose_row_is_taken_is_served_the_same() -> TestResult {
    let (_directory, mut storage, heads) = moved(4, 5)?;
    let inside = heads.get(3).ok_or("too few versions")?;
    let before = storage.capture(&main_at(Some(inside.version)))?;
    take_rows(&mut storage, &[inside])?;
    let after = storage.capture(&main_at(Some(inside.version)))?;
    assert_eq!(after.version, before.version);
    assert_eq!(roots(&after), roots(&before));
    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
}