haematite-cli 0.21.8

//! STOR-003 HR2: a branch's released mark. A version at or below it is
//! refused as `version_released`; the history list starts after it, and a
//! gap above it is still a fault.
//!
//! A release is stood in for here by one service commit that takes the
//! oldest rows and writes the mark together, as the release's last commit
//! will.

use std::error::Error;

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

use super::{
    CommitDurability, CommitRequest, Failure, Hash, Released, SERVICE, Storage, current_timestamp,
    open_branch, released_key,
};
use crate::serve::protocol::{Change, CollectionWrite, Mutation, Selection, Version};

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

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

/// A store whose `main` has moved `writes` times; every version, oldest
/// first, the head last. It keeps one previous version (STOR-003 MD5a-t), so
/// every version below head - 1 has its row, from ordinary writes.
fn moved(writes: 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: 2,
            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 mut versions = vec![storage.capture(&main_at(None))?.version];
    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,
            }),
        };
        versions.push(storage.mutate(&mutation)?.version);
    }
    Ok((directory, storage, versions))
}

/// One service commit that takes `main`'s rows for `taken` and writes
/// `mark`, if any, as `main`'s released mark.
fn release(storage: &mut Storage, taken: &[Version], mark: Option<&[u8]>) -> TestResult {
    let service =
        open_branch(SERVICE, &storage.refs, &storage.registry).map_err(Failure::branch_commit)?;
    for version in taken {
        let key = Key {
            branch: Hash::of(b"main"),
            generation: version.generation,
            sequence: version.sequence,
        }
        .encode();
        service
            .delete(storage.writer.shard_for(&key), key)
            .map_err(Failure::branch)?;
    }
    if let Some(mark) = mark {
        let key = released_key("main");
        service
            .put(storage.writer.shard_for(&key), key, mark)
            .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(())
}

fn mark(version: Version) -> Result<Vec<u8>, Box<dyn Error>> {
    Ok(serde_json::to_vec(&Released {
        version,
        at: 1_759_600_000_000_000_000,
    })?)
}

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

/// Fails while a released version is answered as missing, or read: the
/// versions at and below the mark are refused as `version_released`, naming
/// the version and the mark; the one above it is still captured.
#[test]
fn a_version_at_or_below_the_mark_is_refused_as_released() -> TestResult {
    let (_directory, mut storage, versions) = moved(4)?;
    let (taken, kept) = versions.split_at(2);
    let last = *taken.last().ok_or("nothing taken")?;
    release(&mut storage, taken, Some(&mark(last)?))?;
    for version in taken {
        let refused = storage
            .capture(&main_at(Some(*version)))
            .err()
            .ok_or("a released version was captured")?;
        assert_eq!(refused.code, "version_released", "{}", refused.message);
        let named = format!("version {}.{}", version.generation, version.sequence);
        assert!(refused.message.contains(&named), "{}", refused.message);
    }
    let first_kept = *kept.first().ok_or("nothing kept")?;
    assert_eq!(
        storage.capture(&main_at(Some(first_kept)))?.version,
        first_kept
    );
    Ok(())
}

/// Fails while what was released reads as a gap: the history list starts
/// just after the mark and runs whole to the head.
#[test]
fn the_history_list_starts_after_the_mark() -> TestResult {
    let (_directory, mut storage, versions) = moved(4)?;
    let (taken, kept) = versions.split_at(2);
    let last = *taken.last().ok_or("nothing taken")?;
    release(&mut storage, taken, Some(&mark(last)?))?;
    assert_eq!(listed(&storage)?, kept);
    Ok(())
}

/// Fails while the mark hides a gap it does not cover: a row missing above
/// the mark is still `version_missing`, in the list and in a capture.
#[test]
fn a_gap_above_the_mark_is_still_missing() -> TestResult {
    let (_directory, mut storage, versions) = moved(4)?;
    let first = *versions.first().ok_or("no versions")?;
    let above = *versions.get(2).ok_or("too few versions")?;
    release(&mut storage, &[first], Some(&mark(first)?))?;
    release(&mut storage, &[above], None)?;
    let refused = listed(&storage).err().ok_or("a gap was listed")?;
    assert_eq!(refused.code, "version_missing", "{}", refused.message);
    let missing = storage
        .capture(&main_at(Some(above)))
        .err()
        .ok_or("a missing version was captured")?;
    assert_eq!(missing.code, "version_missing", "{}", missing.message);
    Ok(())
}

/// Fails while a mark that does not decode is taken for none: a capture
/// below where it would reach answers `history_marker_corrupt`.
#[test]
fn a_mark_that_does_not_decode_is_refused_by_name() -> TestResult {
    let (_directory, mut storage, versions) = moved(2)?;
    let first = *versions.first().ok_or("no versions")?;
    release(&mut storage, &[first], Some(b"{\"version\":"))?;
    let refused = storage
        .capture(&main_at(Some(first)))
        .err()
        .ok_or("a version was captured past a corrupt mark")?;
    assert_eq!(
        refused.code, "history_marker_corrupt",
        "{}",
        refused.message
    );
    Ok(())
}

/// Fails while the mark is compared by sequence alone: (generation,
/// sequence) is one order, generation first.
#[test]
fn the_mark_orders_generation_before_sequence() {
    let at = |generation, sequence| Version {
        generation,
        sequence,
    };
    let released = Released {
        version: at(5, 3),
        at: 0,
    };
    for (version, covered) in [
        (at(5, 3), true),
        (at(5, 2), true),
        (at(4, 9), true),
        (at(5, 4), false),
        (at(6, 0), false),
    ] {
        assert_eq!(released.covers(version), covered, "{version:?}");
    }
}

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
}