haematite-cli 0.21.0

//! STOR-003 MD2, as `MD5a` changes it: what one served write costs, as counts.
//!
//! The expected numbers are derived from the source before any run, from
//! `Storage::mutate` (mutations.rs) over the engine's commit (MD1's derivation):
//! - a write keeps no row of its own (`MD5a`): `rows_before_advance` writes
//!   nothing while the advance drops no kept version without a row, which a
//!   new store's first K writes never do (`history_spill_tests` counts a spill);
//! - the data and its receipt in one durable commit on the branch: 1 record
//!   install, 1 barrier device flush for each shard written, 1 ref-file full
//!   flush, 1 directory full flush.
//!
//! On a store of one shard one write is therefore 1 install and 3 full
//! flushes: 1 barrier, 1 ref-file, 1 directory (2 + d, d = 1). A retry of a
//! request already applied answers its receipt with no commit, but first
//! re-fences the branch record its receipt was read at (`receipt` →
//! `confirm_version`, mutations.rs), one directory full flush. A row already
//! there for the current version is not read or re-fenced by the write.

use std::error::Error;

use haematite::SyncCounts;
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::value::RawValue;

use crate::serve::protocol::{Change, CollectionWrite, DocumentWrite, Mutation, Version};
use crate::serve::storage::{Storage, branch_name, version};

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

/// A store of one shard holding the collection `notes`, and the counts after
/// making it.
fn create() -> Result<(tempfile::TempDir, Storage), 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(),
    )?;
    apply(&mut storage, "notes", collection("notes"))?;
    Ok((directory, storage))
}

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

/// `change` applied to `main` at its head as request `request_id`.
fn apply(storage: &mut Storage, request_id: &str, change: Change) -> TestResult {
    let expected = head(storage)?;
    storage.mutate(&Mutation {
        request_id: request_id.to_owned(),
        branch: "main".to_owned(),
        expected,
        change,
    })?;
    Ok(())
}

fn collection(name: &str) -> Change {
    Change::Collection(CollectionWrite {
        collection: name.to_owned(),
        schema: None,
    })
}

fn document(id: &str) -> Result<Change, Box<dyn Error>> {
    Ok(Change::Put(DocumentWrite {
        collection: "notes".to_owned(),
        document_id: id.to_owned(),
        document: RawValue::from_string(r#"{"body":"text"}"#.to_owned())?,
    }))
}

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

/// One write: one commit, 1 install and 3 full flushes on one shard.
const ONE_WRITE: SyncCounts = SyncCounts {
    record_installs: 1,
    barrier_device_flushes: 1,
    ref_file_full_flushes: 1,
    directory_full_flushes: 1,
};

/// Fails while one served write costs other than derived: a document.put
/// on one shard makes 1 install, 1 barrier, 1 ref-file and 1 directory
/// full flush.
#[test]
fn one_document_put_costs_one_commit_and_three_flushes() -> TestResult {
    let (_directory, mut storage) = create()?;
    let before = counts(&storage);
    apply(&mut storage, "put-a", document("a")?)?;
    assert_eq!(counts(&storage).since(before), ONE_WRITE);
    Ok(())
}

/// Fails while a retry costs other than derived: the same request again
/// answers its receipt with no commit and one directory full flush, the
/// re-fence of the record its receipt was read at.
#[test]
fn a_retry_of_an_applied_request_commits_nothing_and_re_fences_once() -> TestResult {
    let (_directory, mut storage) = create()?;
    let expected = head(&storage)?;
    let request = Mutation {
        request_id: "put-a".to_owned(),
        branch: "main".to_owned(),
        expected,
        change: document("a")?,
    };
    storage.mutate(&request)?;
    let before = counts(&storage);
    storage.mutate(&request)?;
    let expected = SyncCounts {
        directory_full_flushes: 1,
        ..SyncCounts::default()
    };
    assert_eq!(counts(&storage).since(before), expected);
    Ok(())
}

/// Fails while a write reads or re-fences the current version's row: with
/// that row already there, a write makes the data commit only.
#[test]
fn a_write_whose_row_is_kept_already_commits_once() -> TestResult {
    let (_directory, mut storage) = create()?;
    let record = storage
        .refs
        .get(&branch_name("main"))
        .cloned()
        .ok_or("no main")?;
    storage.retain_history_row("main", &record)?;
    let before = counts(&storage);
    apply(&mut storage, "put-a", document("a")?)?;
    assert_eq!(counts(&storage).since(before), ONE_WRITE);
    Ok(())
}

/// Fails while a collection.put costs other than a document.put: the same
/// one commit and three flushes.
#[test]
fn one_collection_put_costs_the_same() -> TestResult {
    let (_directory, mut storage) = create()?;
    let before = counts(&storage);
    apply(&mut storage, "letters", collection("letters"))?;
    assert_eq!(counts(&storage).since(before), ONE_WRITE);
    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
}