haematite-cli 0.21.8

#![forbid(unsafe_code)]

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

use runtime_options::runtime_options;

use std::error::Error;
use std::num::NonZeroU32;

use haematite::db::{DatabaseConfig, vacuum::CANONICAL_REFS_DIR};
use haematite::store::NodeCacheBudget;
use haematite::tree::Hash;
use haematite_cli::serve::protocol::{
    BranchDiscard, BranchPage, Change, CollectionWrite, DocumentRead, DocumentWrite, Fork,
    Mutation, Operation, ReceiptQuery, Selection, Version,
};
use haematite_cli::serve::service::Service;
use haematite_cli::serve::storage::{Receipt, Storage};
use serde_json::{json, value::RawValue};

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

fn setup() -> Result<(tempfile::TempDir, Storage), Box<dyn Error>> {
    let directory = tempfile::tempdir()?;
    let storage = Storage::create(
        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,
        fixture_long_wait_interval(),
    )?;
    Ok((directory, storage))
}

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

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, None))?.version,
        change,
    })
}

fn fork(storage: &mut Storage, name: &str) -> Result<Receipt, Box<dyn Error>> {
    let from = storage.capture(&selected("main", None))?.version;
    let mutation = request(
        storage,
        "main",
        &format!("fork-{name}-{}-{}", from.generation, from.sequence),
        Change::Fork(Fork {
            name: name.to_owned(),
            from,
        }),
    )?;
    Ok(storage.mutate(&mutation)?)
}

fn discard(storage: &Storage, branch: &str, id: &str) -> Result<Mutation, Box<dyn Error>> {
    request(
        storage,
        branch,
        id,
        Change::Discard(BranchDiscard {
            name: branch.to_owned(),
        }),
    )
}

fn seed(storage: &mut Storage) -> TestResult {
    let collection = request(
        storage,
        "main",
        "collection",
        Change::Collection(CollectionWrite {
            collection: "records".to_owned(),
            schema: None,
        }),
    )?;
    storage.mutate(&collection)?;
    let document = request(
        storage,
        "main",
        "document",
        Change::Put(DocumentWrite {
            collection: "records".to_owned(),
            document_id: "one".to_owned(),
            document: RawValue::from_string("{\"value\":1}".to_owned())?,
        }),
    )?;
    storage.mutate(&document)?;
    Ok(())
}

#[test]
fn discard_removes_only_the_name_and_keeps_captured_documents() -> TestResult {
    let (_directory, mut storage) = setup()?;
    seed(&mut storage)?;
    let branch = fork(&mut storage, "work")?;
    let mutation = discard(&storage, "work", "discard")?;
    let receipt = storage.mutate(&mutation)?;
    assert_eq!(receipt.version, branch.version);
    assert_eq!(
        storage
            .capture(&selected("work", None))
            .err()
            .ok_or("branch stayed visible")?
            .code,
        "branch_missing"
    );
    let service = Service::new(storage, NonZeroU32::MIN)?;
    let retained = service.dispatch(Operation::Get(DocumentRead {
        selection: selected("work", Some(branch.version)),
        collection: "records".to_owned(),
        document_id: "one".to_owned(),
    }))?;
    assert_eq!(retained["document"]["raw_json"], "{\"value\":1}");
    let main = service.dispatch(Operation::Get(DocumentRead {
        selection: selected("main", None),
        collection: "records".to_owned(),
        document_id: "one".to_owned(),
    }))?;
    assert!(!main["document"].is_null());
    let listed = service.dispatch(Operation::Branches(BranchPage {
        after: None,
        limit: NonZeroU32::MIN,
    }))?;
    assert_eq!(
        listed["entries"]
            .as_array()
            .ok_or("branches not an array")?
            .len(),
        1
    );
    assert_eq!(listed["entries"][0]["branch"], "main");
    Ok(())
}

#[test]
fn lost_discard_answer_replays_after_restart() -> TestResult {
    let (directory, mut storage) = setup()?;
    fork(&mut storage, "work")?;
    let mutation = discard(&storage, "work", "lost-answer")?;
    let receipt = storage.mutate(&mutation)?;
    drop(storage);
    let mut storage = Storage::open(
        &directory.path().join("database"),
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    let kept = storage
        .receipt(&ReceiptQuery {
            branch: "work".to_owned(),
            request_id: "lost-answer".to_owned(),
        })?
        .ok_or("discard receipt missing")?;
    assert_eq!(kept.version, receipt.version);
    assert_eq!(kept.fingerprint, receipt.fingerprint);
    assert_eq!(storage.mutate(&mutation)?.fingerprint, receipt.fingerprint);
    assert!(storage.capture(&selected("work", None)).is_err());
    Ok(())
}

#[test]
fn changed_discard_payload_cannot_reuse_the_receipt() -> TestResult {
    let (_directory, mut storage) = setup()?;
    fork(&mut storage, "work")?;
    let mutation = discard(&storage, "work", "discard")?;
    storage.mutate(&mutation)?;
    let changed = Mutation {
        change: Change::Discard(BranchDiscard {
            name: "other".to_owned(),
        }),
        ..mutation
    };
    assert_eq!(
        storage
            .mutate(&changed)
            .err()
            .ok_or("changed request accepted")?
            .code,
        "request_conflict"
    );
    Ok(())
}

#[test]
fn main_mismatched_target_and_stale_version_are_refused() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let main = discard(&storage, "main", "main-discard")?;
    assert_eq!(
        storage.mutate(&main).err().ok_or("main discarded")?.code,
        "main_protected"
    );
    let branch = fork(&mut storage, "work")?;
    let mismatch = request(
        &storage,
        "work",
        "wrong-target",
        Change::Discard(BranchDiscard {
            name: "other".to_owned(),
        }),
    )?;
    assert_eq!(
        storage
            .mutate(&mismatch)
            .err()
            .ok_or("different branch discarded")?
            .code,
        "discard_target_mismatch"
    );
    let stale = discard(&storage, "work", "stale")?;
    let advance = request(
        &storage,
        "work",
        "advance-work",
        Change::Collection(CollectionWrite {
            collection: "new".to_owned(),
            schema: None,
        }),
    )?;
    let advanced = storage.mutate(&advance)?;
    assert_eq!(
        storage
            .mutate(&stale)
            .err()
            .ok_or("stale discard accepted")?
            .code,
        "version_conflict"
    );
    assert_ne!(advanced.version, branch.version);
    assert_eq!(
        storage.capture(&selected("work", None))?.version,
        advanced.version
    );
    Ok(())
}

#[test]
fn old_discard_replay_never_removes_a_recreated_branch() -> TestResult {
    let (directory, mut storage) = setup()?;
    fork(&mut storage, "work")?;
    let mutation = discard(&storage, "work", "old-discard")?;
    let receipt = storage.mutate(&mutation)?;
    let main = request(
        &storage,
        "main",
        "new-main-version",
        Change::Collection(CollectionWrite {
            collection: "new".to_owned(),
            schema: None,
        }),
    )?;
    storage.mutate(&main)?;
    let recreated = fork(&mut storage, "work")?;
    assert_ne!(recreated.version.generation, receipt.version.generation);
    assert_eq!(storage.mutate(&mutation)?.version, receipt.version);
    assert_eq!(
        storage.capture(&selected("work", None))?.version,
        recreated.version
    );
    drop(storage);
    let storage = Storage::open(
        &directory.path().join("database"),
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    assert_eq!(
        storage.capture(&selected("work", None))?.version,
        recreated.version
    );
    Ok(())
}

#[test]
fn open_finishes_the_durable_receipt_before_unlink_crash_point() -> TestResult {
    let (directory, mut storage) = setup()?;
    let branch = fork(&mut storage, "work")?;
    let reference = directory
        .path()
        .join("database")
        .join(CANONICAL_REFS_DIR)
        .join(format!(
            "{}.ref",
            Hash::of(b"@haem/branch/work")
                .to_string()
                .chars()
                .take(32)
                .collect::<String>()
        ));
    let original_reference = std::fs::read(&reference)?;
    let mutation = discard(&storage, "work", "interrupted")?;
    let receipt = storage.mutate(&mutation)?;
    drop(storage);
    // Restore the cut after the receipt row commit and before branch unlink.
    std::fs::write(&reference, original_reference)?;
    install_discard_intent(&directory.path().join("database"), &mutation)?;
    let mut storage = Storage::open(
        &directory.path().join("database"),
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    assert!(!reference.try_exists()?);
    assert!(storage.capture(&selected("work", None)).is_err());
    assert_eq!(storage.mutate(&mutation)?.fingerprint, receipt.fingerprint);
    assert_eq!(
        storage
            .capture(&selected("work", Some(branch.version)))?
            .version,
        branch.version
    );
    Ok(())
}

#[test]
fn fork_provenance_is_explicit_historical_and_replay_stable() -> TestResult {
    let (directory, mut storage) = setup()?;
    let origin = storage.capture(&selected("main", None))?.version;
    let mutation = request(
        &storage,
        "main",
        "historical-fork",
        Change::Fork(Fork {
            name: "work".to_owned(),
            from: origin,
        }),
    )?;
    let receipt = storage.mutate(&mutation)?;
    let at = storage.capture(&selected("work", None))?;
    let provenance = storage.fork_provenance(&at)?;
    assert_eq!(provenance.branch, "main");
    assert_eq!(provenance.version, origin);
    assert_eq!(storage.mutate(&mutation)?.fingerprint, receipt.fingerprint);
    let child = request(
        &storage,
        "work",
        "child-fork",
        Change::Fork(Fork {
            name: "child".to_owned(),
            from: at.version,
        }),
    )?;
    storage.mutate(&child)?;
    let child_origin = storage.fork_provenance(&storage.capture(&selected("child", None))?)?;
    assert_eq!(child_origin.branch, "work");
    assert_eq!(child_origin.version, at.version);
    drop(storage);
    let storage = Storage::open(
        &directory.path().join("database"),
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    let reopened = storage.fork_provenance(&storage.capture(&selected("work", None))?)?;
    assert_eq!(reopened.branch, "main");
    assert_eq!(reopened.version, origin);
    Ok(())
}

#[test]
fn absent_fork_provenance_is_named_without_guessing() -> TestResult {
    let (_directory, storage) = setup()?;
    let refusal = storage
        .fork_provenance(&storage.capture(&selected("main", None))?)
        .err()
        .ok_or("origin guessed")?;
    assert_eq!(refusal.code, "fork_provenance_missing");
    let change: Change =
        serde_json::from_value(json!({"kind":"branch.discard", "value":{"name":"work"}}))?;
    assert!(matches!(change, Change::Discard(_)));
    assert!(
        serde_json::from_value::<Change>(
            json!({"kind":"branch.discard", "value":{"name":"work", "extra":true}})
        )
        .is_err()
    );
    Ok(())
}

fn reference_count(directory: &std::path::Path) -> Result<u64, Box<dyn Error>> {
    let mut count = 0_u64;
    for entry in std::fs::read_dir(directory)? {
        if entry?
            .path()
            .extension()
            .and_then(|extension| extension.to_str())
            == Some("ref")
        {
            count = count.checked_add(1).ok_or("reference count overflow")?;
        }
    }
    Ok(count)
}

#[test]
fn completed_discards_add_only_reference_listing_reads_at_reopen() -> TestResult {
    let (directory, storage) = setup()?;
    let database = directory.path().join("database");
    let reference_directory = database.join(CANONICAL_REFS_DIR);
    let initial_references = reference_count(&reference_directory)?;
    drop(storage);
    // The first open changes retention and can leave a reclaim pass pending.
    // Finish that pass before measuring the same quiet state used below.
    drop(Storage::open(
        &database,
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?);
    let rested = haematite::db::Database::open(&database, runtime_options())?;
    rested.quiesce_reclaim()?;
    drop(rested);
    haematite::private_mode::log_file_opens(true);
    haematite::private_mode::reset_file_opens();
    let mut storage = Storage::open(
        &database,
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    let baseline_opens = haematite::private_mode::file_opens();
    let baseline_log = haematite::private_mode::file_open_log();
    haematite::private_mode::log_file_opens(false);
    for name in ["first", "second", "third"] {
        fork(&mut storage, name)?;
        let mutation = discard(&storage, name, &format!("discard-{name}"))?;
        storage.mutate(&mutation)?;
    }
    let final_references = reference_count(&reference_directory)?;
    drop(storage);
    // The discard writes can leave reclaim work pending at close.
    // Finish it before the second quiet-state measurement.
    let rested = haematite::db::Database::open(&database, runtime_options())?;
    rested.quiesce_reclaim()?;
    drop(rested);
    haematite::private_mode::log_file_opens(true);
    haematite::private_mode::reset_file_opens();
    let storage = Storage::open(
        &database,
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    let actual_opens = haematite::private_mode::file_opens();
    let log = haematite::private_mode::file_open_log();
    haematite::private_mode::log_file_opens(false);
    assert_eq!(final_references, initial_references);
    // Completed receipts live under the service root; no intent remains and
    // open seeks no receipt rows, so completed discards add no file opens.
    assert_eq!(
        actual_opens, baseline_opens,
        "baseline: {baseline_log:#?}; after discards: {log:#?}"
    );
    for name in ["first", "second", "third"] {
        assert!(storage.capture(&selected(name, None)).is_err());
    }
    Ok(())
}

fn install_discard_intent(database: &std::path::Path, mutation: &Mutation) -> TestResult {
    let mut refs = haematite::branch::RefStore::open(database.join(CANONICAL_REFS_DIR))?;
    let record = refs
        .get(&format!("@haem/branch/{}", mutation.branch))
        .ok_or("target reference missing")?
        .clone();
    let name = format!(
        "@haem/discard-pending/{}",
        serde_json::to_string(&json!({
            "branch": mutation.branch,
            "request_id": mutation.request_id,
            "version": mutation.expected,
        }))?
    );
    haematite::branch::create_branch_with_kind(
        &name,
        record
            .shards
            .iter()
            .map(|shard| (shard.shard_id, shard.head)),
        haematite::branch::Kind::Namespace,
        &mut refs,
        &haematite::branch::Registry::new(),
        haematite::branch::current_timestamp(),
    )?;
    Ok(())
}

#[test]
fn intent_before_receipt_row_is_aborted_without_removing_the_branch() -> TestResult {
    let (directory, mut storage) = setup()?;
    let branch = fork(&mut storage, "work")?;
    let mutation = discard(&storage, "work", "before-row")?;
    drop(storage);
    let database = directory.path().join("database");
    install_discard_intent(&database, &mutation)?;
    let mut storage = Storage::open(
        &database,
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    assert_eq!(
        storage.capture(&selected("work", None))?.version,
        branch.version
    );
    assert!(
        storage
            .receipt(&ReceiptQuery {
                branch: "work".to_owned(),
                request_id: "before-row".to_owned(),
            })?
            .is_none()
    );
    assert_eq!(reference_count(&database.join(CANONICAL_REFS_DIR))?, 3);
    storage.mutate(&mutation)?;
    assert!(storage.capture(&selected("work", None)).is_err());
    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
}