haematite-cli 0.21.8

#![forbid(unsafe_code)]

#[path = "support/history_files.rs"]
mod history_files;
#[path = "support/kept_refs.rs"]
mod kept_refs;

#[path = "../../haematite/tests/common/runtime_options.rs"]
mod runtime_options;

use runtime_options::runtime_options;

use std::error::Error;
use std::path::Path;

use haematite::branch::history::{Key, SERVICE_BRANCH};
use haematite::branch::{
    CommitDurability, CommitRequest, KeptLimit, RefRecord, RefStore, Registry, current_timestamp,
    open,
};
use haematite::db::branches::BranchDatabase;
use haematite::db::vacuum::{CANONICAL_REFS_DIR, Options, sweep};
use haematite::db::{Database, DatabaseConfig};
use haematite::store::NodeCacheBudget;
use haematite::tree::Hash;
use haematite_cli::serve::access::{self, Standalone};
use haematite_cli::serve::protocol::{
    BranchDiscard, Change, CollectionWrite, DocumentWrite, Fork, Mutation, ReceiptQuery, Selection,
    Version,
};
use haematite_cli::serve::storage::{Receipt, Storage, reads};
use serde_json::value::RawValue;

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

/// STOR-003 MD5a-t: these fixtures fork from a version's row, so each keeps
/// one previous version: every version below head - 1 has its row, from
/// ordinary writes.
const KEEP: usize = 1;

fn setup() -> 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: 2,
            distributed: None,
            executor_threads: Some(2),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::KeepAll,
        },
        NodeCacheBudget::Unlimited,
        KeptLimit::new(KEEP)?,
        long_wait_interval(),
    )?;
    Ok((directory, storage))
}

fn selected(branch: &str) -> Selection {
    Selection {
        branch: branch.to_owned(),
        version: None,
    }
}

fn request(
    storage: &Storage,
    branch: &str,
    id: &str,
    change: Change,
) -> Result<Mutation, Box<dyn Error>> {
    Ok(Mutation {
        request_id: id.to_owned(),
        branch: branch.to_owned(),
        expected: storage.capture(&selected(branch))?.version,
        change,
    })
}

fn apply(
    storage: &mut Storage,
    branch: &str,
    id: &str,
    change: Change,
) -> Result<Receipt, Box<dyn Error>> {
    let mutation = request(storage, branch, id, change)?;
    Ok(storage.mutate(&mutation)?)
}

fn put(value: &str) -> Result<Change, Box<dyn Error>> {
    Ok(Change::Put(DocumentWrite {
        collection: "records".to_owned(),
        document_id: "one".to_owned(),
        document: RawValue::from_string(value.to_owned())?,
    }))
}

fn seed(storage: &mut Storage) -> Result<Version, Box<dyn Error>> {
    apply(
        storage,
        "main",
        "declare",
        Change::Collection(CollectionWrite {
            collection: "records".to_owned(),
            schema: None,
        }),
    )?;
    let old = apply(storage, "main", "old", put(r#"{"value":"old"}"#)?)?.version;
    apply(storage, "main", "new", put(r#"{"value":"new"}"#)?)?;
    // One more, so `old` is below head - 1 and has its row (MD5a-t, K = 1).
    apply(
        storage,
        "main",
        "later",
        Change::Collection(CollectionWrite {
            collection: "later".to_owned(),
            schema: None,
        }),
    )?;
    Ok(old)
}

fn fork_request(
    storage: &Storage,
    branch: &str,
    name: &str,
    id: &str,
    from: Version,
) -> Result<Mutation, Box<dyn Error>> {
    request(
        storage,
        branch,
        id,
        Change::Fork(Fork {
            name: name.to_owned(),
            from,
        }),
    )
}

fn history_name(branch: &str, version: Version) -> String {
    format!(
        "@haem/history/{}/{}/{:020}",
        Hash::of(branch.as_bytes()),
        version.generation,
        version.sequence
    )
}

fn remove_history(data: &Path, branch: &str, version: Version) -> TestResult {
    history_files::create(data, KEEP, Some((branch, version)))?;
    let mut refs = RefStore::open(data.join(CANONICAL_REFS_DIR))?;
    assert!(refs.remove(&history_name(branch, version))?.is_some());
    Ok(())
}

fn references(data: &Path) -> Result<Vec<RefRecord>, Box<dyn Error>> {
    let refs = RefStore::open(data.join(CANONICAL_REFS_DIR))?;
    Ok(refs.list_prefix("").cloned().collect())
}

fn document(storage: &Storage, branch: &str) -> Result<String, Box<dyn Error>> {
    let at = storage.capture(&selected(branch))?;
    Ok(reads::document(
        &access::request(&Standalone)?,
        &storage.reader()?,
        &at,
        "records",
        "one",
    )?
    .ok_or("missing document")?
    .raw_json)
}

fn overwrite_row(data: &Path, branch: &str, version: Version, value: &[u8]) -> TestResult {
    let mut writer = BranchDatabase::new(
        Database::open(data, runtime_options())?,
        NodeCacheBudget::Unlimited,
    )?;
    let mut refs = kept_refs::open_refs(data, KEEP)?;
    let registry = Registry::new();
    let service = open(SERVICE_BRANCH, &refs, &registry)?;
    let key = Key {
        branch: Hash::of(branch.as_bytes()),
        generation: version.generation,
        sequence: version.sequence,
    }
    .encode();
    service.put(writer.shard_for(&key), key, value)?;
    writer.commit(
        &service,
        &registry,
        CommitRequest {
            durability: CommitDurability::Durable { refs: &mut refs },
            extra_parents: &[],
            timestamp: current_timestamp(),
        },
    )?;
    Ok(())
}

#[test]
fn kept_row_fork_works_after_its_history_file_is_removed() -> TestResult {
    let (directory, mut storage) = setup()?;
    let old = seed(&mut storage)?;
    drop(storage);
    let data = directory.path().join("database");
    remove_history(&data, "main", old)?;
    let mut storage = Storage::open(&data, NodeCacheBudget::Unlimited, long_wait_interval())?;
    let mutation = fork_request(&storage, "main", "work", "fork", old)?;
    let receipt = storage.mutate(&mutation)?;
    assert_eq!(document(&storage, "work")?, r#"{"value":"old"}"#);
    assert_eq!(document(&storage, "main")?, r#"{"value":"new"}"#);
    let at = storage.capture(&selected("work"))?;
    let provenance = storage.fork_provenance(&at)?;
    assert_eq!(provenance.branch, "main");
    assert_eq!(provenance.version, old);
    assert_eq!(storage.mutate(&mutation)?.version, receipt.version);
    let refs = RefStore::open(data.join(CANONICAL_REFS_DIR))?;
    assert!(refs.get(&history_name("main", old)).is_none());
    Ok(())
}

#[test]
fn earlier_generation_row_forks_after_the_logical_name_is_recreated() -> TestResult {
    let (directory, mut storage) = setup()?;
    let old = seed(&mut storage)?;
    let first = fork_request(&storage, "main", "work", "first-work", old)?;
    let first = storage.mutate(&first)?.version;
    apply(
        &mut storage,
        "work",
        "discard-work",
        Change::Discard(BranchDiscard {
            name: "work".to_owned(),
        }),
    )?;
    let later = apply(
        &mut storage,
        "main",
        "replacement",
        put(r#"{"value":"replacement"}"#)?,
    )?
    .version;
    let recreate = fork_request(&storage, "main", "work", "recreate-work", later)?;
    let recreated = storage.mutate(&recreate)?.version;
    assert_ne!(first.generation, recreated.generation);
    drop(storage);
    let data = directory.path().join("database");
    remove_history(&data, "work", first)?;
    let mut storage = Storage::open(&data, NodeCacheBudget::Unlimited, long_wait_interval())?;
    let mutation = fork_request(&storage, "work", "earlier", "earlier-work", first)?;
    storage.mutate(&mutation)?;
    assert_eq!(document(&storage, "earlier")?, r#"{"value":"old"}"#);
    assert_eq!(document(&storage, "work")?, r#"{"value":"replacement"}"#);
    let at = storage.capture(&selected("earlier"))?;
    let provenance = storage.fork_provenance(&at)?;
    assert_eq!(provenance.branch, "work");
    assert_eq!(provenance.version, first);
    assert!(
        RefStore::open(data.join(CANONICAL_REFS_DIR))?
            .get(&history_name("work", first))
            .is_none()
    );
    Ok(())
}

#[test]
fn corrupt_kept_row_refuses_by_name_without_installing_any_reference() -> TestResult {
    let (directory, mut storage) = setup()?;
    let old = seed(&mut storage)?;
    drop(storage);
    let data = directory.path().join("database");
    overwrite_row(&data, "main", old, b"invalid")?;
    let before = references(&data)?;
    let mut storage = Storage::open(&data, NodeCacheBudget::Unlimited, long_wait_interval())?;
    let mutation = fork_request(&storage, "main", "work", "fork-invalid", old)?;
    let error = storage
        .mutate(&mutation)
        .err()
        .ok_or("corrupt row accepted")?;
    assert_eq!(error.code, "history_row_corrupt");
    assert_eq!(error.message, "history_value_truncated");
    assert_eq!(references(&data)?, before);
    assert_eq!(
        storage
            .capture(&selected("work"))
            .err()
            .ok_or("destination became visible")?
            .code,
        "branch_missing"
    );
    assert_eq!(
        storage.capture(&selected("main"))?.version,
        mutation.expected
    );
    Ok(())
}

#[test]
fn row_fork_data_provenance_and_receipt_survive_vacuum_without_history_files() -> TestResult {
    let (directory, mut storage) = setup()?;
    let old = seed(&mut storage)?;
    drop(storage);
    let data = directory.path().join("database");
    remove_history(&data, "main", old)?;
    let mut storage = Storage::open(&data, NodeCacheBudget::Unlimited, long_wait_interval())?;
    let mutation = fork_request(&storage, "main", "work", "fork-vacuum", old)?;
    let receipt = storage.mutate(&mutation)?;
    drop(storage);
    history_files::create(&data, KEEP, None)?;
    let mut refs = RefStore::open(data.join(CANONICAL_REFS_DIR))?;
    let histories: Vec<_> = refs
        .list_prefix("@haem/history/")
        .map(|record| record.name.clone())
        .collect();
    for name in histories {
        assert!(refs.remove(&name)?.is_some());
    }
    assert!(refs.list_prefix("@haem/history/").next().is_none());
    drop(refs);
    let mut options = Options::new(data.clone());
    options.attest_metadata_complete = true;
    sweep(&options, true)?;
    let mut storage = Storage::open(&data, NodeCacheBudget::Unlimited, long_wait_interval())?;
    assert_eq!(document(&storage, "work")?, r#"{"value":"old"}"#);
    let at = storage.capture(&selected("work"))?;
    let provenance = storage.fork_provenance(&at)?;
    assert_eq!(provenance.branch, "main");
    assert_eq!(provenance.version, old);
    let confirmed = storage
        .receipt(&ReceiptQuery {
            branch: "work".to_owned(),
            request_id: mutation.request_id.clone(),
        })?
        .ok_or("fork receipt missing")?;
    assert_eq!(confirmed.version, receipt.version);
    assert_eq!(confirmed.fingerprint, receipt.fingerprint);
    assert_eq!(storage.mutate(&mutation)?.version, receipt.version);
    assert!(
        RefStore::open(data.join(CANONICAL_REFS_DIR))?
            .list_prefix("@haem/history/")
            .next()
            .is_none()
    );
    Ok(())
}

/// The predecessor can be retained without a durable row. A fork must write
/// that row before it can publish a reference to the retained version.
#[test]
fn a_fork_from_the_version_before_the_head_forks_at_k_one() -> TestResult {
    let (_directory, mut storage) = setup()?;
    apply(
        &mut storage,
        "main",
        "declare",
        Change::Collection(CollectionWrite {
            collection: "records".to_owned(),
            schema: None,
        }),
    )?;
    let old = apply(&mut storage, "main", "old", put(r#"{"value":"old"}"#)?)?.version;
    apply(&mut storage, "main", "new", put(r#"{"value":"new"}"#)?)?;
    let mutation = fork_request(&storage, "main", "work", "fork", old)?;
    storage.mutate(&mutation)?;
    assert_eq!(document(&storage, "work")?, r#"{"value":"old"}"#);
    let at = storage.capture(&selected("work"))?;
    assert_eq!(storage.fork_provenance(&at)?.version, old);
    Ok(())
}

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