haematite-cli 0.35.1

//! 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::collections::HashSet;
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 haematite::{DiskStore, Hash, Node};
use serde_json::{Value, json};

use super::history::{KEPT_VERSIONS_DEFAULT, KEPT_VERSIONS_MOST, oldest_rows};
use super::lineage::ROW_BYTES;
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(())
}

/// Every node reached from `roots` on shard 0, read from disk.
fn reached(data: &Path, roots: &[Hash]) -> Result<HashSet<Hash>, Box<dyn Error>> {
    let store = DiskStore::new(
        data.join("shard-0").join("store"),
        NodeCacheBudget::Unlimited,
    )?;
    let mut seen = HashSet::new();
    let mut stack = roots.to_vec();
    while let Some(node) = stack.pop() {
        if !seen.insert(node) {
            continue;
        }
        let read = store.get(&node)?.ok_or_else(|| format!("{node} missing"))?;
        if let Node::Internal(internal) = read.as_ref() {
            stack.extend(internal.children().iter().map(|(_, child)| *child));
        }
    }
    Ok(seen)
}

/// The roots of `branch` at each of `versions`.
fn roots_at(
    storage: &Storage,
    branch: &str,
    versions: &[Version],
) -> Result<Vec<Hash>, Box<dyn Error>> {
    let mut roots = Vec::new();
    for version in versions {
        let at = storage.capture(&Selection {
            branch: branch.to_owned(),
            version: Some(*version),
        })?;
        roots.extend(at.shards.iter().map(|shard| shard.head));
    }
    Ok(roots)
}

/// The nodes only `line`'s `versions` reach: none of main's kept versions,
/// nor the service namespace's head, reaches them.
fn only_line(
    storage: &Storage,
    data: &Path,
    versions: &[Version],
) -> Result<HashSet<Hash>, Box<dyn Error>> {
    let main = storage.capture(&Selection {
        branch: "main".to_owned(),
        version: None,
    })?;
    let mut kept: Vec<Hash> = main.shards.iter().map(|shard| shard.head).collect();
    let main_record = storage
        .refs
        .get(&super::branch_name("main"))
        .ok_or("no main")?;
    kept.extend(
        storage
            .refs
            .kept(&main_record.name)
            .iter()
            .flat_map(|head| head.heads.iter().copied()),
    );
    let service = storage.refs.get(SERVICE).ok_or("no service")?;
    kept.extend(service.shards.iter().map(|shard| shard.head));
    let kept = reached(data, &kept)?;
    Ok(reached(data, &roots_at(storage, "line", versions)?)?
        .into_iter()
        .filter(|node| !kept.contains(node))
        .collect())
}

fn reopen(directory: &tempfile::TempDir) -> Result<Storage, Box<dyn Error>> {
    Ok(Storage::open_kept(
        &directory.path().join("database"),
        NodeCacheBudget::Unlimited,
        KeptLimit::new(KEPT_VERSIONS_DEFAULT)?,
        std::time::Duration::from_secs(1),
    )?)
}

/// Gap 1 red (Tom's pruning rule): a crash after a line's discard is
/// durable and before its release finishes leaves its history rows, and
/// every node they keep reachable. Reopening serve, with no replay of the
/// discard, finishes the release: the rows are gone, and a reclaim pass to
/// rest prunes every node only the line reached, counted.
#[test]
fn a_crash_inside_a_lines_release_is_finished_by_the_next_open() -> TestResult {
    let (directory, mut storage) = create(KEPT_VERSIONS_DEFAULT)?;
    let main = collection(&mut storage)?;
    let anchor = fork(&mut storage, "main", "line", main, true)?.version;
    let mut versions = vec![anchor];
    for index in 1..=3 {
        versions.push(stage(&mut storage, "line", index, true)?);
    }
    let data = directory.path().join("database");
    let only = only_line(&storage, &data, &versions)?;
    assert!(!only.is_empty(), "the line wrote nodes of its own");
    super::release::crash_at(super::release::Crash::AfterDecision);
    let stopped = discard(&mut storage, "line")
        .err()
        .ok_or("the crash was not injected")?;
    assert!(
        stopped.to_string().contains("line_release_unfinished"),
        "{stopped}"
    );
    assert_eq!(
        rows_left(&storage, "line")?,
        versions.len(),
        "the crash left the rows"
    );
    drop(storage);
    let storage = reopen(&directory)?;
    assert_eq!(
        rows_left(&storage, "line")?,
        0,
        "the open finished the line's release"
    );
    storage.activate_reclaim()?;
    storage.quiesce_reclaim()?;
    let store = DiskStore::new(
        data.join("shard-0").join("store"),
        NodeCacheBudget::Unlimited,
    )?;
    let mut left = 0_usize;
    let nodes: Vec<&Hash> = only.iter().collect();
    for node in nodes {
        if store.get(node)?.is_some() {
            left += 1;
        }
    }
    eprintln!(
        "lane 27 gap 1: {} nodes only the line reached, {left} left after the reopen and a \
         reclaim pass to rest",
        only.len()
    );
    assert_eq!(left, 0, "of {} nodes only the line reached", only.len());
    Ok(())
}

/// Gap 1: opening a store with no unfinished line release reads one page
/// of the unfinished rows, finds none, and writes nothing. A line whose
/// discard finished leaves none: its last round's commit took the row.
#[test]
fn an_open_with_no_unfinished_release_reads_one_page_and_writes_nothing() -> TestResult {
    let (directory, mut storage) = create(KEPT_VERSIONS_DEFAULT)?;
    let main = collection(&mut storage)?;
    fork(&mut storage, "main", "line", main, true)?;
    stage(&mut storage, "line", 1, true)?;
    let service = storage.refs.get(SERVICE).ok_or("no service")?;
    let before_discard = service.seq;
    let gone = discard(&mut storage, "line")?;
    assert_eq!(gone["versions_released"], 2, "{gone}");
    assert_eq!(rows_left(&storage, "line")?, 0);
    let service = storage.refs.get(SERVICE).ok_or("no service")?;
    let before = (service.created, service.seq);
    eprintln!(
        "lane 27 gap 1: the line's discard made {} service commits",
        service.seq - before_discard
    );
    drop(storage);
    super::lineage::unfinished_reads::take();
    let storage = reopen(&directory)?;
    let (reads, rows) = super::lineage::unfinished_reads::take();
    eprintln!(
        "lane 27 gap 1: a clean open made {reads} reads of the unfinished rows and found {rows}"
    );
    assert_eq!((reads, rows), (1, 0), "one read of the one shard, no row");
    let service = storage.refs.get(SERVICE).ok_or("no service")?;
    assert_eq!(
        (service.created, service.seq),
        before,
        "the open wrote nothing"
    );
    Ok(())
}

/// Gap 2 red: a merged commit's stage row lists every stage its source
/// committed since its fork, so a line of 4,100 stages made it one row of
/// about a megabyte. Stored in bounded pieces, every row the merge writes
/// is at most `ROW_BYTES`, and history answers the whole list.
#[test]
fn a_spring_back_of_4100_stages_writes_bounded_stage_rows() -> TestResult {
    let (_directory, mut storage) = create(KEPT_VERSIONS_DEFAULT)?;
    let main = collection(&mut storage)?;
    fork(&mut storage, "main", "line", main, true)?;
    for index in 1..=STAGES {
        stage(&mut storage, "line", index, true)?;
    }
    let reviewed = head(&storage, "line")?;
    let merged = change(
        &mut storage,
        "merge",
        "main",
        json!({"kind":"branch.merge", "value":{"from":"line", "reviewed": reviewed}}),
    )?
    .version;
    let at = storage.capture(&Selection {
        branch: "main".to_owned(),
        version: Some(merged),
    })?;
    let mut rows = Vec::new();
    for shard in &at.shards {
        let page = storage.reader.range_page_at(
            shard.shard_id,
            shard.head,
            super::lineage::TAG,
            b"s\x01",
            NonZeroUsize::new(65_536).ok_or("zero")?,
        )?;
        rows.extend(page.entries.into_iter().map(|(_key, value)| value.len()));
    }
    let largest = rows.iter().copied().max().ok_or("no stage row")?;
    let total: usize = rows.iter().sum();
    let lineage = super::lineage::read(&storage.reader, &at)?;
    let listed = lineage.merged.ok_or("no merged row")?;
    let answered = serde_json::to_vec(&listed)?.len();
    eprintln!(
        "lane 27 gap 2: a merge of {STAGES} stages wrote {} stage rows, {total} bytes, the \
         largest {largest} bytes; history's merged member is {answered} bytes",
        rows.len()
    );
    assert_eq!(
        listed.stages.len(),
        usize::try_from(STAGES)?,
        "every stage listed"
    );
    for (index, listed) in (1..=STAGES).zip(&listed.stages) {
        assert_eq!(listed.stage.stage, format!("s{index}"), "in commit order");
    }
    assert!(
        largest <= ROW_BYTES,
        "the largest stage row is {largest} bytes, over {ROW_BYTES}"
    );
    Ok(())
}