use std::error::Error;
use std::num::NonZeroU32;
use haematite::branch::{CommitDurability, CommitRequest, current_timestamp, open};
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::value::RawValue;
use super::{READY_KEY, READY_VALUE, document_count_key, is_ready, probe, required_documents};
use crate::serve::protocol::{
Change, CollectionWrite, CountReason, DocumentDelete, DocumentImport, DocumentWrite,
ImportedDocument, Mutation, ReceiptQuery, Selection,
};
use crate::serve::storage::{Captured, Storage, branch_name, document_key, reads, revision_key};
use crate::serve::protocol::{
Cardinality, Edge, EdgeDelete, MutationContext, MutationOutcome, Node, Relation,
ResponseFailure, VectorDelete, VectorWrite,
};
use crate::serve::storage::{Failure, edges, mutations::PreparedWrite};
type TestResult = Result<(), Box<dyn Error>>;
fn rebuild_request(id: &str) -> crate::serve::protocol::CountsRebuild {
crate::serve::protocol::CountsRebuild {
branch: "main".to_owned(),
request_id: id.to_owned(),
}
}
fn audit(fixture: &Fixture) -> Result<serde_json::Value, Box<dyn Error>> {
let access = crate::serve::access::request(&crate::serve::access::Standalone)?;
Ok(serde_json::to_value(Storage::audit_counts(
&access,
&fixture.storage.reader,
&fixture.at()?,
)?)?)
}
#[test]
fn rebuild_unready_counts_every_kind_and_marks_one_version() -> TestResult {
let mut fixture = populated()?;
fixture.apply(Change::EdgePut(edge("record-0", "record-1")))?;
fixture.apply(vector("record-0"))?;
fixture.edit_batch(vec![
(READY_KEY.to_vec(), None),
(document_count_key("records"), None),
(EDGE_COUNT.to_vec(), None),
(VECTOR_COUNT.to_vec(), None),
])?;
let before = fixture.at()?;
let receipt = fixture
.storage
.rebuild_counts(&rebuild_request("rebuild"))?;
let after = fixture.at()?;
assert_eq!(after.version.sequence, before.version.sequence + 1);
assert_eq!(receipt.version, after.version);
assert!(is_ready(&fixture.storage.reader, &after)?);
for (key, prefix, parts) in [
(document_count_key("records"), b"d\0".as_slice(), 2),
(EDGE_COUNT.to_vec(), b"e\0".as_slice(), 5),
(VECTOR_COUNT.to_vec(), b"v\0".as_slice(), 3),
] {
assert_eq!(
stored_count(&fixture, &key, &after)?,
reads::test_support::count_keys(&fixture.storage.reader, &after, prefix, parts)?,
);
assert!(reads::get_bytes(&fixture.storage.reader, &before, &key)?.is_none());
}
assert!(!is_ready(&fixture.storage.reader, &before)?);
let replay = fixture
.storage
.rebuild_counts(&rebuild_request("rebuild"))?;
assert_eq!(replay.version, receipt.version);
assert_eq!(fixture.at()?.version, after.version);
let fixture = fixture.reopen()?;
assert_eq!(fixture.at()?.version, after.version);
assert!(
audit(&fixture)?["buckets"]
.as_array()
.ok_or("no buckets")?
.iter()
.all(|row| row["result"] == "agree")
);
Ok(())
}
#[test]
fn audit_names_unready_withheld_and_differ_then_rebuild_repairs() -> TestResult {
let mut fixture = populated()?;
fixture.storage.rebuild_counts(&rebuild_request("first"))?;
assert!(
audit(&fixture)?["buckets"]
.as_array()
.ok_or("no buckets")?
.iter()
.all(|row| row["result"] == "agree")
);
fixture.edit(document_count_key("records"), Some(b"9".to_vec()))?;
let before = fixture.at()?.version;
let answer = audit(&fixture)?;
let row = answer["buckets"]
.as_array()
.ok_or("no buckets")?
.iter()
.find(|row| row["kind"] == "documents")
.ok_or("no documents")?;
assert_eq!(row["result"], "differ");
assert_eq!(row["kept"], 9);
assert_eq!(row["counted"], 2);
assert_eq!(fixture.at()?.version, before);
fixture.storage.rebuild_counts(&rebuild_request("repair"))?;
assert_eq!(fixture.count()?, 2);
let denied = crate::serve::access::request(&crate::serve::access::TestSource::denying(&[(
"records", "record-0",
)]))?;
reads::take_content_reads();
let withheld = serde_json::to_value(Storage::audit_counts(
&denied,
&fixture.storage.reader,
&fixture.at()?,
)?)?;
assert_eq!(withheld["result"], "cannot_tell");
assert_eq!(withheld["reason"], "withheld");
assert!(withheld.get("buckets").is_none());
assert_eq!(reads::take_content_reads(), 0);
fixture.edit(READY_KEY.to_vec(), None)?;
assert_eq!(audit(&fixture)?["reason"], "counts_not_ready");
Ok(())
}
#[test]
fn rebuild_declared_zeros_and_undeclared_relationship_and_removes_orphan_count() -> TestResult {
let mut fixture = populated()?;
fixture.apply(vector("record-0"))?;
fixture.apply(vector_delete("record-0"))?;
let mut undeclared = edge("record-0", "record-1");
undeclared.relation = "unregistered".to_owned();
fixture.apply(Change::EdgePut(undeclared))?;
fixture.edit(b"k\0e\0orphan".to_vec(), Some(b"17".to_vec()))?;
fixture.storage.rebuild_counts(&rebuild_request("zeros"))?;
let at = fixture.at()?;
assert_eq!(stored_count(&fixture, EDGE_COUNT, &at)?, 0);
assert_eq!(stored_count(&fixture, VECTOR_COUNT, &at)?, 0);
assert_eq!(stored_count(&fixture, b"k\0e\0unregistered", &at)?, 1);
assert!(reads::get_bytes(&fixture.storage.reader, &at, b"k\0e\0orphan")?.is_none());
Ok(())
}
#[test]
fn rebuild_refuses_merge_and_discard_before_staging() -> TestResult {
let mut fixture = populated()?;
let name = branch_name("main");
fixture.storage.pending_discards.insert(name.clone());
let before = fixture.at()?.version;
let error = fixture
.storage
.rebuild_counts(&rebuild_request("discard"))
.err()
.ok_or("discard was not refused")?;
assert_eq!(error.code, "counts_discard_pending");
fixture.storage.pending_discards.remove(&name);
let plan = crate::serve::storage::MergePlan {
id: "open".to_owned(),
state: crate::serve::storage::MergeState::Open,
source: crate::serve::storage::MergePin {
branch: "other".to_owned(),
version: before,
},
target: crate::serve::storage::MergePin {
branch: "main".to_owned(),
version: before,
},
origin: crate::serve::storage::MergePin {
branch: "main".to_owned(),
version: before,
},
};
let key = format!("j\0merge\0{}", haematite::tree::Hash::of(b"open")).into_bytes();
fixture
.storage
.write_merge_service_rows(&[(key, serde_json::to_vec(&plan)?)])?;
let error = fixture
.storage
.rebuild_counts(&rebuild_request("merge"))
.err()
.ok_or("merge was not refused")?;
assert_eq!(error.code, "counts_merge_pending");
assert_eq!(fixture.at()?.version, before);
assert!(
fixture
.storage
.receipt(&ReceiptQuery {
branch: "main".to_owned(),
request_id: "merge".to_owned(),
})?
.is_none()
);
Ok(())
}
#[test]
fn rebuild_missing_and_corrupt_counters_are_audited_and_repaired() -> TestResult {
let mut fixture = populated()?;
fixture.edit(document_count_key("records"), None)?;
let answer = audit(&fixture)?;
let row = answer["buckets"]
.as_array()
.ok_or("no buckets")?
.iter()
.find(|row| row["kind"] == "documents")
.ok_or("no documents")?;
assert_eq!(row["result"], "differ");
assert!(row["kept"].is_null());
assert_eq!(row["counted"], 2);
fixture.edit(document_count_key("records"), Some(b"01".to_vec()))?;
let owner = crate::serve::access::request(&crate::serve::access::Standalone)?;
let error = Storage::audit_counts(&owner, &fixture.storage.reader, &fixture.at()?)
.err()
.ok_or("noncanonical count was accepted")?;
assert_eq!(error.code, "counts_corrupt");
fixture
.storage
.rebuild_counts(&rebuild_request("corrupt"))?;
assert_eq!(fixture.count()?, 2);
Ok(())
}
#[test]
fn rebuild_writer_is_held_until_commit_and_audit_does_not_repair() -> TestResult {
use crate::serve::protocol::{CountsRebuild, Operation};
use std::sync::{Arc, mpsc};
let fixture = populated()?;
let before = fixture.at()?.version;
let request = Mutation {
request_id: "racing".to_owned(),
branch: "main".to_owned(),
expected: before,
change: put("racing")?,
};
let Fixture {
storage, directory, ..
} = fixture;
let service = Arc::new(crate::serve::service::Service::new(
storage,
NonZeroU32::MIN,
)?);
let (paused_tx, paused_rx) = mpsc::channel();
let (release_tx, release_rx) = mpsc::channel();
let rebuilding = Arc::clone(&service);
let worker = std::thread::spawn(move || {
Storage::arm_count_rebuild_pause(paused_tx, release_rx);
rebuilding.dispatch(Operation::CountsRebuild(CountsRebuild {
branch: "main".to_owned(),
request_id: "held".to_owned(),
}))
});
paused_rx.recv()?;
let raced = service.dispatch(Operation::Mutate(request));
let read = service.dispatch(Operation::CountsAudit(Selection {
branch: "main".to_owned(),
version: None,
}));
release_tx.send(())?;
let built = worker.join().map_err(|_| "rebuild thread panicked")?;
let error = serde_json::to_value(raced.err().ok_or("write escaped admission")?)?;
assert_eq!(error["code"], "busy");
assert_eq!(error["outcome"], "not_applied");
assert_eq!(
read?["selection"]["version"]["sequence"],
before.sequence.to_string()
);
let next = built?["version"]["sequence"]
.as_str()
.ok_or("no sequence")?
.parse::<u64>()?;
assert_eq!(next, before.sequence + 1);
service.dispatch(Operation::Mutate(Mutation {
request_id: "after".to_owned(),
branch: "main".to_owned(),
expected: crate::serve::protocol::Version {
sequence: next,
..before
},
change: put("after")?,
}))?;
let answer = service.dispatch(Operation::CountsAudit(Selection {
branch: "main".to_owned(),
version: None,
}))?;
assert!(
answer["buckets"]
.as_array()
.ok_or("no buckets")?
.iter()
.all(|row| row["result"] == "agree")
);
drop(service);
drop(directory);
Ok(())
}
#[test]
fn rebuild_response_limit_is_unknown_and_receipt_reconciles_without_second_commit() -> TestResult {
use crate::serve::protocol::Operation;
let fixture = populated()?;
let before = fixture.at()?.version;
let Fixture {
storage, directory, ..
} = fixture;
let service = crate::serve::service::Service::new(storage, NonZeroU32::MIN)?;
let error = service
.dispatch_bounded(
Operation::CountsRebuild(rebuild_request("limited")),
std::num::NonZeroUsize::MIN,
)
.err()
.ok_or("large answer was admitted")?;
let error = serde_json::to_value(error)?;
assert_eq!(error["outcome"], "unknown");
assert_eq!(error["request_id"], "limited");
assert_eq!(error["receipt_branch"], "main");
let receipt = service.dispatch(Operation::Receipt(ReceiptQuery {
branch: "main".to_owned(),
request_id: "limited".to_owned(),
}))?;
let replay = service.dispatch(Operation::CountsRebuild(rebuild_request("limited")))?;
assert_eq!(replay, receipt);
assert_eq!(
replay["version"]["sequence"],
(before.sequence + 1).to_string()
);
drop(service);
drop(directory);
Ok(())
}
#[test]
fn rebuild_protocol_requires_identity_and_malformed_request_is_not_applied() -> TestResult {
let request = crate::serve::protocol::decode_request(
br#"{"id":"decode","operation":{"method":"counts.rebuild","params":{"branch":"main"}}}"#,
)?;
let crate::serve::protocol::DecodedRequest::Refused { failure, .. } = request else {
return Err("missing rebuild request identity was admitted".into());
};
let answer = serde_json::to_value(failure)?;
assert_eq!(answer["code"], "invalid_params");
assert_eq!(answer["outcome"], "not_applied");
assert!(answer["request_id"].is_null());
assert_eq!(answer["receipt_branch"], "main");
Ok(())
}
#[test]
fn rebuild_scan_refusal_is_not_applied_and_leaves_no_receipt() -> TestResult {
use crate::serve::protocol::Operation;
let mut fixture = populated()?;
fixture.edit(READY_KEY.to_vec(), None)?;
fixture.edit(b"d\0malformed".to_vec(), Some(b"{}".to_vec()))?;
let before = fixture.at()?.version;
let Fixture {
storage, directory, ..
} = fixture;
let service = crate::serve::service::Service::new(storage, NonZeroU32::MIN)?;
let error = service
.dispatch(Operation::CountsRebuild(rebuild_request("scan-refusal")))
.err()
.ok_or("bad scan was admitted")?;
let error = serde_json::to_value(error)?;
assert_eq!(error["code"], "counts_scan_refused");
assert_eq!(error["outcome"], "not_applied");
assert_eq!(error["request_id"], "scan-refusal");
assert!(
service
.dispatch(Operation::Receipt(ReceiptQuery {
branch: "main".to_owned(),
request_id: "scan-refusal".to_owned(),
}))?
.is_null()
);
let answer = service.dispatch(Operation::CountsAudit(Selection {
branch: "main".to_owned(),
version: None,
}))?;
assert_eq!(answer["reason"], "counts_not_ready");
assert_eq!(
answer["selection"]["version"]["sequence"],
before.sequence.to_string()
);
drop(service);
drop(directory);
Ok(())
}
#[test]
fn audit_can_select_the_unready_kept_version_after_rebuild() -> TestResult {
use crate::serve::protocol::Operation;
let mut fixture = populated()?;
fixture.edit(READY_KEY.to_vec(), None)?;
let before = fixture.at()?.version;
fixture
.storage
.rebuild_counts(&rebuild_request("historical"))?;
let Fixture {
storage, directory, ..
} = fixture;
let service = crate::serve::service::Service::new(storage, NonZeroU32::MIN)?;
let answer = service.dispatch(Operation::CountsAudit(Selection {
branch: "main".to_owned(),
version: Some(before),
}))?;
assert_eq!(answer["result"], "cannot_tell");
assert_eq!(answer["reason"], "counts_not_ready");
assert_eq!(
answer["selection"]["version"]["sequence"],
before.sequence.to_string()
);
drop(service);
drop(directory);
Ok(())
}
#[test]
fn rebuild_crash_before_commit_has_no_rows_or_mark_after_reopen() -> TestResult {
let mut fixture = populated()?;
fixture.edit_batch(vec![
(READY_KEY.to_vec(), None),
(document_count_key("records"), None),
(EDGE_COUNT.to_vec(), None),
])?;
let before = fixture.at()?.version;
super::rebuild_probe::crash_next();
let crashed = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
fixture.storage.rebuild_counts(&rebuild_request("crash"))
}));
assert!(crashed.is_err());
let fixture = fixture.reopen()?;
assert_eq!(fixture.at()?.version, before);
assert!(!is_ready(&fixture.storage.reader, &fixture.at()?)?);
assert!(
reads::get_bytes(
&fixture.storage.reader,
&fixture.at()?,
&document_count_key("records")
)?
.is_none()
);
assert!(reads::get_bytes(&fixture.storage.reader, &fixture.at()?, EDGE_COUNT)?.is_none());
assert!(
fixture
.storage
.capture(&Selection {
branch: "main".to_owned(),
version: Some(before),
})
.is_ok()
);
Ok(())
}
#[test]
fn whole_collection_and_store_authority_is_explicit() -> TestResult {
use crate::serve::access::{Standalone, TestSource, request};
let owner = request(&Standalone)?;
assert!(owner.can_read_collection("records"));
assert!(owner.can_read_store());
let restricted = request(&TestSource::denying(&[("records", "record-0")]))?;
assert!(!restricted.can_read_collection("records"));
assert!(restricted.can_read_collection("other"));
assert!(!restricted.can_read_store());
Ok(())
}
#[test]
fn mixed_collection_authority_withholds_global_totals() -> TestResult {
use crate::serve::access::{TestSource, request};
let source = TestSource::denying(&[("records", "record-0")]).allowing_collection("other");
let restricted = request(&source)?;
assert!(!restricted.can_read_collection("records"));
assert!(restricted.can_read_collection("other"));
assert!(!restricted.can_read_store());
let fixture = populated()?;
let stats = reads::stats(&restricted, &fixture.storage.reader()?, &fixture.at()?)?;
for total in [
&stats.collection_count,
&stats.relation_count,
&stats.documents,
&stats.edges,
&stats.vectors,
] {
assert_eq!(total.value, None);
assert_eq!(total.reason, Some(CountReason::Withheld));
}
Ok(())
}
#[test]
fn kept_document_total_is_pinned_without_reading_records() -> TestResult {
let mut fixture = populated()?;
let owner = crate::serve::access::request(&crate::serve::access::Standalone)?;
let before = fixture.at()?;
fixture.apply(put("third")?)?;
let after = fixture.at()?;
for (at, expected) in [(&before, 2), (&after, 3)] {
probe::take();
let view = super::View::new(&fixture.storage.reader, at)?;
let total = view.documents(&owner, "records")?;
assert_eq!(total.value, Some(expected));
assert!(total.reason.is_none());
let observed = probe::take();
assert_eq!(observed.ready, 1);
assert_eq!(observed.counts, 1);
assert_eq!(observed.bodies, 0);
assert_eq!(observed.objects, 0);
assert_eq!(observed.revisions, 0);
}
Ok(())
}
#[test]
fn three_total_reads_match_full_counts_at_current_and_kept_versions() -> TestResult {
let mut fixture = populated()?;
fixture.apply(Change::EdgePut(edge("record-0", "record-1")))?;
fixture.apply(vector("record-0"))?;
let before = fixture.at()?;
fixture.apply(delete("record-0"))?;
let after = fixture.at()?;
let owner = crate::serve::access::request(&crate::serve::access::Standalone)?;
for at in [&before, &after] {
let documents = reads::test_support::count_keys(&fixture.storage.reader, at, b"d\0", 2)?;
let relationships =
reads::test_support::count_keys(&fixture.storage.reader, at, b"e\0", 5)?;
let vectors = reads::test_support::count_keys(&fixture.storage.reader, at, b"v\0", 3)?;
reads::take_content_reads();
let stats = reads::stats(&owner, &fixture.storage.reader, at)?;
assert_eq!(stats.documents.value, Some(documents));
assert_eq!(stats.edges.value, Some(relationships));
assert_eq!(stats.vectors.value, Some(vectors));
let collections = reads::collections(
&fixture.storage.reader,
at,
None,
std::num::NonZeroUsize::MIN,
)?;
let totals =
reads::collection_totals(&owner, &fixture.storage.reader, at, &collections.entries)?;
assert_eq!(totals[0].value, Some(documents));
let relations = crate::serve::storage::relations::list(
&owner,
&fixture.storage.reader,
at,
std::num::NonZeroUsize::MAX,
)?;
assert_eq!(relations[0].relationships, Some(relationships));
assert_eq!(reads::take_content_reads(), 0, "a total read content rows");
}
Ok(())
}
#[test]
fn statistics_withhold_totals_and_unready_keeps_metadata() -> TestResult {
let mut fixture = populated()?;
let restricted =
crate::serve::access::request(&crate::serve::access::TestSource::denying(&[(
"records", "record-0",
)]))?;
reads::take_content_reads();
let stats = reads::stats(&restricted, &fixture.storage.reader, &fixture.at()?)?;
for total in [
stats.documents,
stats.edges,
stats.vectors,
stats.collection_count,
stats.relation_count,
] {
assert!(total.value.is_none());
assert_eq!(serde_json::to_value(total.reason)?, "withheld");
}
let collections = serde_json::to_value(stats.collections)?;
assert_eq!(collections[0]["name"], "records");
assert!(collections[0]["documents"].is_null());
assert_eq!(collections[0]["documents_reason"], "withheld");
assert_eq!(reads::take_content_reads(), 0);
fixture.edit(READY_KEY.to_vec(), None)?;
let owner = crate::serve::access::request(&crate::serve::access::Standalone)?;
let stats = reads::stats(&owner, &fixture.storage.reader, &fixture.at()?)?;
assert_eq!(stats.collection_count.value, Some(1));
assert_eq!(stats.relation_count.value, Some(1));
for total in [stats.documents, stats.edges, stats.vectors] {
assert!(total.value.is_none());
assert_eq!(serde_json::to_value(total.reason)?, "counts_not_ready");
}
let collections = serde_json::to_value(stats.collections)?;
assert_eq!(collections[0]["name"], "records");
assert_eq!(collections[0]["documents_reason"], "counts_not_ready");
Ok(())
}
#[test]
fn missing_declared_count_refuses_total_reads() -> TestResult {
let mut fixture = populated()?;
fixture.edit(EDGE_COUNT.to_vec(), None)?;
let owner = crate::serve::access::request(&crate::serve::access::Standalone)?;
let at = fixture.at()?;
let relations = crate::serve::storage::relations::list(
&owner,
&fixture.storage.reader,
&at,
std::num::NonZeroUsize::MAX,
);
assert_eq!(
relations.err().ok_or("missing relation refusal")?.code,
"counts_corrupt"
);
let stats = reads::stats(&owner, &fixture.storage.reader, &at);
assert_eq!(
stats.err().ok_or("missing stats refusal")?.code,
"counts_corrupt"
);
Ok(())
}
#[test]
fn undeclared_kind_and_removed_declaration_remain_in_global_total() -> TestResult {
let mut fixture = populated()?;
fixture.apply(Change::EdgePut(edge("record-0", "record-1")))?;
fixture.apply(Change::RelationDelete(
crate::serve::protocol::RelationDelete {
relation: "linked".to_owned(),
},
))?;
let mut undeclared = edge("record-0", "record-1");
undeclared.relation = "free".to_owned();
fixture.apply(Change::EdgePut(undeclared))?;
let owner = crate::serve::access::request(&crate::serve::access::Standalone)?;
let at = fixture.at()?;
let stats = reads::stats(&owner, &fixture.storage.reader, &at)?;
assert_eq!(stats.edges.value, Some(2));
assert_eq!(stored_count(&fixture, EDGE_COUNT, &at)?, 1);
assert_eq!(stored_count(&fixture, b"k\0e\0free", &at)?, 1);
let relations = crate::serve::storage::relations::list(
&owner,
&fixture.storage.reader,
&at,
std::num::NonZeroUsize::MAX,
)?;
assert!(relations.is_empty());
fixture.edit(EDGE_COUNT.to_vec(), None)?;
assert_refused(
&mut fixture,
edge_delete("record-0", "record-1"),
"counts_corrupt",
)?;
Ok(())
}
#[test]
fn served_total_answers_keep_members_and_omit_known_reasons() -> TestResult {
use crate::serve::protocol::{CollectionPage, Operation, RelationsRead, StoreRead};
let fixture = populated()?;
let at = fixture.at()?;
let Fixture {
storage, directory, ..
} = fixture;
let service = crate::serve::service::Service::new(storage, NonZeroU32::MIN)?;
let selection = || Selection {
branch: "main".to_owned(),
version: Some(at.version),
};
reads::take_content_reads();
let collections = service.dispatch(Operation::Collections(CollectionPage {
selection: selection(),
after: None,
limit: NonZeroU32::MIN,
}))?;
assert_eq!(collections["entries"][0]["documents"], 2);
assert!(collections["entries"][0].get("documents_reason").is_none());
let relations = service.dispatch(Operation::Relations(RelationsRead {
selection: selection(),
}))?;
assert_eq!(relations["relations"][0]["relationships"], 0);
assert!(
relations["relations"][0]
.get("relationships_reason")
.is_none()
);
let stats = service.dispatch(Operation::StoreStats(StoreRead {
selection: selection(),
}))?;
assert_eq!(stats["documents"], 2);
assert_eq!(stats["edges"], 0);
assert_eq!(stats["vectors"], 0);
assert_eq!(stats["collection_count"], 1);
assert_eq!(stats["relation_count"], 1);
assert!(stats.get("documents_reason").is_none());
assert_eq!(reads::take_content_reads(), 0);
drop(service);
drop(directory);
Ok(())
}
#[test]
fn unready_service_answers_have_named_nulls_and_keep_metadata() -> TestResult {
use crate::serve::protocol::{CollectionPage, Operation, RelationsRead, StoreRead};
let mut fixture = populated()?;
fixture.edit(READY_KEY.to_vec(), None)?;
let Fixture {
storage, directory, ..
} = fixture;
let service = crate::serve::service::Service::new(storage, NonZeroU32::MIN)?;
let selection = || Selection {
branch: "main".to_owned(),
version: None,
};
let collections = service.dispatch(Operation::Collections(CollectionPage {
selection: selection(),
after: None,
limit: NonZeroU32::MIN,
}))?;
assert_eq!(collections["entries"][0]["name"], "records");
assert!(collections["entries"][0]["documents"].is_null());
assert_eq!(
collections["entries"][0]["documents_reason"],
"counts_not_ready"
);
let relations = service.dispatch(Operation::Relations(RelationsRead {
selection: selection(),
}))?;
assert_eq!(relations["relations"][0]["relation"], "linked");
assert!(relations["relations"][0]["relationships"].is_null());
assert_eq!(
relations["relations"][0]["relationships_reason"],
"counts_not_ready"
);
let stats = service.dispatch(Operation::StoreStats(StoreRead {
selection: selection(),
}))?;
for name in ["documents", "edges", "vectors"] {
assert!(stats[name].is_null());
assert_eq!(stats[format!("{name}_reason")], "counts_not_ready");
}
assert_eq!(stats["collection_count"], 1);
assert_eq!(stats["relation_count"], 1);
assert!(stats.get("relation_count_reason").is_none());
drop(service);
drop(directory);
Ok(())
}
#[test]
fn document_total_withheld_or_unready_is_named_without_record_reads() -> TestResult {
let mut fixture = populated()?;
let restricted =
crate::serve::access::request(&crate::serve::access::TestSource::denying(&[(
"records", "record-0",
)]))?;
probe::take();
let at = fixture.at()?;
let view = super::View::new(&fixture.storage.reader, &at)?;
let total = view.documents(&restricted, "records")?;
assert_eq!(total.value, None);
assert_eq!(serde_json::to_value(total.reason)?, "withheld");
assert_eq!(probe::take().counts, 0);
fixture.edit(READY_KEY.to_vec(), None)?;
let owner = crate::serve::access::request(&crate::serve::access::Standalone)?;
probe::take();
let at = fixture.at()?;
let view = super::View::new(&fixture.storage.reader, &at)?;
let total = view.documents(&owner, "records")?;
assert_eq!(total.value, None);
assert_eq!(serde_json::to_value(total.reason)?, "counts_not_ready");
let observed = probe::take();
assert_eq!(observed.counts, 0);
assert_eq!(observed.bodies, 0);
Ok(())
}
#[test]
fn relation_declaration_writes_zero_once_and_missing_declared_count_refuses() -> TestResult {
let mut fixture = populated()?;
assert_eq!(stored_count(&fixture, EDGE_COUNT, &fixture.at()?)?, 0);
probe::take();
fixture.apply(Change::RelationPut(Relation {
relation: "linked".to_owned(),
from_collection: "records".to_owned(),
to_collection: "records".to_owned(),
cardinality: Cardinality::ManyToMany,
}))?;
assert_eq!(probe::take().count_writes, 0);
fixture.edit(EDGE_COUNT.to_vec(), None)?;
assert_refused(
&mut fixture,
Change::EdgePut(edge("record-0", "record-1")),
"counts_corrupt",
)?;
Ok(())
}
#[test]
fn declared_field_missing_count_refuses_even_when_its_vector_is_absent() -> TestResult {
let mut fixture = populated()?;
fixture.apply(vector("record-0"))?;
fixture.apply(vector_delete("record-0"))?;
assert_eq!(stored_count(&fixture, VECTOR_COUNT, &fixture.at()?)?, 0);
fixture.edit(VECTOR_COUNT.to_vec(), None)?;
assert_refused(&mut fixture, vector("record-1"), "counts_corrupt")?;
Ok(())
}
fn node(id: &str) -> Node {
Node {
collection: "records".to_owned(),
document_id: id.to_owned(),
}
}
fn edge(from: &str, to: &str) -> Edge {
Edge {
from: node(from),
relation: "linked".to_owned(),
to: node(to),
properties: serde_json::json!({"weight": 1}),
}
}
fn edge_delete(from: &str, to: &str) -> Change {
Change::EdgeDelete(EdgeDelete {
from: node(from),
relation: "linked".to_owned(),
to: node(to),
})
}
fn vector(id: &str) -> Change {
Change::VectorPut(VectorWrite {
collection: "records".to_owned(),
field: "embedding".to_owned(),
document_id: id.to_owned(),
vector: vec![1.0, 2.0],
})
}
fn vector_delete(id: &str) -> Change {
Change::VectorDelete(VectorDelete {
collection: "records".to_owned(),
field: "embedding".to_owned(),
document_id: id.to_owned(),
})
}
const EDGE_COUNT: &[u8] = b"k\0e\0linked";
const VECTOR_COUNT: &[u8] = b"k\0v\0records\0embedding";
fn stored_count(fixture: &Fixture, key: &[u8], at: &Captured) -> Result<u64, Box<dyn Error>> {
let bytes = reads::get_bytes(&fixture.storage.reader, at, key)?.ok_or("count missing")?;
Ok(super::decode(&bytes)?)
}
fn populated() -> Result<Fixture, Box<dyn Error>> {
let mut fixture = Fixture::new()?;
fixture.collection()?;
fixture.apply(import(2)?)?;
fixture.apply(Change::RelationPut(Relation {
relation: "linked".to_owned(),
from_collection: "records".to_owned(),
to_collection: "records".to_owned(),
cardinality: Cardinality::ManyToMany,
}))?;
Ok(fixture)
}
#[test]
fn edge_add_replace_delete_counts_one_canonical_edge() -> TestResult {
let mut fixture = populated()?;
fixture.apply(Change::EdgePut(edge("record-0", "record-0")))?;
let inserted = fixture.at()?;
assert_eq!(stored_count(&fixture, EDGE_COUNT, &inserted)?, 1);
let (out, incoming) = edges::keys(&node("record-0"), "linked", &node("record-0"))?;
assert!(reads::get_bytes(&fixture.storage.reader, &inserted, &out)?.is_some());
assert!(reads::get_bytes(&fixture.storage.reader, &inserted, &incoming)?.is_some());
let mut replaced = edge("record-0", "record-0");
replaced.properties = serde_json::json!({"weight": 2});
probe::take();
fixture.apply(Change::EdgePut(replaced))?;
assert_eq!(probe::take().count_writes, 0);
assert_eq!(stored_count(&fixture, EDGE_COUNT, &fixture.at()?)?, 1);
fixture.apply(edge_delete("record-0", "record-0"))?;
assert_eq!(stored_count(&fixture, EDGE_COUNT, &fixture.at()?)?, 0);
assert_eq!(stored_count(&fixture, EDGE_COUNT, &inserted)?, 1);
probe::take();
fixture.apply(edge_delete("record-0", "record-0"))?;
assert_eq!(probe::take().count_writes, 0);
assert_eq!(stored_count(&fixture, EDGE_COUNT, &fixture.at()?)?, 0);
Ok(())
}
#[test]
fn vector_add_replace_delete_keeps_field_count() -> TestResult {
let mut fixture = populated()?;
fixture.apply(vector("record-0"))?;
let inserted = fixture.at()?;
assert_eq!(stored_count(&fixture, VECTOR_COUNT, &inserted)?, 1);
probe::take();
fixture.apply(Change::VectorPut(VectorWrite {
collection: "records".to_owned(),
field: "embedding".to_owned(),
document_id: "record-0".to_owned(),
vector: vec![3.0, 4.0],
}))?;
assert_eq!(probe::take().count_writes, 0);
assert_eq!(stored_count(&fixture, VECTOR_COUNT, &fixture.at()?)?, 1);
fixture.apply(vector_delete("record-0"))?;
assert_eq!(stored_count(&fixture, VECTOR_COUNT, &fixture.at()?)?, 0);
assert_eq!(stored_count(&fixture, VECTOR_COUNT, &inserted)?, 1);
probe::take();
fixture.apply(vector_delete("record-0"))?;
assert_eq!(probe::take().count_writes, 0);
Ok(())
}
#[test]
fn document_delete_updates_incident_edges_and_present_vectors_only() -> TestResult {
let mut fixture = populated()?;
fixture.apply(Change::EdgePut(edge("record-0", "record-1")))?;
fixture.apply(Change::EdgePut(edge("record-1", "record-0")))?;
fixture.apply(Change::EdgePut(edge("record-0", "record-0")))?;
fixture.apply(vector("record-0"))?;
fixture.apply(vector("record-1"))?;
let before = fixture.at()?;
fixture.apply(delete("record-0"))?;
let after = fixture.at()?;
assert_eq!(stored_count(&fixture, EDGE_COUNT, &after)?, 0);
assert_eq!(stored_count(&fixture, VECTOR_COUNT, &after)?, 1);
assert_eq!(fixture.count()?, 1);
assert_eq!(stored_count(&fixture, EDGE_COUNT, &before)?, 3);
assert_eq!(stored_count(&fixture, VECTOR_COUNT, &before)?, 2);
fixture.apply(delete("record-0"))?;
assert_eq!(stored_count(&fixture, VECTOR_COUNT, &fixture.at()?)?, 1);
Ok(())
}
#[test]
fn prepared_batch_aggregates_one_count_per_bucket() -> TestResult {
let mut fixture = populated()?;
let at = fixture.at()?;
let mut writes: Vec<PreparedWrite> = Vec::new();
for (from, to) in [("record-0", "record-1"), ("record-1", "record-0")] {
writes.extend(edges::put(&fixture.storage.reader, &at, &edge(from, to))?);
}
for id in ["record-0", "record-1"] {
writes.extend(
fixture
.storage
.prepare_change(&at, &vector(id), at.version)?,
);
}
let repeated = writes.first().ok_or("no prepared edge")?.clone();
writes.push(repeated);
probe::take();
super::append(
&fixture.storage.reader,
&at,
&edge_delete("absent", "absent"),
&mut writes,
)?;
assert_eq!(probe::take().count_writes, 2);
for key in [EDGE_COUNT, VECTOR_COUNT] {
let rows: Vec<_> = writes
.iter()
.filter(|(stored, _)| stored.as_slice() == key)
.collect();
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].1.as_deref(), Some(b"2".as_slice()));
}
fixture.edit_batch(writes)?;
assert_eq!(fixture.at()?.version.sequence, at.version.sequence + 1);
assert_eq!(stored_count(&fixture, EDGE_COUNT, &fixture.at()?)?, 2);
assert_eq!(stored_count(&fixture, VECTOR_COUNT, &fixture.at()?)?, 2);
Ok(())
}
fn assert_refused(fixture: &mut Fixture, change: Change, code: &str) -> TestResult {
let before = fixture.at()?;
let request = fixture.request(change)?;
let result = fixture.storage.mutate(&request);
assert!(result.is_err(), "count refusal missing");
assert_eq!(result.err().ok_or("no refusal")?.code, code);
assert_eq!(fixture.at()?.version, before.version);
let receipt = ReceiptQuery {
branch: "main".to_owned(),
request_id: request.request_id,
};
assert!(fixture.storage.receipt(&receipt)?.is_none());
Ok(())
}
#[test]
fn edge_and_vector_overflow_underflow_refuse_before_staging() -> TestResult {
let mut fixture = populated()?;
for is_edge in [true, false] {
let key = if is_edge { EDGE_COUNT } else { VECTOR_COUNT };
fixture.edit(key.to_vec(), Some(u64::MAX.to_string().into_bytes()))?;
let put = if is_edge {
Change::EdgePut(edge("record-0", "record-1"))
} else {
vector("record-0")
};
assert_refused(&mut fixture, put, "count_overflow")?;
fixture.edit(key.to_vec(), Some(b"0".to_vec()))?;
let put = if is_edge {
Change::EdgePut(edge("record-0", "record-1"))
} else {
vector("record-0")
};
fixture.apply(put)?;
fixture.edit(key.to_vec(), Some(b"0".to_vec()))?;
let delete = if is_edge {
edge_delete("record-0", "record-1")
} else {
vector_delete("record-0")
};
assert_refused(&mut fixture, delete, "count_underflow")?;
assert_eq!(stored_count(&fixture, key, &fixture.at()?)?, 0);
}
Ok(())
}
#[test]
fn existing_edge_and_vector_require_canonical_counts() -> TestResult {
let mut fixture = populated()?;
fixture.apply(Change::EdgePut(edge("record-0", "record-1")))?;
fixture.apply(vector("record-0"))?;
for is_edge in [true, false] {
let key = if is_edge { EDGE_COUNT } else { VECTOR_COUNT };
for value in [None, Some(b"01".to_vec())] {
fixture.edit(key.to_vec(), value)?;
let change = if is_edge {
Change::EdgePut(edge("record-0", "record-1"))
} else {
vector("record-0")
};
assert_refused(&mut fixture, change, "counts_corrupt")?;
}
}
Ok(())
}
#[test]
fn edge_vector_replay_and_unready_writes_do_not_move_counts() -> TestResult {
let mut fixture = populated()?;
for change in [
Change::EdgePut(edge("record-0", "record-1")),
vector("record-0"),
] {
let request = fixture.request(change)?;
let first = fixture.storage.mutate(&request)?;
probe::take();
let repeated = fixture.storage.mutate(&request)?;
assert_eq!(first.version, repeated.version);
assert_eq!(probe::take(), probe::Reads::default());
}
let mut unready = populated()?;
unready.edit(READY_KEY.to_vec(), None)?;
unready.apply(Change::EdgePut(edge("record-0", "record-1")))?;
unready.apply(vector("record-0"))?;
unready.apply(delete("record-0"))?;
let at = unready.at()?;
for key in [READY_KEY, VECTOR_COUNT] {
assert!(reads::get_bytes(&unready.storage.reader, &at, key)?.is_none());
}
assert_eq!(stored_count(&unready, EDGE_COUNT, &at)?, 0);
Ok(())
}
#[test]
fn missing_edge_and_vector_delete_do_not_create_counts() -> TestResult {
let mut fixture = populated()?;
fixture.apply(edge_delete("record-0", "record-1"))?;
fixture.apply(vector_delete("record-0"))?;
let at = fixture.at()?;
assert_eq!(stored_count(&fixture, EDGE_COUNT, &at)?, 0);
assert!(reads::get_bytes(&fixture.storage.reader, &at, VECTOR_COUNT)?.is_none());
Ok(())
}
#[test]
fn edge_vector_presence_uses_own_keys_and_three_point_reads() -> TestResult {
let mut fixture = populated()?;
for change in [
Change::EdgePut(edge("record-0", "record-1")),
vector("record-0"),
] {
probe::take();
fixture.apply(change)?;
let observed = probe::take();
assert_eq!(observed.ready, 1);
assert_eq!(observed.objects, 1);
assert_eq!(observed.counts, 1);
assert_eq!(observed.bodies, 0);
}
fixture.edit(document_key("records", "record-0"), None)?;
fixture.apply(edge_delete("record-0", "record-1"))?;
fixture.apply(vector_delete("record-0"))?;
assert_eq!(stored_count(&fixture, EDGE_COUNT, &fixture.at()?)?, 0);
assert_eq!(stored_count(&fixture, VECTOR_COUNT, &fixture.at()?)?, 0);
Ok(())
}
fn classified_count_refusal(code: &str) -> TestResult {
let failure = ResponseFailure::new(
Failure::new(code, "count refused"),
Some(MutationContext {
outcome: MutationOutcome::Unknown,
request_id: serde_json::json!("count-request"),
receipt_branch: serde_json::json!("main"),
}),
);
let value = serde_json::to_value(failure)?;
assert_eq!(value["outcome"], "not_applied");
assert_eq!(value["code"], code);
assert_eq!(value["request_id"], "count-request");
assert_eq!(value["receipt_branch"], "main");
Ok(())
}
#[test]
fn counts_corrupt_reports_not_applied() -> TestResult {
classified_count_refusal("counts_corrupt")
}
#[test]
fn count_overflow_reports_not_applied() -> TestResult {
classified_count_refusal("count_overflow")
}
#[test]
fn count_underflow_reports_not_applied() -> TestResult {
classified_count_refusal("count_underflow")
}
struct Fixture {
storage: Storage,
directory: tempfile::TempDir,
request: u64,
}
impl Fixture {
fn new() -> Result<Self, 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(Self {
storage,
directory,
request: 0,
})
}
fn at(&self) -> Result<Captured, Box<dyn Error>> {
Ok(self.storage.capture(&Selection {
branch: "main".to_owned(),
version: None,
})?)
}
fn request(&mut self, change: Change) -> Result<Mutation, Box<dyn Error>> {
self.request = self.request.checked_add(1).ok_or("request exhausted")?;
Ok(Mutation {
request_id: format!("request-{}", self.request),
branch: "main".to_owned(),
expected: self.at()?.version,
change,
})
}
fn apply(&mut self, change: Change) -> TestResult {
let request = self.request(change)?;
self.storage
.mutate_bounded(&request, NonZeroU32::new(64).ok_or("zero page")?)?;
Ok(())
}
fn collection(&mut self) -> TestResult {
self.apply(Change::Collection(CollectionWrite {
collection: "records".to_owned(),
schema: None,
}))
}
fn count(&self) -> Result<u64, Box<dyn Error>> {
Ok(required_documents(
&self.storage.reader,
&self.at()?,
"records",
)?)
}
fn edit(&mut self, key: Vec<u8>, value: Option<Vec<u8>>) -> TestResult {
self.edit_batch(vec![(key, value)])
}
fn edit_batch(&mut self, writes: Vec<PreparedWrite>) -> TestResult {
let name = branch_name("main");
let branch = open(&name, &self.storage.refs, &self.storage.registry)?;
for (key, value) in writes {
let shard = self.storage.writer.shard_for(&key);
match value {
Some(value) => branch.put(shard, key, value)?,
None => branch.delete(shard, key)?,
}
}
self.storage.writer.commit(
&branch,
&self.storage.registry,
CommitRequest {
durability: CommitDurability::Durable {
refs: &mut self.storage.refs,
},
extra_parents: &[],
timestamp: current_timestamp(),
},
)?;
self.storage.publish_record(&name)?;
Ok(())
}
fn reopen(self) -> Result<Self, Box<dyn Error>> {
let Self {
storage,
directory,
request,
} = self;
drop(storage);
let storage = Storage::open(
&directory.path().join("database"),
NodeCacheBudget::Unlimited,
fixture_long_wait_interval(),
)?;
Ok(Self {
storage,
directory,
request,
})
}
}
fn put(id: &str) -> Result<Change, Box<dyn Error>> {
Ok(Change::Put(DocumentWrite {
collection: "records".to_owned(),
document_id: id.to_owned(),
document: RawValue::from_string("{\"value\":1}".to_owned())?,
}))
}
fn delete(id: &str) -> Change {
Change::Delete(DocumentDelete {
collection: "records".to_owned(),
document_id: id.to_owned(),
})
}
fn import(rows: usize) -> Result<Change, Box<dyn Error>> {
let documents = (0..rows)
.map(|index| {
Ok(ImportedDocument {
document_id: format!("record-{index}"),
document: RawValue::from_string("{\"value\":1}".to_owned())?,
})
})
.collect::<Result<Vec<_>, Box<dyn Error>>>()?;
Ok(Change::PutMany(DocumentImport {
collection: "records".to_owned(),
documents,
}))
}
#[test]
fn new_store_and_new_collection_have_explicit_ready_and_zero() -> TestResult {
let mut fixture = Fixture::new()?;
assert_eq!(
reads::get_bytes(&fixture.storage.reader, &fixture.at()?, READY_KEY)?,
Some(READY_VALUE.to_vec()),
);
assert!(is_ready(&fixture.storage.reader, &fixture.at()?)?);
fixture.collection()?;
assert_eq!(
reads::get_bytes(
&fixture.storage.reader,
&fixture.at()?,
&document_count_key("records")
)?,
Some(b"0".to_vec()),
);
assert_eq!(fixture.at()?.version.sequence, 1);
let fixture = fixture.reopen()?;
assert!(is_ready(&fixture.storage.reader, &fixture.at()?)?);
assert_eq!(fixture.count()?, 0);
Ok(())
}
#[test]
fn new_replace_and_delete_keep_document_counts_in_their_versions() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
let empty = fixture.at()?;
fixture.apply(put("one")?)?;
assert_eq!(fixture.count()?, 1);
assert_eq!(
required_documents(&fixture.storage.reader, &empty, "records")?,
0
);
probe::take();
fixture.apply(put("one")?)?;
assert_eq!(probe::take().count_writes, 0, "replacement writes no count");
assert_eq!(fixture.count()?, 1);
fixture.apply(delete("one"))?;
assert_eq!(fixture.count()?, 0);
probe::take();
fixture.apply(delete("one"))?;
assert_eq!(
probe::take().count_writes,
0,
"absent delete writes no count"
);
assert_eq!(fixture.count()?, 0);
Ok(())
}
#[test]
fn collection_replacement_preserves_its_count_without_a_count_write() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
fixture.apply(put("one")?)?;
probe::take();
fixture.collection()?;
assert_eq!(probe::take().count_writes, 0);
assert_eq!(fixture.count()?, 1);
Ok(())
}
#[test]
fn mixed_import_reads_revisions_only_and_writes_one_count() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
fixture.apply(import(20)?)?;
probe::take();
fixture.apply(import(2000)?)?;
let observed = probe::take();
assert_eq!(observed.revisions, 2000);
assert_eq!(observed.bodies, 0, "existence never reads an old body");
assert_eq!(observed.counts, 1);
assert_eq!(observed.ready, 1);
assert_eq!(
observed.count_writes, 1,
"one count write for the entire import"
);
assert_eq!(fixture.count()?, 2000);
Ok(())
}
#[test]
fn receipt_replay_does_not_read_or_write_counts() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
let request = fixture.request(put("one")?)?;
let first = fixture.storage.mutate(&request)?;
let at = fixture.at()?;
probe::take();
let repeated = fixture.storage.mutate(&request);
assert!(
repeated.is_ok(),
"receipt replay must succeed without applying the request again"
);
let repeated = repeated?;
assert_eq!(probe::take(), probe::Reads::default());
assert_eq!(first.version, repeated.version);
assert_eq!(fixture.at()?.version, at.version);
assert_eq!(fixture.count()?, 1);
Ok(())
}
#[test]
fn unready_writes_succeed_without_count_rows_or_ready_mark() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.edit(READY_KEY.to_vec(), None)?;
fixture.collection()?;
fixture.apply(put("one")?)?;
fixture.apply(import(2)?)?;
fixture.apply(delete("one"))?;
let at = fixture.at()?;
assert!(!is_ready(&fixture.storage.reader, &at)?);
let end = b"k\x01";
for shard in &at.shards {
let page = fixture.storage.reader.range_page_at(
shard.shard_id,
shard.head,
b"k\0",
end,
std::num::NonZeroUsize::MIN,
)?;
assert!(
page.entries.is_empty(),
"unready writes created count metadata"
);
}
let fixture = fixture.reopen()?;
assert!(!is_ready(&fixture.storage.reader, &fixture.at()?)?);
Ok(())
}
#[test]
fn overflow_and_underflow_refuse_before_staging() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
fixture.edit(
document_count_key("records"),
Some(u64::MAX.to_string().into_bytes()),
)?;
let request = fixture.request(put("one")?)?;
let before = fixture.at()?;
let result = fixture.storage.mutate(&request);
assert!(result.is_err(), "overflow was accepted");
assert_eq!(
result.err().ok_or("overflow error missing")?.code,
"count_overflow"
);
assert_eq!(fixture.at()?.version, before.version);
assert!(
reads::get_bytes(
&fixture.storage.reader,
&before,
&revision_key("records", "one"),
)?
.is_none()
);
assert!(
fixture
.storage
.receipt(&ReceiptQuery {
branch: "main".to_owned(),
request_id: request.request_id,
})?
.is_none()
);
fixture.edit(document_count_key("records"), Some(b"0".to_vec()))?;
fixture.apply(put("one")?)?;
fixture.edit(document_count_key("records"), Some(b"0".to_vec()))?;
let request = fixture.request(delete("one"))?;
let before = fixture.at()?;
let result = fixture.storage.mutate(&request);
assert!(result.is_err(), "underflow was accepted");
assert_eq!(
result.err().ok_or("underflow error missing")?.code,
"count_underflow"
);
assert_eq!(fixture.at()?.version, before.version);
assert!(
reads::get_bytes(
&fixture.storage.reader,
&before,
&revision_key("records", "one"),
)?
.is_some()
);
assert!(
fixture
.storage
.receipt(&ReceiptQuery {
branch: "main".to_owned(),
request_id: request.request_id,
})?
.is_none()
);
Ok(())
}
#[test]
fn corrupt_count_or_ready_refuses_without_publishing_a_write() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
for value in [
None,
Some(b"01".to_vec()),
Some(b"-1".to_vec()),
Some(b"18446744073709551616".to_vec()),
] {
fixture.edit(document_count_key("records"), value)?;
let request = fixture.request(put("one")?)?;
let before = fixture.at()?;
let result = fixture.storage.mutate(&request);
assert!(result.is_err(), "corrupt count was accepted");
assert_eq!(
result.err().ok_or("count error missing")?.code,
"counts_corrupt"
);
assert_eq!(fixture.at()?.version, before.version);
}
fixture.edit(document_count_key("records"), Some(b"0".to_vec()))?;
fixture.edit(READY_KEY.to_vec(), Some(b"2".to_vec()))?;
let request = fixture.request(put("one")?)?;
let before = fixture.at()?;
let result = fixture.storage.mutate(&request);
assert!(result.is_err(), "unknown readiness format was accepted");
assert_eq!(
result.err().ok_or("ready error missing")?.code,
"counts_corrupt"
);
assert_eq!(fixture.at()?.version, before.version);
Ok(())
}
#[test]
fn served_document_writes_have_no_expiry_envelope() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
let at = fixture.at()?;
let change = put("one")?;
let writes = fixture.storage.prepare_change(&at, &change, at.version)?;
let body = writes
.iter()
.find(|(key, _)| key == &document_key("records", "one"))
.and_then(|(_, value)| value.as_ref())
.ok_or("prepared body missing")?;
assert_eq!(body.as_slice(), b"{\"value\":1}");
assert!(haematite::ttl::Entry::decode(body)?.is_none());
assert!(haematite::ttl::entry::Stamped::decode(body)?.is_none());
fixture.apply(change)?;
assert_eq!(
reads::get_bytes(
&fixture.storage.reader,
&fixture.at()?,
&document_key("records", "one")
)?,
Some(body.clone()),
);
Ok(())
}
fn embedded_declaration_bytes() -> Result<Vec<u8>, serde_json::Error> {
serde_json::to_vec(&serde_json::json!({
"kind":"dense", "source":"body", "model": {
"name":"local-model", "revision":"v1", "manifest_sha256":"a".repeat(64)
}, "dimensions":2, "distance":"cosine"
}))
}
#[test]
fn counts_audit_and_rebuild_do_not_invent_an_embedded_vector_bucket() -> TestResult {
let mut fixture = populated()?;
fixture.edit(
b"w\0records\0embedding".to_vec(),
Some(embedded_declaration_bytes()?),
)?;
let assert_no_embedded = |answer: serde_json::Value| -> TestResult {
assert_eq!(answer["result"], "compared");
assert!(
answer["buckets"]
.as_array()
.ok_or("no count buckets")?
.iter()
.all(|row| row["kind"] != "vectors")
);
Ok(())
};
assert_no_embedded(audit(&fixture)?)?;
fixture
.storage
.rebuild_counts(&rebuild_request("declared-rebuild"))?;
assert_no_embedded(audit(&fixture)?)?;
assert!(reads::get_bytes(&fixture.storage.reader, &fixture.at()?, VECTOR_COUNT)?.is_none());
fixture.collection()?;
fixture.apply(put("later")?)?;
assert_no_embedded(audit(&fixture)?)?;
Ok(())
}
#[test]
fn candidate_count_seam_keeps_declared_rows_without_a_vector_bucket() -> TestResult {
let mut fixture = populated()?;
let at = fixture.at()?;
let field = b"w\0records\0embedding".to_vec();
let bytes = embedded_declaration_bytes()?;
let mut rows = std::collections::BTreeMap::from([(field.clone(), Some(bytes.clone()))]);
super::CandidateCounts::new(&fixture.storage.reader, &at)?.append(
&fixture.storage.reader,
&at,
&mut rows,
)?;
assert_eq!(
rows,
std::collections::BTreeMap::from([(field, Some(bytes))])
);
fixture.edit_batch(rows.into_iter().collect())?;
let at = fixture.at()?;
let mut repeated = std::collections::BTreeMap::from([(
b"w\0records\0embedding".to_vec(),
Some(embedded_declaration_bytes()?),
)]);
super::CandidateCounts::new(&fixture.storage.reader, &at)?.append(
&fixture.storage.reader,
&at,
&mut repeated,
)?;
assert!(!repeated.contains_key(VECTOR_COUNT));
fixture.collection()?;
fixture.apply(put("later")?)?;
assert!(reads::get_bytes(&fixture.storage.reader, &fixture.at()?, VECTOR_COUNT)?.is_none());
Ok(())
}
#[test]
fn candidate_traversal_carries_declarations_without_vector_counts() -> TestResult {
use crate::serve::protocol::Fork;
use crate::serve::storage::MergePin;
let mut fixture = populated()?;
let from = fixture.at()?.version;
fixture.apply(Change::Fork(Fork {
name: "feature".to_owned(),
from,
}))?;
let selection = Selection {
branch: "feature".to_owned(),
version: None,
};
let expected = fixture.storage.capture(&selection)?.version;
let change = serde_json::from_value(serde_json::json!({"kind":"vector.declare", "value": {
"collection":"records", "field":"embedding", "declaration":
serde_json::from_slice::<serde_json::Value>(&embedded_declaration_bytes()?)?
}}))?;
fixture.storage.mutate(&Mutation {
request_id: "declare-feature".to_owned(),
branch: "feature".to_owned(),
expected,
change,
})?;
let source = MergePin {
branch: "feature".to_owned(),
version: fixture.storage.capture(&selection)?.version,
};
let source_at = fixture.storage.capture(&Selection {
branch: source.branch.clone(),
version: Some(source.version),
})?;
let declared_bytes = reads::get_bytes(
&fixture.storage.reader,
&source_at,
b"w\0records\0embedding",
)?
.ok_or("source declaration absent")?;
let target = MergePin {
branch: "main".to_owned(),
version: fixture.at()?.version,
};
fixture
.storage
.merge_open("declared-merge", &source, &target)?;
for _ in 0..16 {
let page = fixture.storage.merge_candidate_page("declared-merge")?;
if page.end {
let name =
crate::serve::storage::merge_holders::holder_name("declared-merge", "candidate");
let record = fixture
.storage
.refs
.get(&name)
.ok_or("candidate reference absent")?;
let at = Captured {
branch: name,
version: crate::serve::storage::version(record),
shards: record.shards.clone(),
};
assert!(reads::get_bytes(&fixture.storage.reader, &at, VECTOR_COUNT)?.is_none());
let bytes = reads::get_bytes(&fixture.storage.reader, &at, b"w\0records\0embedding")?
.ok_or("candidate declaration absent")?;
assert_eq!(bytes, declared_bytes);
return Ok(());
}
}
Err("candidate failed to complete in bounded fixture pages".into())
}
#[test]
fn document_removal_advances_past_a_page_of_only_declared_fields() -> TestResult {
let mut fixture = populated()?;
let bytes = embedded_declaration_bytes()?;
let mut writes = (0..65)
.map(|index| {
(
format!("w\0records\0embedded-{index:03}").into_bytes(),
Some(bytes.clone()),
)
})
.collect::<Vec<_>>();
writes.push((b"w\0records\0z-supplied".to_vec(), Some(b"2".to_vec())));
fixture.edit_batch(writes)?;
let removals = crate::serve::storage::vectors::removals_for_document(
&fixture.storage.reader,
&fixture.at()?,
"records",
"record-0",
)?;
assert_eq!(
removals,
vec![(b"v\0records\0z-supplied\0record-0".to_vec(), None)]
);
Ok(())
}
fn append_prune_change(id: &str, keep: u32) -> Result<Change, Box<dyn Error>> {
Ok(serde_json::from_value(serde_json::json!({
"kind": "document.append_prune", "value": {
"collection": "records", "scope": ["person", "a"],
"suffix": id.strip_prefix("2/person/a/").unwrap_or(id),
"document": {"query": "records"}, "keep": keep,
}
}))?)
}
fn seed_append_prune(fixture: &mut Fixture, count: usize) -> TestResult {
let documents = (0..count)
.map(|index| {
Ok(ImportedDocument {
document_id: format!("2/person/a/{index:06}"),
document: RawValue::from_string("{}".to_owned())?,
})
})
.collect::<Result<Vec<_>, Box<dyn Error>>>()?;
fixture.apply(Change::PutMany(DocumentImport {
collection: "records".to_owned(),
documents,
}))
}
fn append_prune(
fixture: &mut Fixture,
id: &str,
keep: u32,
) -> Result<serde_json::Value, Box<dyn Error>> {
let request = fixture.request(append_prune_change(id, keep)?)?;
Ok(serde_json::to_value(fixture.storage.mutate_bounded(
&request,
NonZeroU32::MIN.saturating_add(63),
)?)?)
}
fn has_retained(fixture: &Fixture, id: &str) -> Result<bool, Box<dyn Error>> {
Ok(reads::get_bytes(
&fixture.storage.reader,
&fixture.at()?,
&document_key("records", id),
)?
.is_some())
}
#[test]
fn append_prune_under_bound_retains_the_new_record() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
let before = fixture.at()?.version;
let receipt = append_prune(&mut fixture, "2/person/a/000000", 1_000)?;
assert_eq!(
receipt["retention"],
serde_json::json!({"document_id": "2/person/a/000000", "removed": 0, "more": false})
);
assert!(has_retained(&fixture, "2/person/a/000000")?);
assert_eq!(fixture.count()?, 1);
assert_eq!(fixture.at()?.version.sequence, before.sequence + 1);
Ok(())
}
#[test]
fn append_prune_1001st_record_prunes_oldest_in_the_same_version() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
seed_append_prune(&mut fixture, 1_000)?;
let before = fixture.at()?;
let receipt = append_prune(&mut fixture, "2/person/a/001000", 1_000)?;
assert_eq!(
receipt["retention"],
serde_json::json!({"document_id": "2/person/a/001000", "removed": 1, "more": false})
);
assert_eq!(fixture.at()?.version.sequence, before.version.sequence + 1);
assert_eq!(fixture.count()?, 1_000);
assert!(!has_retained(&fixture, "2/person/a/000000")?);
assert!(has_retained(&fixture, "2/person/a/000001")?);
assert!(has_retained(&fixture, "2/person/a/001000")?);
assert!(
reads::get_bytes(
&fixture.storage.reader,
&before,
&document_key("records", "2/person/a/000000")
)?
.is_some()
);
let fixture = fixture.reopen()?;
assert!(!has_retained(&fixture, "2/person/a/000000")?);
assert!(has_retained(&fixture, "2/person/a/001000")?);
assert!(
audit(&fixture)?["buckets"]
.as_array()
.ok_or("no buckets")?
.iter()
.all(|row| row["result"] == "agree")
);
Ok(())
}
#[test]
fn append_prune_leaves_other_scopes_unchanged() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
fixture.apply(put("2/person/a/000000")?)?;
fixture.apply(put("2/person/b/000000")?)?;
let receipt = append_prune(&mut fixture, "2/person/a/000001", 1)?;
assert_eq!(receipt["retention"]["removed"], 1);
assert!(has_retained(&fixture, "2/person/b/000000")?);
assert_eq!(fixture.count()?, 2);
Ok(())
}
#[test]
fn append_prune_caps_removals_and_reports_remaining_excess() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
seed_append_prune(&mut fixture, 100)?;
let first = append_prune(&mut fixture, "2/person/a/000100", 1)?;
assert_eq!(
first["retention"],
serde_json::json!({"document_id": "2/person/a/000100", "removed": 64, "more": true})
);
assert_eq!(fixture.count()?, 37);
let second = append_prune(&mut fixture, "2/person/a/000101", 1)?;
assert_eq!(
second["retention"],
serde_json::json!({"document_id": "2/person/a/000101", "removed": 37, "more": false})
);
assert_eq!(fixture.count()?, 1);
assert!(has_retained(&fixture, "2/person/a/000101")?);
assert!(!has_retained(&fixture, "2/person/a/000100")?);
Ok(())
}
#[test]
fn append_prune_replay_returns_the_stored_result_without_another_commit() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
seed_append_prune(&mut fixture, 2)?;
let request = fixture.request(append_prune_change("2/person/a/000002", 1)?)?;
let first = fixture.storage.mutate_bounded(&request, NonZeroU32::MIN)?;
let installed = fixture.at()?.version;
let replay = fixture.storage.mutate_bounded(&request, NonZeroU32::MIN)?;
assert_eq!(serde_json::to_value(first)?, serde_json::to_value(replay)?);
assert_eq!(fixture.at()?.version, installed);
assert_eq!(fixture.count()?, 1);
assert!(has_retained(&fixture, "2/person/a/000002")?);
Ok(())
}
#[test]
fn append_prune_refuses_invalid_requests_without_advancing() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
fixture.apply(put("2/person/a/000001")?)?;
let before = fixture.at()?.version;
for (id, keep, code) in [
("2/person/b/000002", 1, "append_prune_suffix_invalid"),
("2/person/a/000001", 1, "append_prune_key_exists"),
("2/person/a/000002", 0, "append_prune_keep_zero"),
("2/person/a/000002", 4_097, "append_prune_keep_limit"),
("2/person/a/000000", 1, "append_prune_key_would_be_removed"),
] {
let request = fixture.request(append_prune_change(id, keep)?)?;
let error = fixture
.storage
.mutate_bounded(&request, NonZeroU32::MIN)
.err()
.ok_or("invalid append was accepted")?;
assert_eq!(error.code, code);
assert_eq!(fixture.at()?.version, before);
assert_eq!(fixture.count()?, 1);
}
Ok(())
}
#[test]
fn append_prune_refuses_incident_edges_in_either_direction() -> TestResult {
for outgoing in [true, false] {
let mut fixture = Fixture::new()?;
fixture.collection()?;
fixture.apply(put("2/person/a/000000")?)?;
fixture.apply(put("other")?)?;
let (from, to, code) = if outgoing {
("2/person/a/000000", "other", "append_prune_outgoing_edge")
} else {
("other", "2/person/a/000000", "append_prune_incoming_edge")
};
fixture.apply(Change::EdgePut(edge(from, to)))?;
let before = fixture.at()?.version;
let request = fixture.request(append_prune_change("2/person/a/000001", 1)?)?;
assert_eq!(
fixture
.storage
.mutate_bounded(&request, NonZeroU32::MIN)
.err()
.ok_or("attached record was pruned")?
.code,
code
);
assert_eq!(fixture.at()?.version, before);
assert!(has_retained(&fixture, "2/person/a/000000")?);
assert!(!has_retained(&fixture, "2/person/a/000001")?);
}
Ok(())
}
#[test]
fn append_prune_refuses_supplied_and_declared_fields() -> TestResult {
for embedded in [false, true] {
let mut fixture = Fixture::new()?;
fixture.collection()?;
fixture.apply(put("2/person/a/000000")?)?;
if embedded {
fixture.edit(
b"w\0records\0embedding".to_vec(),
Some(embedded_declaration_bytes()?),
)?;
} else {
fixture.apply(vector("2/person/a/000000"))?;
}
let before = fixture.at()?.version;
let request = fixture.request(append_prune_change("2/person/a/000001", 1)?)?;
assert_eq!(
fixture
.storage
.mutate_bounded(&request, NonZeroU32::MIN)
.err()
.ok_or("field-bearing collection was pruned")?
.code,
"append_prune_vector_field"
);
assert_eq!(fixture.at()?.version, before);
assert!(has_retained(&fixture, "2/person/a/000000")?);
assert!(!has_retained(&fixture, "2/person/a/000001")?);
}
Ok(())
}
#[test]
fn append_prune_reads_an_old_receipt_after_reopening_the_store() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
let query = ReceiptQuery {
branch: "main".to_owned(),
request_id: "request-1".to_owned(),
};
let before = fixture.storage.receipt(&query)?.ok_or("no receipt")?;
let old = serde_json::to_value(&before)?;
assert!(old.get("retention").is_none());
fixture.edit_batch(vec![(
crate::serve::storage::receipt_key("main", before.version.generation, "request-1"),
Some(serde_json::to_vec(&old)?),
)])?;
let mut fixture = fixture.reopen()?;
assert_eq!(
serde_json::to_value(fixture.storage.receipt(&query)?.ok_or("no old receipt")?)?,
old
);
Ok(())
}
#[cfg(unix)]
fn append_prune_config() -> serde_json::Value {
serde_json::json!({
"data_dir": "/unused/database", "socket": "/unused/socket",
"branch_cache_bytes_per_shard": 1, "max_frame": 4_096,
"max_clients": 1, "max_page": 1, "timeout_ms": 1,
})
}
#[cfg(unix)]
#[test]
fn append_prune_config_defaults_are_independent_of_max_page() -> TestResult {
let config: crate::serve::cli::Config = serde_json::from_value(append_prune_config())?;
let debug = format!("{config:?}");
assert!(debug.contains("append_prune_keep_ceiling: 4096"));
assert!(debug.contains("append_prune_removal_cap: 64"));
let mut changed = append_prune_config();
changed["append_prune_keep_ceiling"] = 2.into();
changed["append_prune_removal_cap"] = 2.into();
let config: crate::serve::cli::Config = serde_json::from_value(changed)?;
let debug = format!("{config:?}");
assert!(debug.contains("append_prune_keep_ceiling: 2"));
assert!(debug.contains("append_prune_removal_cap: 2"));
Ok(())
}
#[cfg(unix)]
#[test]
fn append_prune_config_zero_refuses_before_opening_storage() -> TestResult {
let directory = tempfile::tempdir()?;
let path = directory.path().join("serve.json");
for (name, code) in [
(
"append_prune_keep_ceiling",
"append_prune_keep_ceiling_zero",
),
("append_prune_removal_cap", "append_prune_removal_cap_zero"),
] {
let mut config = append_prune_config();
config[name] = 0.into();
std::fs::write(&path, serde_json::to_vec(&config)?)?;
let error = crate::serve::cli::run(
[
"--config".to_owned(),
path.to_str().ok_or("invalid path")?.to_owned(),
]
.into_iter(),
)
.err()
.ok_or("zero configuration was accepted")?;
assert!(error.starts_with(code), "{error}");
}
Ok(())
}
#[test]
fn append_prune_lost_answer_preserves_retention_for_receipt_reconciliation() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
seed_append_prune(&mut fixture, 2)?;
let request = fixture.request(append_prune_change("2/person/a/000002", 1)?)?;
Storage::lose_next_mutation_answer();
assert_eq!(
fixture
.storage
.mutate_bounded(&request, NonZeroU32::MIN)
.err()
.ok_or("answer was not lost")?
.code,
"durability_unknown"
);
let mut fixture = fixture.reopen()?;
let receipt = fixture
.storage
.receipt(&ReceiptQuery {
branch: "main".to_owned(),
request_id: request.request_id.clone(),
})?
.ok_or("committed receipt missing")?;
assert_eq!(
serde_json::to_value(&receipt)?["retention"],
serde_json::json!({"document_id": "2/person/a/000002", "removed": 2, "more": false})
);
let installed = fixture.at()?.version;
let replay = fixture.storage.mutate_bounded(&request, NonZeroU32::MIN)?;
assert_eq!(
serde_json::to_value(receipt)?,
serde_json::to_value(replay)?
);
assert_eq!(fixture.at()?.version, installed);
Ok(())
}
#[test]
fn append_prune_served_refusal_reports_not_applied() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
let before = fixture.at()?.version;
let request = fixture.request(append_prune_change("2/person/b/000001", 1)?)?;
let service = crate::serve::service::Service::new(fixture.storage, NonZeroU32::MIN)?;
let error = service
.dispatch(crate::serve::protocol::Operation::Mutate(request))
.err()
.ok_or("out-of-scope mutation was accepted")?;
let failure = serde_json::to_value(error)?;
assert_eq!(failure["code"], "append_prune_suffix_invalid");
assert_eq!(failure["outcome"], "not_applied");
let answer = service.dispatch(crate::serve::protocol::Operation::Receipt(ReceiptQuery {
branch: "main".to_owned(),
request_id: "request-2".to_owned(),
}))?;
assert!(answer.is_null());
let read = service.dispatch(crate::serve::protocol::Operation::Get(
crate::serve::protocol::DocumentRead {
selection: Selection {
branch: "main".to_owned(),
version: Some(before),
},
collection: "records".to_owned(),
document_id: "2/person/b/000001".to_owned(),
},
))?;
assert!(read["document"].is_null());
Ok(())
}
fn scoped_append_prune(scope: &[&str], suffix: &str, keep: u32) -> Result<Change, Box<dyn Error>> {
Ok(serde_json::from_value(serde_json::json!({
"kind": "document.append_prune", "value": {
"collection": "records", "scope": scope, "suffix": suffix,
"document": {"query": "records"}, "keep": keep,
}
}))?)
}
fn assert_scopes_separate(other: &[&str], other_prefix: &str) -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
let mut ids = Vec::new();
for prefix in ["1/a/", other_prefix] {
for index in 0..3 {
ids.push(format!("{prefix}{index:06}"));
}
}
let documents = ids
.iter()
.map(|id| {
Ok(ImportedDocument {
document_id: id.clone(),
document: RawValue::from_string("{}".to_owned())?,
})
})
.collect::<Result<Vec<_>, Box<dyn Error>>>()?;
fixture.apply(Change::PutMany(DocumentImport {
collection: "records".to_owned(),
documents,
}))?;
let before = fixture.at()?;
let request = fixture.request(scoped_append_prune(&["a"], "000003", 1)?)?;
let receipt = fixture.storage.mutate_bounded(&request, NonZeroU32::MIN)?;
assert_eq!(
serde_json::to_value(receipt)?["retention"],
serde_json::json!({
"document_id": "1/a/000003", "removed": 3, "more": false,
})
);
let after = fixture.at()?;
assert_eq!(after.version.sequence, before.version.sequence + 1);
assert_eq!(fixture.count()?, 4);
for (prefix, expected) in [("1/a/", 1), (other_prefix, 3)] {
assert_eq!(
reads::prefix_keys(
&fixture.storage.reader,
&after,
&document_key("records", prefix),
std::num::NonZeroUsize::MIN.saturating_add(7)
)?
.len(),
expected
);
}
for id in ids.iter().filter(|id| id.starts_with(other_prefix)) {
let key = document_key("records", id);
assert_eq!(
reads::get_bytes(&fixture.storage.reader, &before, &key)?,
reads::get_bytes(&fixture.storage.reader, &after, &key)?
);
}
assert!(has_retained(&fixture, "1/a/000003")?);
let other_request = fixture.request(scoped_append_prune(other, "000003", 1)?)?;
let receipt = fixture
.storage
.mutate_bounded(&other_request, NonZeroU32::MIN)?;
assert_eq!(
serde_json::to_value(receipt)?["retention"]["document_id"],
format!("{other_prefix}000003")
);
assert_eq!(fixture.count()?, 2);
assert!(has_retained(&fixture, "1/a/000003")?);
Ok(())
}
#[test]
fn append_prune_component_scopes_keep_a_and_ab_separate() -> TestResult {
assert_scopes_separate(&["ab"], "1/ab/")
}
#[test]
fn append_prune_component_count_keeps_parent_and_child_scopes_separate() -> TestResult {
assert_scopes_separate(&["a", "b"], "2/a/b/")
}
fn refuse_scoped_append(scope: &[&str], suffix: &str, code: &str) -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
let before = fixture.at()?.version;
let request = fixture.request(scoped_append_prune(scope, suffix, 1)?)?;
let error = fixture
.storage
.mutate_bounded(&request, NonZeroU32::MIN)
.err()
.ok_or("invalid scoped append was accepted")?;
assert_eq!(error.code, code);
assert_eq!(fixture.at()?.version, before);
assert_eq!(fixture.count()?, 0);
assert!(
fixture
.storage
.receipt(&ReceiptQuery {
branch: "main".to_owned(),
request_id: request.request_id,
})?
.is_none()
);
Ok(())
}
#[test]
fn append_prune_separator_in_a_component_cannot_alias_two_components() -> TestResult {
refuse_scoped_append(&["a/b"], "000000", "append_prune_scope_component_invalid")
}
#[test]
fn append_prune_scope_components_and_suffix_refuse_empty_separator_and_nul() -> TestResult {
for component in ["", "a/b", "a\0b"] {
refuse_scoped_append(
&[component],
"000000",
"append_prune_scope_component_invalid",
)?;
}
for suffix in ["", "a/b", "a\0b"] {
refuse_scoped_append(&["a"], suffix, "append_prune_suffix_invalid")?;
}
Ok(())
}
#[test]
fn append_prune_scope_refuses_zero_or_nine_components() -> TestResult {
refuse_scoped_append(&[], "000000", "append_prune_scope_components_invalid")?;
refuse_scoped_append(&["a"; 9], "000000", "append_prune_scope_components_invalid")
}
#[test]
fn append_prune_formed_id_enforces_the_whole_512_byte_boundary() -> TestResult {
refuse_scoped_append(&["a"], &"z".repeat(509), "append_prune_document_id_invalid")?;
let mut fixture = Fixture::new()?;
fixture.collection()?;
let suffix = "z".repeat(508);
let request = fixture.request(scoped_append_prune(&["a"], &suffix, 1)?)?;
let receipt = fixture.storage.mutate_bounded(&request, NonZeroU32::MIN)?;
let answer = serde_json::to_value(receipt)?;
let id = answer["retention"]["document_id"]
.as_str()
.ok_or("formed id missing")?;
assert_eq!(id.len(), 512);
assert!(has_retained(&fixture, id)?);
Ok(())
}
#[cfg(unix)]
#[test]
fn append_prune_cap_one_refuses_before_opening_storage() -> TestResult {
let directory = tempfile::tempdir()?;
let path = directory.path().join("serve.json");
let mut config = append_prune_config();
config["append_prune_removal_cap"] = 1.into();
std::fs::write(&path, serde_json::to_vec(&config)?)?;
let error = crate::serve::cli::run(
[
"--config".to_owned(),
path.to_str().ok_or("invalid path")?.to_owned(),
]
.into_iter(),
)
.err()
.ok_or("non-progressing cap was accepted")?;
assert!(
error.starts_with("append_prune_removal_cap_too_small"),
"{error}"
);
Ok(())
}
#[test]
fn vector_declaration_attribution_round_trips_collection_and_field_only() -> TestResult {
let origin = crate::serve::protocol::Version {
generation: 7,
sequence: 1,
};
let installed = crate::serve::protocol::Version {
generation: 7,
sequence: 2,
};
let declaration: serde_json::Value = serde_json::from_slice(&embedded_declaration_bytes()?)?;
let request = crate::serve::contracts::MutationRequest::decode(&serde_json::to_vec(
&serde_json::json!({
"request_id": "declare-attribution", "branch": "main",
"caller": {"issuer": "lys", "subject": "person-a"},
"contract": {"kind": "dependencies", "origin": origin, "reads": []},
"change": {"kind": "vector.declare", "value": {
"collection": "records", "field": "embedding", "declaration": declaration,
}},
}),
)?)?
.resolve("socket-owner")?;
let row = crate::serve::contracts::attribution::Row::for_request(&request, installed)?;
let bytes = row.encode()?;
let decoded = crate::serve::contracts::attribution::Row::decode(&row.key()?, &bytes)?;
assert_eq!(decoded.version(), installed);
assert_eq!(decoded.caller().subject(), "person-a");
assert_eq!(serde_json::to_value(&row)?, serde_json::to_value(&decoded)?);
assert_eq!(
serde_json::to_value(&decoded)?["change"],
serde_json::json!({
"kind": "vector.declare", "collection": "records", "field": "embedding",
})
);
assert!(!String::from_utf8(bytes)?.contains("manifest_sha256"));
Ok(())
}
#[test]
fn vector_declaration_attribution_refuses_invalid_names_in_requests_and_rows() -> TestResult {
let installed = crate::serve::protocol::Version {
generation: 7,
sequence: 2,
};
let declaration: serde_json::Value = serde_json::from_slice(&embedded_declaration_bytes()?)?;
let valid = serde_json::json!({
"request_id": "declare-attribution", "branch": "main",
"caller": {"issuer": "lys", "subject": "person-a"},
"contract": {"kind": "dependencies", "origin": {
"generation": "7", "sequence": "1"
}, "reads": []},
"change": {"kind": "vector.declare", "value": {
"collection": "records", "field": "embedding", "declaration": declaration,
}},
});
let request = crate::serve::contracts::MutationRequest::decode(&serde_json::to_vec(&valid)?)?
.resolve("socket-owner")?;
let row = crate::serve::contracts::attribution::Row::for_request(&request, installed)?;
for name in ["collection", "field"] {
let mut invalid = valid.clone();
invalid["change"]["value"][name] = "".into();
let request =
crate::serve::contracts::MutationRequest::decode(&serde_json::to_vec(&invalid)?)?
.resolve("socket-owner")?;
assert_eq!(
crate::serve::contracts::attribution::Row::for_request(&request, installed)
.err()
.ok_or("invalid declaration attribution was accepted")?
.code,
"invalid_identifier"
);
let mut invalid_row = serde_json::to_value(&row)?;
invalid_row["change"][name] = "".into();
assert_eq!(
crate::serve::contracts::attribution::Row::decode(
&row.key()?,
&serde_json::to_vec(&invalid_row)?
)
.err()
.ok_or("invalid stored declaration attribution was accepted")?
.code,
"change_record_corrupt"
);
}
Ok(())
}
#[test]
fn append_prune_attribution_records_components_and_suffix_without_document_payload() -> TestResult {
let origin = crate::serve::protocol::Version {
generation: 7,
sequence: 1,
};
let installed = crate::serve::protocol::Version {
generation: 7,
sequence: 2,
};
let request = crate::serve::contracts::MutationRequest::decode(&serde_json::to_vec(
&serde_json::json!({
"request_id": "retain-attribution", "branch": "main",
"caller": {"issuer": "lys", "subject": "person-a"},
"contract": {"kind": "dependencies", "origin": origin, "reads": []},
"change": {"kind": "document.append_prune", "value": {
"collection": "records", "scope": ["a", "b"], "suffix": "x",
"document": {"private_payload": "discarded from attribution"}, "keep": 1,
}},
}),
)?)?
.resolve("socket-owner")?;
let row = crate::serve::contracts::attribution::Row::for_request(&request, installed)?;
let bytes = row.encode()?;
let decoded = crate::serve::contracts::attribution::Row::decode(&row.key()?, &bytes)?;
assert_eq!(
serde_json::to_value(&decoded)?["change"],
serde_json::json!({
"kind": "document.append_prune", "collection": "records", "scope": ["a", "b"],
"suffix": "x", "keep": 1,
})
);
assert!(!String::from_utf8(bytes)?.contains("private_payload"));
let mut invalid = serde_json::to_value(&row)?;
invalid["change"]["scope"] = serde_json::json!(["a/b"]);
assert_eq!(
crate::serve::contracts::attribution::Row::decode(
&row.key()?,
&serde_json::to_vec(&invalid)?
)
.err()
.ok_or("invalid stored scope was accepted")?
.code,
"change_record_corrupt"
);
Ok(())
}
#[test]
fn append_prune_minimum_cap_two_reduces_backlog_and_finishes() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.collection()?;
seed_append_prune(&mut fixture, 3)?;
let limits = crate::serve::storage::AppendPruneLimits::new(1, 2)?;
let request = fixture.request(append_prune_change("2/person/a/000003", 1)?)?;
let first =
fixture
.storage
.mutate_with_append_prune_limits(&request, NonZeroU32::MIN, limits)?;
assert_eq!(
serde_json::to_value(first)?["retention"],
serde_json::json!({
"document_id": "2/person/a/000003", "removed": 2, "more": true,
})
);
assert_eq!(fixture.count()?, 2);
let request = fixture.request(append_prune_change("2/person/a/000004", 1)?)?;
let second =
fixture
.storage
.mutate_with_append_prune_limits(&request, NonZeroU32::MIN, limits)?;
assert_eq!(
serde_json::to_value(second)?["retention"],
serde_json::json!({
"document_id": "2/person/a/000004", "removed": 2, "more": false,
})
);
assert_eq!(fixture.count()?, 1);
assert!(has_retained(&fixture, "2/person/a/000004")?);
Ok(())
}
fn embedded_change() -> Result<Change, Box<dyn Error>> {
Ok(serde_json::from_value(serde_json::json!({
"kind": "vector.declare",
"value": {
"collection": "records",
"field": "embedding",
"declaration": serde_json::from_slice::<serde_json::Value>(&embedded_declaration_bytes()?)?
}
}))?)
}
fn stats_operation() -> crate::serve::protocol::Operation {
crate::serve::protocol::Operation::StoreStats(crate::serve::protocol::StoreRead {
selection: Selection {
branch: "main".to_owned(),
version: None,
},
})
}
fn into_stats_service(
fixture: Fixture,
) -> Result<(tempfile::TempDir, crate::serve::service::Service), Box<dyn Error>> {
let Fixture {
storage,
directory,
request: _,
} = fixture;
let service = crate::serve::service::Service::new(storage, NonZeroU32::MIN)?;
Ok((directory, service))
}
fn assert_companion_field(answer: &serde_json::Value) {
assert_eq!(
answer["embedded_fields"],
serde_json::json!([{
"collection": "records", "field": "embedding", "kind": "dense",
"storage": "companion", "count": null, "count_reason": "companion_count_not_kept"
}])
);
assert!(answer["embedded_fields_reason"].is_null());
}
#[test]
fn normal_service_declare_keeps_embedded_stats_visible_after_rebuild_and_reopen() -> TestResult {
let fixture = populated()?;
let expected = fixture.at()?.version;
let (directory, service) = into_stats_service(fixture)?;
service.dispatch(crate::serve::protocol::Operation::Mutate(Mutation {
request_id: "declare-count-stats".to_owned(),
branch: "main".to_owned(),
expected,
change: embedded_change()?,
}))?;
let answer = service.dispatch(stats_operation())?;
assert_companion_field(&answer);
assert_eq!(answer["documents"], 2);
assert_eq!(answer["vectors"], 0);
service.dispatch(crate::serve::protocol::Operation::CountsRebuild(
rebuild_request("embedded-stats-rebuild"),
))?;
assert_companion_field(&service.dispatch(stats_operation())?);
drop(service);
let storage = Storage::open(
&directory.path().join("database"),
NodeCacheBudget::Unlimited,
fixture_long_wait_interval(),
)?;
let at = storage.capture(&Selection {
branch: "main".to_owned(),
version: None,
})?;
assert!(reads::get_bytes(&storage.reader, &at, VECTOR_COUNT)?.is_none());
let service = crate::serve::service::Service::new(storage, NonZeroU32::MIN)?;
assert_companion_field(&service.dispatch(stats_operation())?);
Ok(())
}
#[test]
fn stats_still_refuses_a_supplied_field_with_no_count() -> TestResult {
let mut fixture = populated()?;
fixture.apply(vector("record-0"))?;
fixture.apply(vector_delete("record-0"))?;
fixture.edit(VECTOR_COUNT.to_vec(), None)?;
let (directory, service) = into_stats_service(fixture)?;
let error = service
.dispatch(stats_operation())
.err()
.ok_or("missing supplied count accepted")?;
assert_eq!(error.code, "counts_corrupt");
assert_eq!(error.message, "declared field has no count");
drop(service);
drop(directory);
Ok(())
}
#[test]
fn embedded_stats_metadata_is_listed_when_main_counts_are_not_ready() -> TestResult {
let mut fixture = populated()?;
fixture.apply(embedded_change()?)?;
fixture.edit(READY_KEY.to_vec(), None)?;
let (directory, service) = into_stats_service(fixture)?;
let answer = service.dispatch(stats_operation())?;
assert_companion_field(&answer);
assert!(answer["vectors"].is_null());
assert_eq!(answer["vectors_reason"], "counts_not_ready");
drop(service);
drop(directory);
Ok(())
}
#[test]
fn embedded_stats_metadata_is_withheld_without_store_authority() -> TestResult {
let mut fixture = populated()?;
fixture.apply(embedded_change()?)?;
let Fixture {
storage,
directory,
request: _,
} = fixture;
let service = crate::serve::service::Service::with_access_source(
storage,
NonZeroU32::MIN,
None,
std::sync::Arc::new(crate::serve::access::TestSource::denying(&[(
"records", "record-0",
)])),
)?;
let answer = service.dispatch(stats_operation())?;
assert!(answer["embedded_fields"].is_null());
assert_eq!(answer["embedded_fields_reason"], "withheld");
assert!(answer["vectors"].is_null());
assert_eq!(answer["vectors_reason"], "withheld");
drop(service);
drop(directory);
Ok(())
}
const fn fixture_long_wait_interval() -> std::time::Duration {
std::time::Duration::from_secs(1)
}