#![forbid(unsafe_code)]
use std::error::Error;
use std::num::{NonZeroU32, NonZeroUsize};
use haematite::db::DatabaseConfig;
use haematite::db::vacuum::{Options as VacuumOptions, sweep as vacuum_sweep};
use haematite::store::NodeCacheBudget;
use haematite_cli::serve::access::{self, Standalone};
use haematite_cli::serve::protocol::{
Change, CollectionWrite, DocumentRead, DocumentWrite, Fork, Mutation, Operation, Selection,
};
use haematite_cli::serve::service::Service;
use haematite_cli::serve::storage::{Receipt, Storage, reads};
use serde_json::{json, value::RawValue};
type TestResult = Result<(), Box<dyn Error>>;
fn setup() -> Result<(tempfile::TempDir, Storage), Box<dyn Error>> {
let dir = tempfile::tempdir()?;
let storage = Storage::create(
DatabaseConfig {
data_dir: dir.path().join("database"),
shard_count: 2,
distributed: None,
executor_threads: Some(2),
node_cache_budget: Some(NodeCacheBudget::Unlimited),
retention: haematite::Retention::KeepAll,
},
NodeCacheBudget::Unlimited,
long_wait_interval(),
)?;
Ok((dir, storage))
}
fn current(branch: &str) -> Selection {
Selection {
branch: branch.to_owned(),
version: None,
}
}
fn put(document: &str) -> Result<Change, serde_json::Error> {
Ok(Change::Put(DocumentWrite {
collection: "records".to_owned(),
document_id: "one".to_owned(),
document: RawValue::from_string(document.to_owned())?,
}))
}
fn request(storage: &Storage, id: &str, change: Change) -> Result<Mutation, Box<dyn Error>> {
Ok(Mutation {
request_id: id.to_owned(),
branch: "main".to_owned(),
expected: storage.capture(¤t("main"))?.version,
change,
})
}
fn apply(storage: &mut Storage, id: &str, change: Change) -> Result<Receipt, Box<dyn Error>> {
let request = request(storage, id, change)?;
Ok(storage.mutate(&request)?)
}
fn collection(storage: &mut Storage) -> TestResult {
apply(
storage,
"declare",
Change::Collection(CollectionWrite {
collection: "records".to_owned(),
schema: Some(
json!({"type":"object", "required":["name"], "properties":{"name":{"type":"string"}}}),
),
}),
)?;
Ok(())
}
#[test]
fn schema_refusal_raw_extras_and_schema_survive_restart() -> TestResult {
let (dir, mut storage) = setup()?;
collection(&mut storage)?;
let invalid = request(&storage, "bad", put(r#"{"extra":true}"#)?)?;
let refusal = storage
.mutate(&invalid)
.err()
.ok_or("invalid document was accepted")?;
assert_eq!(refusal.code, "schema_refused");
assert_eq!(refusal.path.as_deref(), Some("/name"));
assert_eq!(storage.capture(¤t("main"))?.version, invalid.expected);
let raw = r#"{ "name": "kept", "extra": {"number": 1.00} }"#;
apply(&mut storage, "good", put(raw)?)?;
drop(storage);
let storage = Storage::open(
&dir.path().join("database"),
NodeCacheBudget::Unlimited,
long_wait_interval(),
)?;
let reader = storage.reader()?;
let at = storage.capture(¤t("main"))?;
assert!(
reads::collection(&reader, &at, "records")?
.ok_or("missing collection")?
.schema
.is_some()
);
let service = Service::new(storage, NonZeroU32::MIN)?;
let answer = service.dispatch(Operation::Get(DocumentRead {
selection: current("main"),
collection: "records".to_owned(),
document_id: "one".to_owned(),
}))?;
assert_eq!(answer["document"]["raw_json"], raw);
Ok(())
}
#[test]
fn lost_reply_intervening_write_restart_and_replay_preserve_newer_data() -> TestResult {
let (dir, mut storage) = setup()?;
collection(&mut storage)?;
let original = request(&storage, "original", put(r#"{"name":"earlier"}"#)?)?;
let receipt = storage.mutate(&original)?; apply(&mut storage, "later", put(r#"{"name":"later"}"#)?)?;
drop(storage);
let mut storage = Storage::open(
&dir.path().join("database"),
NodeCacheBudget::Unlimited,
long_wait_interval(),
)?;
let replay = storage.mutate(&original)?;
assert_eq!(receipt.version, replay.version);
assert_eq!(receipt.fingerprint, replay.fingerprint);
let at = storage.capture(¤t("main"))?;
assert!(at.version.sequence > replay.version.sequence);
let changed = Mutation {
change: put(r#"{"name":"different"}"#)?,
..original
};
assert_eq!(
storage
.mutate(&changed)
.err()
.ok_or("conflicting request accepted")?
.code,
"request_conflict"
);
let service = Service::new(storage, NonZeroU32::MIN)?;
let answer = service.dispatch(Operation::Get(DocumentRead {
selection: current("main"),
collection: "records".to_owned(),
document_id: "one".to_owned(),
}))?;
assert_eq!(answer["document"]["raw_json"], r#"{"name":"later"}"#);
Ok(())
}
#[test]
fn historical_fork_and_receipt_survive_vacuum_and_restart() -> TestResult {
let (dir, mut storage) = setup()?;
collection(&mut storage)?;
let old = apply(&mut storage, "old", put(r#"{"name":"old"}"#)?)?.version;
apply(&mut storage, "new", put(r#"{"name":"new"}"#)?)?;
let fork = request(
&storage,
"fork",
Change::Fork(Fork {
name: "earlier".to_owned(),
from: old,
}),
)?;
let receipt = storage.mutate(&fork)?;
let historical = storage.capture(&Selection {
branch: "main".to_owned(),
version: Some(old),
})?;
assert_eq!(
reads::document(
&access::request(&Standalone)?,
&storage.reader()?,
&historical,
"records",
"one"
)?
.ok_or("missing history")?
.raw_json,
r#"{"name":"old"}"#
);
drop(storage);
let mut options = VacuumOptions::new(dir.path().join("database"));
options.attest_metadata_complete = true; vacuum_sweep(&options, true)?;
let mut storage = Storage::open(
&dir.path().join("database"),
NodeCacheBudget::Unlimited,
long_wait_interval(),
)?;
assert_eq!(storage.mutate(&fork)?.version, receipt.version);
let at = storage.capture(¤t("earlier"))?;
assert_eq!(
reads::document(
&access::request(&Standalone)?,
&storage.reader()?,
&at,
"records",
"one"
)?
.ok_or("missing fork")?
.raw_json,
r#"{"name":"old"}"#
);
Ok(())
}
#[test]
fn readers_keep_their_captured_version_while_writes_advance() -> TestResult {
let (_dir, mut storage) = setup()?;
collection(&mut storage)?;
apply(&mut storage, "initial", put(r#"{"name":"initial"}"#)?)?;
let at = storage.capture(¤t("main"))?;
let readers = [storage.reader()?, storage.reader()?];
std::thread::scope(|scope| -> TestResult {
let (finished, completions) = std::sync::mpsc::channel();
let mut commands = Vec::new();
let mut workers = Vec::new();
for reader in readers {
let (start, requests) = std::sync::mpsc::channel();
commands.push(start);
let at = at.clone();
let finished = finished.clone();
workers.push(scope.spawn(move || -> Result<(), String> {
while requests
.recv_timeout(std::time::Duration::from_secs(5))
.is_ok()
{
let result = reads::documents(
&access::request(&Standalone).map_err(|error| error.to_string())?,
&reader,
&at,
"records",
None,
NonZeroUsize::MIN,
)
.map_err(|error| error.to_string())
.and_then(|page| {
if page.entries.first().map(|row| row.raw_json.as_str())
== Some(r#"{"name":"initial"}"#)
{
Ok(())
} else {
Err("captured reader changed version".to_owned())
}
});
finished.send(result).map_err(|error| error.to_string())?;
}
Ok(())
}));
}
for index in 0..3 {
for start in &commands {
start.send(())?;
}
apply(
&mut storage,
&format!("advance-{index}"),
put(r#"{"name":"newer"}"#)?,
)?;
for _ in 0..2 {
completions
.recv_timeout(std::time::Duration::from_secs(5))?
.map_err(std::io::Error::other)?;
}
}
drop(commands);
for worker in workers {
worker
.join()
.map_err(|_| "reader panicked")?
.map_err(std::io::Error::other)?;
}
Ok(())
})?;
Ok(())
}
#[test]
fn document_keeps_its_schema_revision_and_large_numbers_are_wire_safe() -> TestResult {
let (_dir, mut storage) = setup()?;
collection(&mut storage)?;
let schema_version = storage.capture(¤t("main"))?.version;
let raw = r#"{"name":"number", "large":18446744073709551616001, "exponent":1e400}"#;
let written = apply(&mut storage, "large", put(raw)?)?.version;
apply(
&mut storage,
"new-schema",
Change::Collection(CollectionWrite {
collection: "records".to_owned(),
schema: Some(json!({"required":["future"]})),
}),
)?;
let at = storage.capture(¤t("main"))?;
let document = reads::document(
&access::request(&Standalone)?,
&storage.reader()?,
&at,
"records",
"one",
)?
.ok_or("missing document")?;
assert_eq!(document.revision.collection_version, schema_version);
assert_eq!(document.revision.collection_branch, "main");
assert_eq!(document.revision.written_at, written);
assert_eq!(document.raw_json, raw);
let wire = serde_json::to_value(&document)?;
assert_eq!(wire["raw_json"], raw);
Ok(())
}
#[test]
fn bounded_pages_keep_order_and_version_when_byte_budget_splits_rows() -> TestResult {
let (_dir, mut storage) = setup()?;
collection(&mut storage)?;
for id in ["d", "a", "c", "b"] {
apply(
&mut storage,
id,
Change::Put(DocumentWrite {
collection: "records".to_owned(),
document_id: id.to_owned(),
document: RawValue::from_string(serde_json::to_string(
&json!({"name":"x".repeat(512)}),
)?)?,
}),
)?;
}
let reader = storage.reader()?;
let at = storage.capture(¤t("main"))?;
let first = reads::document(&access::request(&Standalone)?, &reader, &at, "records", "a")?
.ok_or("missing first document")?;
let budget = NonZeroUsize::new(serde_json::to_vec(&first)?.len() + 1).ok_or("empty budget")?;
apply(
&mut storage,
"later",
Change::Put(DocumentWrite {
collection: "records".to_owned(),
document_id: "later".to_owned(),
document: RawValue::from_string(r#"{"name":"not in the captured page"}"#.to_owned())?,
}),
)?;
let mut after = None;
for expected in ["a", "b", "c", "d"] {
let page = reads::documents_bounded(
&access::request(&Standalone)?,
&reader,
&at,
"records",
after.as_deref(),
reads::PageLimits {
rows: NonZeroUsize::new(20).ok_or("invalid count")?,
bytes: budget,
},
)?;
assert_eq!(page.version, at.version);
assert_eq!(page.entries.len(), 1);
assert_eq!(page.entries[0].id, expected);
after = page.next_after;
assert_eq!(
after.as_deref(),
if expected == "d" {
None
} else {
Some(expected)
}
);
}
let refusal = reads::documents_bounded(
&access::request(&Standalone)?,
&reader,
&at,
"records",
None,
reads::PageLimits {
rows: NonZeroUsize::MIN,
bytes: NonZeroUsize::MIN,
},
)
.err()
.ok_or("oversized row was accepted")?;
assert_eq!(refusal.code, "response_limit");
Ok(())
}
const fn long_wait_interval() -> std::time::Duration {
const LONG_WAIT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
LONG_WAIT_INTERVAL
}