haematite-cli 0.35.0

//! STOR-050 R2 in storage: a line keeps every version it commits past the
//! kept window and the keep settings, until a discard names it; an ordinary
//! branch keeps its old versions only until a release under those settings.

use std::error::Error;
use std::num::{NonZeroU64, NonZeroUsize};
use std::path::Path;

use haematite::branch::{KeptLimit, current_timestamp};
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::{Value, json};

use super::history::{KEPT_VERSIONS_DEFAULT, KEPT_VERSIONS_MOST, oldest_rows};
use super::release::{IntentListing, Keep, Rules};
use super::{Failure, Receipt, SERVICE, Storage};
use crate::serve::protocol::{Mutation, Selection, Version};

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

/// How many stage commits the line makes: the fewest past the 4,096 ceiling
/// of the kept window that also make a discard release more than one page of
/// 4,096 rows, so the release loop runs twice. Each commit is one device
/// flush, so this is the smallest N that proves both.
const STAGES: u64 = 4_100;

fn create(kept: 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: 1,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::reclaim(0),
        },
        NodeCacheBudget::Unlimited,
        KeptLimit::new(kept)?,
        std::time::Duration::from_secs(1),
    )?;
    Ok((directory, storage))
}

fn head(storage: &Storage, branch: &str) -> Result<Version, Failure> {
    Ok(storage
        .capture(&Selection {
            branch: branch.to_owned(),
            version: None,
        })?
        .version)
}

/// `params` as the wire sends them, decoded as the service decodes them.
fn apply(storage: &mut Storage, params: Value) -> Result<Receipt, Box<dyn Error>> {
    let request: Mutation = serde_json::from_value(params)?;
    Ok(storage.mutate(&request)?)
}

fn change(
    storage: &mut Storage,
    id: &str,
    branch: &str,
    change: Value,
) -> Result<Receipt, Box<dyn Error>> {
    let expected = head(storage, branch)?;
    let mut params = json!({"request_id": id, "branch": branch, "expected": expected});
    params["change"] = change;
    apply(storage, params)
}

fn fork(
    storage: &mut Storage,
    from_branch: &str,
    name: &str,
    from: Version,
    line: bool,
) -> Result<Receipt, Box<dyn Error>> {
    let mut value = json!({"name": name, "from": from});
    if line {
        value["line"] = json!(true);
    }
    change(
        storage,
        &format!("fork-{name}"),
        from_branch,
        json!({"kind":"branch.fork", "value": value}),
    )
}

fn discard(storage: &mut Storage, name: &str) -> Result<Value, Box<dyn Error>> {
    let receipt = change(
        storage,
        &format!("discard-{name}"),
        name,
        json!({"kind":"branch.discard", "value":{"name": name}}),
    )?;
    Ok(serde_json::to_value(receipt)?)
}

fn collection(storage: &mut Storage) -> Result<Version, Box<dyn Error>> {
    Ok(change(
        storage,
        "collection",
        "main",
        json!({"kind":"collection.put", "value":{"collection":"stages", "schema":null}}),
    )?
    .version)
}

/// Stage `index` on `branch`: one small record, its stage naming the
/// version before it.
fn stage(
    storage: &mut Storage,
    branch: &str,
    index: u64,
    line: bool,
) -> Result<Version, Box<dyn Error>> {
    let expected = head(storage, branch)?;
    let mut params = json!({
        "request_id": format!("stage-{index}"), "branch": branch, "expected": expected,
        "change": {"kind":"document.put", "value":{"collection":"stages",
            "document_id": format!("s{index}"), "document": {"stage": index}}},
    });
    if line {
        params["stage"] = json!({"line": branch, "run": "run-1", "stage": format!("s{index}"),
            "subject": "ward-7", "inputs": [{"branch": branch, "version": expected}]});
    }
    Ok(apply(storage, params)?.version)
}

/// The keep settings a store may name: the newest 64 versions, no days.
fn keep_settings() -> Rules {
    Rules {
        keep: Some(Keep {
            versions: NonZeroU64::new(64),
            days: Some(0),
        }),
        max_rows: NonZeroUsize::new(65_536).unwrap_or(NonZeroUsize::MIN),
        now: current_timestamp(),
    }
}

fn bytes_under(path: &Path) -> Result<u64, Box<dyn Error>> {
    let mut total = 0;
    for entry in std::fs::read_dir(path)? {
        let entry = entry?;
        let kind = entry.file_type()?;
        if kind.is_dir() {
            total += bytes_under(&entry.path())?;
        } else if kind.is_file() {
            total += entry.metadata()?.len();
        }
    }
    Ok(total)
}

/// Every history row of `branch` left in the service namespace.
fn rows_left(storage: &Storage, branch: &str) -> Result<usize, Box<dyn Error>> {
    let service = storage.refs.get(SERVICE).ok_or("no service")?;
    let rows = oldest_rows(
        &storage.reader,
        service,
        branch,
        (
            Version {
                generation: 0,
                sequence: 0,
            },
            Version {
                generation: u64::MAX,
                sequence: u64::MAX,
            },
        ),
        NonZeroUsize::new(65_536).ok_or("zero")?,
    )?;
    Ok(rows.len())
}

/// R2 red: a line with 4,100 stage commits, past the 4,096 ceiling of the
/// kept window, keeps its first version through a release under the keep
/// settings, forks at it, and gives every version up when it is discarded.
/// Also measures what a kept version costs.
#[test]
fn a_line_of_4100_stages_keeps_and_forks_at_its_first_version() -> TestResult {
    let (directory, mut storage) = create(KEPT_VERSIONS_MOST)?;
    let main = collection(&mut storage)?;
    fork(&mut storage, "main", "line", main, true)?;
    let data = directory.path().join("database");
    let before = bytes_under(&data)?;
    let started = std::time::Instant::now();
    let mut first = None;
    for index in 1..=STAGES {
        let at = stage(&mut storage, "line", index, true)?;
        first.get_or_insert(at);
    }
    let elapsed = started.elapsed();
    let after = bytes_under(&data)?;
    let first = first.ok_or("no stage")?;
    let service = storage.refs.get(SERVICE).ok_or("no service")?;
    let next = Version {
        sequence: first.sequence + 1,
        ..first
    };
    let rows = oldest_rows(
        &storage.reader,
        service,
        "line",
        (first, next),
        NonZeroUsize::MIN,
    )?;
    let row = rows.first().ok_or("the first stage has no history row")?;
    let row_bytes = row.key.encode().len() + row.encode_value()?.len();
    eprintln!(
        "STOR-050 R2 kept version: one history row of {row_bytes} bytes (key and value) on one \
         shard; the store grew {} bytes over {STAGES} stage commits, {} bytes a commit with its \
         record, stage row, receipt and author row, in {elapsed:?}",
        after - before,
        (after - before) / STAGES
    );
    let answer = storage.release("line", None, &keep_settings(), &IntentListing)?;
    assert_eq!(answer.released, None, "a line gives nothing to a release");
    let held = serde_json::to_value(&answer.held_by)?;
    assert_eq!(held[0]["source"], "line", "{held}");
    let rerun = fork(&mut storage, "line", "rerun", first, true)?;
    assert_eq!(rerun.version.sequence, 0);
    let rerun_head = storage.capture(&Selection {
        branch: "rerun".to_owned(),
        version: None,
    })?;
    assert_eq!(storage.fork_provenance(&rerun_head)?.version, first);
    let rerun_gone = discard(&mut storage, "rerun")?;
    assert_eq!(rerun_gone["versions_released"], 1, "{rerun_gone}");
    let gone = discard(&mut storage, "line")?;
    assert_eq!(gone["versions_released"], STAGES + 1, "{gone}");
    assert_eq!(
        rows_left(&storage, "line")?,
        0,
        "every row of the line released"
    );
    assert_eq!(
        rows_left(&storage, "rerun")?,
        0,
        "every row of the rerun released"
    );
    Ok(())
}

/// The control: an ordinary branch past the kept window still forks at its
/// first version, which no window loses, until a release under the keep
/// settings takes it, after which the fork is refused `version_released`.
#[test]
fn an_ordinary_branch_keeps_its_first_version_only_until_a_release() -> TestResult {
    let (_directory, mut storage) = create(KEPT_VERSIONS_DEFAULT)?;
    let main = collection(&mut storage)?;
    fork(&mut storage, "main", "work", main, false)?;
    let mut first = None;
    for index in 1..=200 {
        let at = stage(&mut storage, "work", index, false)?;
        first.get_or_insert(at);
    }
    let first = first.ok_or("no stage")?;
    fork(&mut storage, "work", "probe", first, false)?;
    let probe = discard(&mut storage, "probe")?;
    assert_eq!(probe.get("versions_released"), None, "{probe}");
    let answer = storage.release("work", None, &keep_settings(), &IntentListing)?;
    assert!(answer.released.is_some(), "{answer:?}");
    let refused = fork(&mut storage, "work", "late", first, false)
        .err()
        .ok_or("the released version was forked")?;
    assert!(
        refused.to_string().contains("version_released"),
        "{refused}"
    );
    Ok(())
}

/// R1: a crash between a stage's staged writes and its commit leaves neither
/// its change nor its stage row after a reopen, and the stage written again
/// has both.
#[test]
fn a_crash_before_the_commit_leaves_neither_the_change_nor_its_stage_row() -> TestResult {
    let (directory, mut storage) = create(KEPT_VERSIONS_DEFAULT)?;
    let main = collection(&mut storage)?;
    fork(&mut storage, "main", "line", main, true)?;
    let before = head(&storage, "line")?;
    super::lineage::before_commit::arm();
    let crashed = stage(&mut storage, "line", 1, true)
        .err()
        .ok_or("the crash was not injected")?;
    assert!(crashed.to_string().contains("injected_crash"), "{crashed}");
    drop(storage);
    let mut storage = Storage::open_kept(
        &directory.path().join("database"),
        NodeCacheBudget::Unlimited,
        KeptLimit::new(KEPT_VERSIONS_DEFAULT)?,
        std::time::Duration::from_secs(1),
    )?;
    let at = storage.capture(&Selection {
        branch: "line".to_owned(),
        version: None,
    })?;
    assert_eq!(at.version, before, "nothing was installed");
    let next = Version {
        sequence: before.sequence + 1,
        ..before
    };
    let document = super::document_key("stages", "s1");
    assert_eq!(
        super::reads::get_bytes(&storage.reader, &at, &document)?,
        None
    );
    assert_eq!(
        super::reads::get_bytes(&storage.reader, &at, &super::lineage::key(next))?,
        None
    );
    let written = stage(&mut storage, "line", 1, true)?;
    assert_eq!(written, next);
    let at = storage.capture(&Selection {
        branch: "line".to_owned(),
        version: None,
    })?;
    assert!(super::reads::get_bytes(&storage.reader, &at, &document)?.is_some());
    let lineage = super::lineage::read(&storage.reader, &at)?;
    assert_eq!(
        lineage.stage.map(|stage| stage.stage),
        Some("s1".to_owned()),
        "the stage row with its data"
    );
    Ok(())
}