haematite-cli 0.21.1

//! STOR-003 H1b2-A1: every service read holds the capture it answers from
//! until the answer is built, and a point read whose version has gone says so
//! by name.

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

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

use super::Service;
use crate::serve::protocol::{
    Change, CollectionPage, CollectionWrite, DocumentPage, DocumentRead, DocumentWrite, EdgePage,
    Fork, GraphPage, Mutation, Node, Operation, RelationsRead, ReviewPage, Selection, StoreRead,
    Version,
};
use crate::serve::storage::{Storage, after_held, document_key, reads};

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

const PAGE_LIMIT: NonZeroU32 = NonZeroU32::MIN.saturating_add(9);

fn head() -> Selection {
    Selection {
        branch: "main".to_owned(),
        version: None,
    }
}

/// `change` written on main at its head; the version it made.
fn write(
    storage: &mut Storage,
    request_id: &str,
    change: Change,
) -> Result<Version, Box<dyn Error>> {
    let mutation = Mutation {
        request_id: request_id.to_owned(),
        branch: "main".to_owned(),
        expected: storage.capture(&head())?.version,
        change,
    };
    Ok(storage.mutate(&mutation)?.version)
}

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

/// A one-shard store created as the service creates one, under reclaim,
/// with `records/a` written.
fn storage() -> 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(),
    )?;
    let declare = Change::Collection(CollectionWrite {
        collection: "records".to_owned(),
        schema: None,
    });
    write(&mut storage, "declare", declare)?;
    write(&mut storage, "a", put("a", "first")?)?;
    Ok((directory, storage))
}

/// Every service read that answers from a capture, by name.
fn reads_at_head() -> Vec<(&'static str, Operation)> {
    let node = || Node {
        collection: "records".to_owned(),
        document_id: "a".to_owned(),
    };
    let edges = |node: Node| EdgePage {
        selection: head(),
        node,
        relation: None,
        after: None,
        limit: PAGE_LIMIT,
    };
    vec![
        (
            "documents.get",
            Operation::Get(DocumentRead {
                selection: head(),
                collection: "records".to_owned(),
                document_id: "a".to_owned(),
            }),
        ),
        (
            "documents.list",
            Operation::List(DocumentPage {
                selection: head(),
                collection: "records".to_owned(),
                prefix: None,
                content: None,
                after: None,
                limit: PAGE_LIMIT,
            }),
        ),
        (
            "collections.list",
            Operation::Collections(CollectionPage {
                selection: head(),
                after: None,
                limit: PAGE_LIMIT,
            }),
        ),
        (
            "relations.list",
            Operation::Relations(RelationsRead { selection: head() }),
        ),
        (
            "store.stats",
            Operation::StoreStats(StoreRead { selection: head() }),
        ),
        (
            "vectors.fields",
            Operation::VectorFields(StoreRead { selection: head() }),
        ),
        ("counts.audit", Operation::CountsAudit(head())),
        ("edges.out", Operation::EdgesOut(edges(node()))),
        ("edges.in", Operation::EdgesIn(edges(node()))),
        (
            "graph.export",
            Operation::GraphExport(GraphPage {
                selection: head(),
                after: None,
                limit: PAGE_LIMIT,
                relations: None,
                collections: None,
                members: None,
                records: None,
                unrelated: None,
            }),
        ),
    ]
}

/// Fails while any service read answers from a capture it does not hold:
/// each read answers, and takes exactly one hold; the holds taken equal the
/// read paths exercised.
#[test]
fn every_service_read_holds_the_capture_it_answers_from() -> TestResult {
    let (_directory, storage) = storage()?;
    let service = Service::new(storage, PAGE_LIMIT)?;
    let reads = reads_at_head();
    let paths = reads.len();
    let before = after_held::held();
    for (name, operation) in reads {
        let at = after_held::held();
        service
            .dispatch(operation)
            .map_err(|failure| format!("{name}: {failure:?}"))?;
        assert_eq!(after_held::held() - at, 1, "{name} held no capture");
    }
    assert_eq!(after_held::held() - before, paths);
    Ok(())
}

/// Fails while a point read of a version whose nodes were given back says
/// `storage_unavailable`: an unheld capture of the first version of
/// `records/a`, read once that version's roots are gone, is
/// `version_released`, naming that version.
///
/// A control, as H1c's is: the kept window keeps the version, so its roots
/// are taken here by hand, as a pass would take a version nothing keeps.
#[test]
fn a_point_read_of_a_released_version_is_named_version_released() -> TestResult {
    let (directory, mut storage) = storage()?;
    let first = storage.capture(&head())?;
    write(&mut storage, "a-again", put("a", "second")?)?;
    let mut taken = 0_usize;
    for shard in &first.shards {
        let hex = shard.head.to_string();
        let (prefix, name) = hex.split_at(2);
        let store = directory
            .path()
            .join("database")
            .join(format!("shard-{}", shard.shard_id))
            .join("store");
        match std::fs::remove_file(store.join(prefix).join(name)) {
            Ok(()) => taken += 1,
            // An empty shard's head is no file.
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(error.into()),
        }
    }
    assert!(taken > 0, "no root of the version was taken");
    let reader = storage.reader()?;
    let failure = reads::get_bytes(&reader, &first, &document_key("records", "a"))
        .err()
        .ok_or("a released version was read")?;
    assert_eq!(failure.code, "version_released", "{}", failure.message);
    let named = format!(
        "version {}.{} of branch main",
        first.version.generation, first.version.sequence
    );
    assert!(failure.message.contains(&named), "{}", failure.message);
    Ok(())
}

/// Fails while a review reads unheld (STOR-003 H1b2-A2): each review read
/// holds every version it reads until its page is built. A review holds the
/// branch and its fork version; a comparison, as counts, prepare and
/// dangling make, also the parent's head. The parent's head a review reads
/// for its movement alone is not held.
#[test]
fn every_review_read_holds_each_version_it_reads() -> TestResult {
    let (_directory, mut storage) = storage()?;
    let from = storage.capture(&head())?.version;
    let fork = storage.mutate(&Mutation {
        request_id: "fork".to_owned(),
        branch: "main".to_owned(),
        expected: from,
        change: Change::Fork(Fork {
            name: "feature".to_owned(),
            from,
        }),
    })?;
    storage.mutate(&Mutation {
        request_id: "b".to_owned(),
        branch: "feature".to_owned(),
        expected: fork.version,
        change: put("b", "on feature")?,
    })?;
    let service = Service::new(storage, PAGE_LIMIT)?;
    let review = || ReviewPage {
        branch: "feature".to_owned(),
        after: None,
        limit: PAGE_LIMIT,
    };
    for (name, operation, holds) in [
        ("branch.review", Operation::Review(review()), 2),
        ("branch.review_counts", Operation::ReviewCounts(review()), 3),
        ("branch.merge_prepare", Operation::MergePrepare(review()), 3),
        (
            "branch.merge_dangling",
            Operation::MergeDangling(review()),
            3,
        ),
    ] {
        let at = after_held::held();
        service
            .dispatch(operation)
            .map_err(|failure| format!("{name}: {failure:?}"))?;
        assert_eq!(after_held::held() - at, holds, "{name}");
    }
    Ok(())
}

const fn fixture_long_wait_interval() -> std::time::Duration {
    std::time::Duration::from_secs(1)
}