#![forbid(unsafe_code)]
use std::error::Error;
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use haematite_cli::serve::contracts::{MutationRequest, attribution::Row};
use haematite_cli::serve::protocol::{
BranchMerge, Change, CollectionWrite, DocumentWrite, Fork, Mutation, Operation, Selection,
VectorWrite, Version,
};
use haematite_cli::serve::storage::{Captured, Failure, Receipt, Storage, reads};
use serde_json::{Value, json, value::RawValue};
type TestResult = Result<(), Box<dyn Error>>;
struct Fixture {
storage: Storage,
directory: tempfile::TempDir,
requests: 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,
std::time::Duration::from_secs(30),
)?;
let mut fixture = Self {
storage,
directory,
requests: 0,
};
fixture.apply(
"main",
Change::Collection(CollectionWrite {
collection: "notes".to_owned(),
schema: None,
}),
)?;
Ok(fixture)
}
fn capture(&self, branch: &str, version: Option<Version>) -> Result<Captured, Failure> {
self.storage.capture(&Selection {
branch: branch.to_owned(),
version,
})
}
fn request(&mut self, branch: &str, change: Change) -> Result<Mutation, Failure> {
self.requests += 1;
Ok(Mutation {
request_id: format!("request-{}", self.requests),
branch: branch.to_owned(),
expected: self.capture(branch, None)?.version,
change,
})
}
fn apply(&mut self, branch: &str, change: Change) -> Result<Receipt, Box<dyn Error>> {
let request = self.request(branch, change)?;
Ok(self.storage.mutate(&request)?)
}
fn declaration(
&self,
branch: &str,
version: Option<Version>,
) -> Result<Option<Value>, Box<dyn Error>> {
let at = self.capture(branch, version)?;
reads::get_bytes(&self.storage.reader()?, &at, b"f\0eq\0notes\0ward")?
.map(|bytes| serde_json::from_slice(&bytes).map_err(Into::into))
.transpose()
}
fn refuse(&mut self, change: Change, code: &str) -> TestResult {
let before = self.capture("main", None)?.version;
let request = self.request("main", change)?;
let refusal = self
.storage
.mutate(&request)
.err()
.ok_or("accepted invalid declaration")?;
assert_eq!(refusal.code, code);
assert_eq!(self.capture("main", None)?.version, before);
Ok(())
}
}
fn declaration(field: &str) -> Result<Change, serde_json::Error> {
serde_json::from_value(json!({
"kind":"index.declare", "value":{"collection":"notes", "field":field}
}))
}
#[test]
fn index_declare_protocol_round_trips_and_rejects_unknown_members() -> TestResult {
let wire = json!({"kind":"index.declare", "value":{"collection":"notes", "field":"ward"}});
let change: Change = serde_json::from_value(wire.clone())?;
assert_eq!(serde_json::to_value(change)?, wire);
assert!(Change::WIRE_NAMES.contains(&"index.declare"));
let mut extra = wire.clone();
extra["value"]["dimensions"] = json!(3);
assert!(serde_json::from_value::<Change>(extra).is_err());
let mut missing = wire;
missing["value"]
.as_object_mut()
.ok_or("declaration is not an object")?
.remove("field");
assert!(serde_json::from_value::<Change>(missing).is_err());
Ok(())
}
#[test]
fn index_declare_records_versioned_change_and_receipt() -> TestResult {
let mut fixture = Fixture::new()?;
let before = fixture.capture("main", None)?.version;
let request = fixture.request("main", declaration("ward")?)?;
let receipt = fixture.storage.mutate(&request)?;
assert_eq!(receipt.version.sequence, before.sequence + 1);
assert!(fixture.declaration("main", Some(before))?.is_none());
let row = fixture
.declaration("main", Some(receipt.version))?
.ok_or("declaration event missing")?;
assert_eq!(row["format"], json!(1));
assert_eq!(row["declared_in"], json!("main"));
assert_eq!(row["declared_at"], serde_json::to_value(receipt.version)?);
assert_eq!(
row["change"],
json!({"kind":"index.declare", "value":{"collection":"notes", "field":"ward"}})
);
let history = fixture.storage.catalogue().history("main", None, 10)?;
let at = history
.iter()
.find(|at| at.version == receipt.version)
.ok_or("history omitted declaration version")?;
let bytes = reads::get_bytes(&fixture.storage.reader()?, at, b"f\0eq\0notes\0ward")?
.ok_or("history root lost declaration event")?;
assert_eq!(serde_json::from_slice::<Value>(&bytes)?, row);
let replay = fixture.storage.mutate(&request)?;
assert_eq!(
serde_json::to_value(replay)?,
serde_json::to_value(&receipt)?
);
assert_eq!(fixture.capture("main", None)?.version, receipt.version);
Ok(())
}
#[test]
fn index_declare_forks_at_before_and_after_versions() -> TestResult {
let mut fixture = Fixture::new()?;
let before = fixture.capture("main", None)?.version;
let declared = fixture.apply("main", declaration("ward")?)?.version;
let row = fixture
.declaration("main", None)?
.ok_or("declaration missing")?;
for (name, from, present) in [("before", before, false), ("after", declared, true)] {
fixture.apply(
"main",
Change::Fork(Fork {
name: name.to_owned(),
from,
}),
)?;
assert_eq!(
fixture.declaration(name, None)?,
present.then(|| row.clone())
);
}
fixture.refuse(declaration("ward")?, "index_already_declared")?;
assert_eq!(fixture.declaration("main", None)?, Some(row));
Ok(())
}
#[test]
fn index_declare_refuses_paths_duplicates_and_vectors_without_writes() -> TestResult {
let mut fixture = Fixture::new()?;
for field in ["patient.ward", "/ward", "ward[0]"] {
fixture.refuse(declaration(field)?, "index_field_not_top_level")?;
assert!(fixture.declaration("main", None)?.is_none());
}
fixture.apply(
"main",
Change::Put(DocumentWrite {
collection: "notes".to_owned(),
document_id: "n1".to_owned(),
document: RawValue::from_string(r#"{"ward":"4B"}"#.to_owned())?,
}),
)?;
fixture.apply(
"main",
Change::VectorPut(VectorWrite {
collection: "notes".to_owned(),
document_id: "n1".to_owned(),
field: "ward".to_owned(),
vector: vec![1.0, 0.0],
}),
)?;
fixture.refuse(declaration("ward")?, "index_vector_field")?;
assert!(fixture.declaration("main", None)?.is_none());
Ok(())
}
#[test]
fn index_declare_prevents_reciprocal_vector_field_and_missing_collection() -> TestResult {
let mut fixture = Fixture::new()?;
let missing = serde_json::from_value(json!({
"kind":"index.declare", "value":{"collection":"absent", "field":"ward"}
}))?;
fixture.refuse(missing, "collection_missing")?;
fixture.apply("main", declaration("ward")?)?;
fixture.apply(
"main",
Change::Put(DocumentWrite {
collection: "notes".to_owned(),
document_id: "n1".to_owned(),
document: RawValue::from_string("{}".to_owned())?,
}),
)?;
fixture.refuse(
Change::VectorPut(VectorWrite {
collection: "notes".to_owned(),
document_id: "n1".to_owned(),
field: "ward".to_owned(),
vector: vec![1.0, 0.0],
}),
"index_vector_field",
)?;
assert!(fixture.declaration("main", None)?.is_some());
Ok(())
}
#[test]
fn index_declare_reopens_without_changing_document_bytes() -> TestResult {
let mut fixture = Fixture::new()?;
let body = r#"{ "ward": "4B", "number": 9007199254740993 }"#;
let before = fixture
.apply(
"main",
Change::Put(DocumentWrite {
collection: "notes".to_owned(),
document_id: "n1".to_owned(),
document: RawValue::from_string(body.to_owned())?,
}),
)?
.version;
let declared = fixture.apply("main", declaration("ward")?)?.version;
let Fixture {
storage, directory, ..
} = fixture;
drop(storage);
let storage = Storage::open(
&directory.path().join("database"),
NodeCacheBudget::Unlimited,
std::time::Duration::from_secs(30),
)?;
let reader = storage.reader()?;
for version in [before, declared] {
let at = storage.capture(&Selection {
branch: "main".to_owned(),
version: Some(version),
})?;
assert_eq!(
reads::get_bytes(&reader, &at, b"d\0notes\0n1")?,
Some(body.as_bytes().to_vec())
);
assert_eq!(
reads::get_bytes(&reader, &at, b"f\0eq\0notes\0ward")?.is_some(),
version == declared
);
}
Ok(())
}
#[test]
fn index_declare_attribution_round_trips_names_only() -> TestResult {
let origin = Version {
generation: 7,
sequence: 1,
};
let installed = Version {
generation: 7,
sequence: 2,
};
let request = MutationRequest::decode(&serde_json::to_vec(&json!({
"request_id":"declaration", "branch":"main",
"caller":{"issuer":"lys", "subject":"person-a"},
"contract":{"kind":"dependencies", "origin":origin, "reads":[]},
"change":{"kind":"index.declare", "value":{"collection":"notes", "field":"ward"}}
}))?)?
.resolve("socket-owner")?;
let row = Row::for_request(&request, installed)?;
let decoded = Row::decode(&row.key()?, &row.encode()?)?;
assert_eq!(decoded.version(), installed);
assert_eq!(decoded.caller().subject(), "person-a");
assert_eq!(serde_json::to_value(&decoded)?, serde_json::to_value(&row)?);
assert_eq!(
serde_json::to_value(decoded)?["change"],
json!({"kind":"index.declare", "collection":"notes", "field":"ward"})
);
Ok(())
}
#[test]
fn index_declare_requires_collection_and_branch_write_authority() -> TestResult {
let mut fixture = Fixture::new()?;
let operation = Operation::Mutate(fixture.request("main", declaration("ward")?)?);
let subjects = haematite_cli::serve::authorize::subjects(&operation)?;
assert_eq!(subjects.len(), 2);
for (subject, wire, name) in [
(&subjects[0], "index.declare", "notes"),
(&subjects[1], "mutation.apply", "main"),
] {
assert_eq!(subject.route.wire, wire);
assert_eq!(subject.name.as_deref(), Some(name));
assert_eq!(subject.route.action, "write");
assert_eq!(
subject.route.class,
haematite_cli::serve::route_matrix::Class::Hot
);
}
Ok(())
}
#[test]
fn index_declare_fast_forward_merge_carries_the_recorded_event() -> TestResult {
let mut fixture = Fixture::new()?;
let from = fixture.capture("main", None)?.version;
fixture.apply(
"main",
Change::Fork(Fork {
name: "work".to_owned(),
from,
}),
)?;
let reviewed = fixture.apply("work", declaration("ward")?)?.version;
let source = fixture
.declaration("work", None)?
.ok_or("source lost declaration")?;
assert!(fixture.declaration("main", None)?.is_none());
let merged = fixture.apply(
"main",
Change::Merge(BranchMerge {
from: "work".to_owned(),
reviewed,
}),
)?;
assert_eq!(
fixture.declaration("main", Some(merged.version))?,
Some(source.clone())
);
assert_eq!(fixture.declaration("work", None)?, Some(source));
assert!(fixture.declaration("main", Some(from))?.is_none());
Ok(())
}
fn embedded_declaration() -> Result<Change, serde_json::Error> {
serde_json::from_value(json!({"kind":"vector.declare", "value": {
"collection":"notes", "field":"ward", "declaration": {
"kind":"dense", "source":"body",
"model":{"name":"local-model", "revision":"v1", "manifest_sha256":"a".repeat(64)},
"dimensions":2, "distance":"cosine"
}
}}))
}
#[test]
fn index_declare_and_embedded_vectors_refuse_each_other_without_writes() -> TestResult {
let mut vector = Fixture::new()?;
vector.apply("main", embedded_declaration()?)?;
vector.refuse(declaration("ward")?, "index_vector_field")?;
assert!(vector.declaration("main", None)?.is_none());
let mut equality = Fixture::new()?;
equality.apply("main", declaration("ward")?)?;
equality.refuse(embedded_declaration()?, "index_vector_field")?;
assert!(equality.declaration("main", None)?.is_some());
Ok(())
}