use std::collections::BTreeSet;
use haematite::SyncCounts;
use serde_json::json;
use super::batch_test_support::{ONE_WRITE, TestResult, apply, at, record_id, store, vector};
use super::{Storage, document_key, reads, revision_key};
use crate::serve::protocol::{
Cardinality, Change, DocumentDeleteMany, Edge, EdgeImport, Node, Relation, VectorImport,
};
fn many(document_ids: Vec<String>) -> Change {
Change::DeleteMany(DocumentDeleteMany {
collection: "records".to_owned(),
document_ids,
})
}
fn node(id: &str) -> Node {
Node {
collection: "records".to_owned(),
document_id: id.to_owned(),
}
}
#[test]
fn three_hundred_deletes_deduplicate_the_cascade_and_update_counts_once() -> TestResult {
let (_directory, mut storage) = store(300)?;
apply(
&mut storage,
"relation",
Change::RelationPut(Relation {
relation: "links".to_owned(),
from_collection: "records".to_owned(),
to_collection: "records".to_owned(),
cardinality: Cardinality::ManyToMany,
}),
)?;
apply(
&mut storage,
"vectors",
Change::VectorPutMany(VectorImport {
vectors: (0..300).map(vector).collect(),
}),
)?;
let mut edges: Vec<_> = (0..300)
.map(|index| Edge {
relation: "links".to_owned(),
from: node(&record_id(index)),
to: node(&record_id((index + 1) % 300)),
properties: json!({}),
})
.collect();
edges.push(Edge {
relation: "links".to_owned(),
from: node(&record_id(0)),
to: node("anchor"),
properties: json!({}),
});
edges.push(Edge {
relation: "links".to_owned(),
from: node("anchor"),
to: node(&record_id(299)),
properties: json!({}),
});
edges.push(Edge {
relation: "links".to_owned(),
from: node(&record_id(137)),
to: node(&record_id(137)),
properties: json!({}),
});
apply(
&mut storage,
"edges",
Change::EdgePutMany(EdgeImport { edges }),
)?;
let change = many((0..300).rev().map(record_id).collect());
let before = at(&storage)?;
let next = crate::serve::protocol::Version {
generation: before.version.generation,
sequence: before.version.sequence + 1,
};
let prepared = storage.prepare_change(&before, &change, next)?;
let unique: BTreeSet<_> = prepared.iter().map(|(key, _)| key).collect();
assert_eq!(unique.len(), prepared.len());
assert_eq!(prepared.len(), 300 * 3 + 303 * 2);
let counts = storage.sync_counts();
let receipt = apply(&mut storage, "delete", change)?;
assert_eq!(receipt.version, next);
assert_eq!(storage.sync_counts().since(counts), ONE_WRITE);
let after = at(&storage)?;
for index in 0..300 {
for key in [
document_key("records", &record_id(index)),
revision_key("records", &record_id(index)),
] {
assert!(reads::get_bytes(&storage.reader, &after, &key)?.is_none());
assert!(reads::get_bytes(&storage.reader, &before, &key)?.is_some());
}
}
for (prefix, parts, expected) in [
(b"d\0".as_slice(), 2, 1),
(b"m\0".as_slice(), 2, 1),
(b"e\0".as_slice(), 5, 0),
(b"i\0".as_slice(), 5, 0),
(b"v\0".as_slice(), 3, 1),
] {
assert_eq!(
reads::test_support::count_keys(&storage.reader, &after, prefix, parts)?,
expected,
);
}
let access = crate::serve::access::request(&crate::serve::access::Standalone)?;
let audit = serde_json::to_value(Storage::audit_counts(&access, &storage.reader, &after)?)?;
assert!(
audit["buckets"]
.as_array()
.ok_or("no count buckets")?
.iter()
.all(|row| row["result"] == "agree")
);
Ok(())
}
#[test]
fn unknown_id_at_137_refuses_the_entire_delete_without_flushes() -> TestResult {
let (_directory, mut storage) = store(300)?;
let before = at(&storage)?.version;
let counts = storage.sync_counts();
let mut ids: Vec<_> = (0..300).map(record_id).collect();
ids[137] = "unknown".to_owned();
let failure = apply(&mut storage, "delete", many(ids))
.err()
.ok_or("unknown document accepted")?;
assert_eq!(failure.code, "document_missing");
assert_eq!(failure.path.as_deref(), Some("/document_ids/137"));
assert_eq!(at(&storage)?.version, before);
assert_eq!(storage.sync_counts().since(counts), SyncCounts::default());
assert_eq!(
reads::test_support::count_keys(&storage.reader, &at(&storage)?, b"d\0", 2)?,
301,
);
Ok(())
}
#[test]
fn duplicate_delete_refuses_whole() -> TestResult {
let (_directory, mut storage) = store(2)?;
let before = at(&storage)?.version;
let counts = storage.sync_counts();
let failure = apply(
&mut storage,
"delete",
many(vec![record_id(0), record_id(0)]),
)
.err()
.ok_or("duplicate delete accepted")?;
assert_eq!(failure.code, "import_duplicate");
assert_eq!(failure.path.as_deref(), Some("/document_ids/1"));
assert_eq!(at(&storage)?.version, before);
assert_eq!(storage.sync_counts().since(counts), SyncCounts::default());
Ok(())
}
#[test]
fn empty_over_bound_and_missing_collection_reuse_import_refusals() -> TestResult {
let (_directory, mut storage) = store(2)?;
let before = at(&storage)?.version;
let counts = storage.sync_counts();
for (change, code, path) in [
(many(vec![]), "import_empty", "/document_ids"),
(
many((0..513).map(record_id).collect()),
"import_limit",
"/document_ids/512",
),
(
Change::DeleteMany(DocumentDeleteMany {
collection: "unknown".to_owned(),
document_ids: vec![record_id(0)],
}),
"collection_missing",
"/collection",
),
(
many(vec![record_id(0), String::new()]),
"invalid_identifier",
"/document_ids/1",
),
] {
let failure = apply(&mut storage, "delete", change)
.err()
.ok_or("invalid delete batch accepted")?;
assert_eq!(failure.code, code);
assert_eq!(failure.path.as_deref(), Some(path));
assert_eq!(at(&storage)?.version, before);
}
assert_eq!(storage.sync_counts().since(counts), SyncCounts::default());
Ok(())
}
#[test]
fn replay_of_deleted_ids_answers_the_original_receipt() -> TestResult {
let (_directory, mut storage) = store(2)?;
let request = crate::serve::protocol::Mutation {
request_id: "delete".to_owned(),
branch: "main".to_owned(),
expected: at(&storage)?.version,
change: many(vec![record_id(1), record_id(0)]),
};
let bound = std::num::NonZeroU32::MIN.saturating_add(7);
let first = storage.mutate_bounded(&request, bound)?;
let counts = storage.sync_counts();
let second = storage.mutate_bounded(&request, bound)?;
assert_eq!(first.version, second.version);
assert_eq!(first.fingerprint, second.fingerprint);
assert_eq!(storage.sync_counts().since(counts).record_installs, 0);
Ok(())
}
#[test]
fn cascade_deletion_and_receipt_survive_reopen() -> TestResult {
let (directory, mut storage) = store(2)?;
apply(
&mut storage,
"vectors",
Change::VectorPutMany(VectorImport {
vectors: vec![vector(0), vector(1)],
}),
)?;
let receipt = apply(
&mut storage,
"delete",
many(vec![record_id(1), record_id(0)]),
)?;
drop(storage);
let mut storage = Storage::open(
&directory.path().join("database"),
haematite::store::NodeCacheBudget::Unlimited,
std::time::Duration::from_secs(1),
)?;
let captured = at(&storage)?;
assert_eq!(captured.version, receipt.version);
for (prefix, parts) in [(b"d\0".as_slice(), 2), (b"v\0".as_slice(), 3)] {
assert_eq!(
reads::test_support::count_keys(&storage.reader, &captured, prefix, parts)?,
1,
);
}
let replay = storage
.receipt(&crate::serve::protocol::ReceiptQuery {
branch: "main".to_owned(),
request_id: "delete".to_owned(),
})?
.ok_or("delete receipt missing after reopen")?;
assert_eq!(replay.version, receipt.version);
assert_eq!(replay.fingerprint, receipt.fingerprint);
Ok(())
}