use std::error::Error;
use super::{MAX_VERSION_ROWS, get, version_rows};
use crate::serve::contracts::{
MutationRequest, ResolvedMutation,
attribution::{ActorKind, Changed, Row, key},
};
use crate::serve::protocol::{Selection, Version};
use crate::serve::storage::{Captured, Storage, branch_name};
use haematite::branch::{CommitDurability, CommitRequest, current_timestamp, open};
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::json;
type TestResult = Result<(), Box<dyn Error>>;
type RawRow = (Vec<u8>, Vec<u8>);
fn setup() -> Result<(tempfile::TempDir, Storage), 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((directory, storage))
}
fn capture(storage: &Storage) -> Result<Captured, Box<dyn Error>> {
Ok(storage.capture(&Selection {
branch: "main".to_owned(),
version: None,
})?)
}
fn request(id: &str, caller: &str, origin: Version) -> Result<ResolvedMutation, Box<dyn Error>> {
Ok(MutationRequest::decode(&serde_json::to_vec(&json!({
"request_id":id, "branch":"main", "caller":{"issuer":"lys", "subject":caller},
"contract":{"kind":"dependencies", "origin":origin, "reads":[]},
"change":{"kind":"document.put", "value":{"collection":"charts", "document_id":id, "document":{}}}
}))?)?.resolve("operator-not-the-author")?)
}
const fn next(origin: Version) -> Version {
Version {
generation: origin.generation,
sequence: origin.sequence + 1,
}
}
fn install(storage: &mut Storage, rows: Vec<RawRow>) -> TestResult {
let name = branch_name("main");
let branch = open(&name, &storage.refs, &storage.registry)?;
for (key, value) in rows {
branch.put(storage.writer.shard_for(&key), key, value)?;
}
storage.writer.commit(
&branch,
&storage.registry,
CommitRequest {
durability: CommitDurability::Durable {
refs: &mut storage.refs,
},
extra_parents: &[],
timestamp: current_timestamp(),
},
)?;
storage.publish_record(&name)?;
Ok(())
}
fn other_shard_id(
storage: &Storage,
written: Version,
first_shard: usize,
) -> Result<String, Box<dyn Error>> {
for index in 0..1000 {
let id = format!("second-{index:04}");
let candidate = key(written, &id)?;
if storage.writer.shard_for(&candidate) != first_shard {
return Ok(id);
}
}
Err("could not find an id in the other shard".into())
}
#[test]
fn a_version_reader_returns_two_callers_in_stable_request_order_across_shards() -> TestResult {
let (_directory, mut storage) = setup()?;
let origin = capture(&storage)?.version;
let written = next(origin);
let first = Row::for_request(&request("first", "doctor-a", origin)?, written)?;
let first_key = first.key()?;
let first_shard = storage.writer.shard_for(&first_key);
let second_id = other_shard_id(&storage, written, first_shard)?;
let second = Row::for_request(&request(&second_id, "doctor-b", origin)?, written)?;
install(
&mut storage,
vec![
(second.key()?, second.encode()?),
(first_key, first.encode()?),
],
)?;
let at = capture(&storage)?;
let rows = version_rows(&storage.reader()?, &at, written)?;
assert_eq!(rows.len(), 2);
assert_eq!(rows[0].request_id(), "first");
assert_eq!(rows[0].caller().subject(), "doctor-a");
assert_eq!(rows[1].request_id(), second_id);
assert_eq!(rows[1].caller().subject(), "doctor-b");
let one = get(&storage.reader()?, &at, written, "first")?.ok_or("missing attribution")?;
assert_eq!(one.origin(), origin);
Ok(())
}
#[test]
fn a_point_read_returns_none_for_a_missing_row() -> TestResult {
let (_directory, storage) = setup()?;
let at = capture(&storage)?;
assert!(get(&storage.reader()?, &at, at.version, "missing")?.is_none());
assert!(version_rows(&storage.reader()?, &at, at.version)?.is_empty());
Ok(())
}
#[test]
fn a_reader_refuses_a_row_stored_under_another_request_key() -> TestResult {
let (_directory, mut storage) = setup()?;
let origin = capture(&storage)?.version;
let written = next(origin);
let row = Row::for_request(&request("original", "doctor-a", origin)?, written)?;
install(&mut storage, vec![(key(written, "moved")?, row.encode()?)])?;
let at = capture(&storage)?;
let error = get(&storage.reader()?, &at, written, "moved")
.err()
.ok_or("moved row accepted")?;
assert_eq!(error.code, "change_record_key_mismatch");
Ok(())
}
#[test]
fn a_version_reader_refuses_overflow_instead_of_truncating() -> TestResult {
let (_directory, mut storage) = setup()?;
let origin = capture(&storage)?.version;
let written = next(origin);
let mut raw = Vec::new();
for index in 0..=MAX_VERSION_ROWS {
let row = Row::for_request(
&request(&format!("write-{index:04}"), "doctor-a", origin)?,
written,
)?;
raw.push((row.key()?, row.encode()?));
}
install(&mut storage, raw)?;
let at = capture(&storage)?;
let error = version_rows(&storage.reader()?, &at, written)
.err()
.ok_or("overflow was truncated or accepted")?;
assert_eq!(error.code, "change_record_version_limit");
Ok(())
}
#[test]
fn rows_in_other_versions_do_not_use_the_requested_versions_cap() -> TestResult {
let (_directory, mut storage) = setup()?;
let origin = capture(&storage)?.version;
let written = next(origin);
let later = next(written);
let mut raw = Vec::new();
for index in 0..=MAX_VERSION_ROWS {
let row = Row::for_request(
&request(&format!("old-{index:04}"), "doctor-a", origin)?,
written,
)?;
raw.push((row.key()?, row.encode()?));
}
let wanted = Row::for_request(&request("wanted", "doctor-b", origin)?, later)?;
install(&mut storage, raw)?;
install(&mut storage, vec![(wanted.key()?, wanted.encode()?)])?;
let at = capture(&storage)?;
let rows = version_rows(&storage.reader()?, &at, later)?;
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].request_id(), "wanted");
Ok(())
}
#[test]
fn reader_keeps_agent_responsibility_without_an_operator_identity() -> TestResult {
let (_directory, mut storage) = setup()?;
let origin = capture(&storage)?.version;
let written = next(origin);
let request = MutationRequest::decode(&serde_json::to_vec(&json!({
"request_id":"agent", "branch":"main", "caller":{"issuer":"lys", "subject":"agent-a", "responsible":"doctor-a"},
"contract":{"kind":"exact_version", "expected":origin},
"change":{"kind":"documents.put_many", "value":{"collection":"charts", "documents":[
{"document_id":"a", "document":{}}, {"document_id":"b", "document":{}}, {"document_id":"c", "document":{}}
]}}
}))?)?.resolve("operator-not-the-author")?;
let row = Row::for_request(&request, written)?;
install(&mut storage, vec![(row.key()?, row.encode()?)])?;
let at = capture(&storage)?;
let rows = version_rows(&storage.reader()?, &at, written)?;
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].actor_kind(), ActorKind::Agent);
assert_eq!(rows[0].caller().responsible(), Some("doctor-a"));
assert!(matches!(
rows[0].changed(),
Changed::Import { count: 3, .. }
));
Ok(())
}
#[test]
fn the_same_request_id_is_read_independently_at_two_versions() -> TestResult {
let (_directory, mut storage) = setup()?;
let origin = capture(&storage)?.version;
let first = next(origin);
let second = next(first);
let request = request("same-id", "doctor-a", origin)?;
let row1 = Row::for_request(&request, first)?;
let row2 = Row::for_request(&request, second)?;
install(&mut storage, vec![(row1.key()?, row1.encode()?)])?;
install(&mut storage, vec![(row2.key()?, row2.encode()?)])?;
let at = capture(&storage)?;
let reader = storage.reader()?;
assert_eq!(
get(&reader, &at, first, "same-id")?
.ok_or("first row missing")?
.version(),
first
);
assert_eq!(
get(&reader, &at, second, "same-id")?
.ok_or("second row missing")?
.version(),
second
);
assert_eq!(version_rows(&reader, &at, first)?.len(), 1);
assert_eq!(version_rows(&reader, &at, second)?.len(), 1);
Ok(())
}
#[test]
fn version_reader_accepts_the_exact_cap() -> TestResult {
let (_directory, mut storage) = setup()?;
let origin = capture(&storage)?.version;
let written = next(origin);
let mut raw = Vec::new();
for index in 0..MAX_VERSION_ROWS {
let row = Row::for_request(
&request(&format!("write-{index:04}"), "doctor-a", origin)?,
written,
)?;
raw.push((row.key()?, row.encode()?));
}
install(&mut storage, raw)?;
let rows = version_rows(&storage.reader()?, &capture(&storage)?, written)?;
assert_eq!(rows.len(), MAX_VERSION_ROWS);
Ok(())
}
#[test]
fn incomplete_selections_are_refused_instead_of_omitting_a_shard() -> TestResult {
let (_directory, storage) = setup()?;
let mut at = capture(&storage)?;
at.shards.pop().ok_or("fixture has no shards")?;
let reader = storage.reader()?;
assert_eq!(
get(&reader, &at, at.version, "write")
.err()
.ok_or("partial selection accepted")?
.code,
"incomplete_roots"
);
assert_eq!(
version_rows(&reader, &at, at.version)
.err()
.ok_or("partial page accepted")?
.code,
"incomplete_roots"
);
Ok(())
}
#[test]
fn duplicate_keys_in_different_shards_are_refused_as_ambiguous_attribution() -> TestResult {
let (_directory, mut storage) = setup()?;
let origin = capture(&storage)?.version;
let written = next(origin);
let row = Row::for_request(&request("write", "doctor-a", origin)?, written)?;
let name = branch_name("main");
let branch = open(&name, &storage.refs, &storage.registry)?;
for shard in 0..2 {
branch.put(shard, row.key()?, row.encode()?)?;
}
storage.writer.commit(
&branch,
&storage.registry,
CommitRequest {
durability: CommitDurability::Durable {
refs: &mut storage.refs,
},
extra_parents: &[],
timestamp: current_timestamp(),
},
)?;
storage.publish_record(&name)?;
let at = capture(&storage)?;
let error = version_rows(&storage.reader()?, &at, written)
.err()
.ok_or("ambiguous authorship accepted")?;
assert_eq!(error.code, "change_record_key_invalid");
Ok(())
}
const fn fixture_long_wait_interval() -> std::time::Duration {
std::time::Duration::from_secs(1)
}