haematite-cli 0.21.7

//! REL0: `edges.put_many`, all or none in one commit.

use std::error::Error;
use std::num::{NonZeroU32, NonZeroUsize};

use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::{json, value::RawValue};

use super::export;
use crate::serve::protocol::{
    Cardinality, Change, CollectionWrite, DocumentWrite, Edge, EdgeImport, Mutation, Node,
    Relation, Selection, Version,
};
use crate::serve::storage::{Failure, Receipt, Storage};

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

/// The service's page bound in these tests: imports of up to 64 rows.
const ONE_PAGE: Option<NonZeroU32> = NonZeroU32::new(1);

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

fn apply(storage: &mut Storage, id: &str, change: Change) -> Result<Receipt, Failure> {
    let max_page = ONE_PAGE.ok_or_else(|| Failure::new("page_limit", "one is not zero"))?;
    let expected = head(storage)?;
    storage.mutate_bounded(
        &Mutation {
            request_id: id.to_owned(),
            branch: "main".to_owned(),
            expected,
            change,
        },
        max_page,
    )
}

fn node(id: &str) -> Node {
    Node {
        collection: "records".to_owned(),
        document_id: id.to_owned(),
    }
}

fn edge(relation: &str, from: &str, to: &str) -> Edge {
    Edge {
        relation: relation.to_owned(),
        from: node(from),
        to: node(to),
        properties: json!({}),
    }
}

fn many(edges: Vec<Edge>) -> Change {
    Change::EdgePutMany(EdgeImport { edges })
}

/// Records a to f, `links` many to many and `pairs` one to one, both
/// records to records.
fn store() -> 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: 2,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::KeepAll,
        },
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    apply(
        &mut storage,
        "collection",
        Change::Collection(CollectionWrite {
            collection: "records".to_owned(),
            schema: None,
        }),
    )?;
    for id in ["a", "b", "c", "d", "e", "f"] {
        apply(
            &mut storage,
            id,
            Change::Put(DocumentWrite {
                collection: "records".to_owned(),
                document_id: id.to_owned(),
                document: RawValue::from_string("{}".to_owned())?,
            }),
        )?;
    }
    for (relation, cardinality) in [
        ("links", Cardinality::ManyToMany),
        ("pairs", Cardinality::OneToOne),
    ] {
        apply(
            &mut storage,
            relation,
            Change::RelationPut(Relation {
                relation: relation.to_owned(),
                from_collection: "records".to_owned(),
                to_collection: "records".to_owned(),
                cardinality,
            }),
        )?;
    }
    Ok((directory, storage))
}

/// A stored edge as (relation, from, to).
type Stored = (String, String, String);

/// Every stored edge.
fn stored(storage: &Storage) -> Result<Vec<Stored>, Box<dyn Error>> {
    let at = storage.capture(&Selection {
        branch: "main".to_owned(),
        version: None,
    })?;
    let limit = NonZeroUsize::new(100).ok_or("not zero")?;
    let page = export(&storage.reader, &at, None, limit, None)?;
    Ok(page
        .entries
        .into_iter()
        .map(|edge| (edge.relation, edge.from.document_id, edge.to.document_id))
        .collect())
}

/// REL0 red: five edges, one version and one record install.
#[test]
fn five_edges_are_one_commit() -> TestResult {
    let (_directory, mut storage) = store()?;
    let before = storage.sync_counts();
    let edges = vec![
        edge("links", "a", "b"),
        edge("links", "a", "c"),
        edge("links", "b", "c"),
        edge("pairs", "d", "e"),
        edge("links", "f", "f"),
    ];
    apply(&mut storage, "edges-1", many(edges))?;
    assert_eq!(storage.sync_counts().since(before).record_installs, 1);
    assert_eq!(stored(&storage)?.len(), 5);
    Ok(())
}

/// Refused whole: the version does not move and no edge is stored.
fn refused_whole(storage: &mut Storage, edges: Vec<Edge>) -> Result<Failure, Box<dyn Error>> {
    let version = head(storage)?;
    let refused = apply(storage, "bad", many(edges))
        .err()
        .ok_or("a bad batch was applied")?;
    assert_eq!(head(storage)?, version);
    assert_eq!(stored(storage)?, Vec::new());
    Ok(refused)
}

/// REL0 red: an edge to a record that does not exist refuses the batch,
/// naming its row.
#[test]
fn a_missing_end_refuses_the_whole_batch() -> TestResult {
    let (_directory, mut storage) = store()?;
    let refused = refused_whole(
        &mut storage,
        vec![edge("links", "a", "b"), edge("links", "a", "nobody")],
    )?;
    assert_eq!(refused.code, "document_missing");
    assert_eq!(refused.path.as_deref(), Some("/edges/1/to"));
    Ok(())
}

/// REL0 red: two edges from one record under a one to one relation, both in
/// the batch, refuse it; neither is stored.
#[test]
fn cardinality_across_the_batch_refuses_the_whole_batch() -> TestResult {
    let (_directory, mut storage) = store()?;
    let refused = refused_whole(
        &mut storage,
        vec![edge("pairs", "a", "b"), edge("pairs", "a", "c")],
    )?;
    assert_eq!(refused.code, "relation_cardinality");
    assert_eq!(refused.path.as_deref(), Some("/edges/1"));
    Ok(())
}

/// REL0: an edge whose ends are not the relation's collections refuses the
/// batch, as `edge.put` refuses it.
#[test]
fn a_relation_collection_mismatch_refuses_the_whole_batch() -> TestResult {
    let (_directory, mut storage) = store()?;
    apply(
        &mut storage,
        "others",
        Change::Collection(CollectionWrite {
            collection: "others".to_owned(),
            schema: None,
        }),
    )?;
    apply(
        &mut storage,
        "other-x",
        Change::Put(DocumentWrite {
            collection: "others".to_owned(),
            document_id: "x".to_owned(),
            document: RawValue::from_string("{}".to_owned())?,
        }),
    )?;
    let mut wrong = edge("links", "a", "b");
    wrong.to = Node {
        collection: "others".to_owned(),
        document_id: "x".to_owned(),
    };
    let refused = refused_whole(&mut storage, vec![edge("links", "a", "b"), wrong])?;
    assert_eq!(refused.code, "relation_collection");
    assert_eq!(refused.path.as_deref(), Some("/edges/1/to/collection"));
    Ok(())
}

/// REL0 red: one row past the bound (64 rows at a page bound of 1) is
/// refused before anything is read.
#[test]
fn over_the_bound_is_refused() -> TestResult {
    let (_directory, mut storage) = store()?;
    let edges = (0..65).map(|_| edge("links", "a", "b")).collect();
    let refused = refused_whole(&mut storage, edges)?;
    assert_eq!(refused.code, "import_limit");
    assert_eq!(refused.path.as_deref(), Some("/edges/64"));
    Ok(())
}

/// REL0: the same request again answers the first receipt and writes
/// nothing.
#[test]
fn a_replay_writes_nothing() -> TestResult {
    let (_directory, mut storage) = store()?;
    let edges = || vec![edge("links", "a", "b"), edge("links", "b", "c")];
    // A replay is the same request again, its expected version included:
    // the receipt's fingerprint is taken over the whole request.
    let mutation = Mutation {
        request_id: "edges-1".to_owned(),
        branch: "main".to_owned(),
        expected: head(&storage)?,
        change: many(edges()),
    };
    let max_page = ONE_PAGE.ok_or("one is not zero")?;
    let first = storage.mutate_bounded(&mutation, max_page)?;
    let before = storage.sync_counts();
    let again = storage.mutate_bounded(&mutation, max_page)?;
    assert_eq!(again.version, first.version);
    assert_eq!(storage.sync_counts().since(before).record_installs, 0);
    assert_eq!(stored(&storage)?.len(), 2);
    Ok(())
}

#[test]
fn a_duplicate_edge_refuses_the_whole_batch_without_flushes() -> TestResult {
    let (_directory, mut storage) = store()?;
    let before = storage.sync_counts();
    let refused = refused_whole(
        &mut storage,
        vec![edge("links", "a", "b"), edge("links", "a", "b")],
    )?;
    assert_eq!(refused.code, "import_duplicate");
    assert_eq!(refused.path.as_deref(), Some("/edges/1"));
    assert_eq!(
        storage.sync_counts().since(before),
        haematite::SyncCounts::default(),
    );
    Ok(())
}

#[test]
fn an_undeclared_relation_refuses_the_whole_batch_without_flushes() -> TestResult {
    let (_directory, mut storage) = store()?;
    let before = storage.sync_counts();
    let refused = refused_whole(
        &mut storage,
        vec![edge("links", "a", "b"), edge("unknown", "b", "c")],
    )?;
    assert_eq!(refused.code, "relation_missing");
    assert_eq!(refused.path.as_deref(), Some("/edges/1/relation"));
    assert_eq!(
        storage.sync_counts().since(before),
        haematite::SyncCounts::default(),
    );
    Ok(())
}

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