use std::num::{NonZeroU32, NonZeroUsize};
use std::sync::Arc;
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::value::RawValue;
use super::Service;
use crate::serve::access::{self, Source, Standalone, TestSource};
use crate::serve::protocol::{
BranchDiscard, BranchPage, Change, CollectionWrite, DocumentPage, DocumentRead, DocumentWrite,
Fork, Mutation, Operation, ResponseFailure, Selection, Version,
};
use crate::serve::storage::{Storage, reads};
type TestResult = Result<(), Box<dyn std::error::Error>>;
fn with_storage(test: impl FnOnce(Storage) -> TestResult) -> TestResult {
let directory = tempfile::tempdir()?;
let mut storage = Storage::create(
DatabaseConfig {
data_dir: directory.path().join("database"),
shard_count: 1,
distributed: None,
executor_threads: Some(1),
node_cache_budget: Some(NodeCacheBudget::Unlimited),
retention: haematite::Retention::KeepAll,
},
NodeCacheBudget::Unlimited,
fixture_long_wait_interval(),
)?;
let initial = storage.capture(&selection())?.version;
let declared = storage.mutate(&Mutation {
request_id: "declare".to_owned(),
branch: "main".to_owned(),
expected: initial,
change: Change::Collection(CollectionWrite {
collection: "records".to_owned(),
schema: None,
}),
})?;
storage.mutate(&Mutation {
request_id: "write".to_owned(),
branch: "main".to_owned(),
expected: declared.version,
change: Change::Put(DocumentWrite {
collection: "records".to_owned(),
document_id: "one".to_owned(),
document: RawValue::from_string(r#"{"sensitive":"marker"}"#.to_owned())?,
}),
})?;
test(storage)
}
fn selection() -> Selection {
Selection {
branch: "main".to_owned(),
version: None,
}
}
fn get(id: &str) -> Operation {
Operation::Get(DocumentRead {
selection: selection(),
collection: "records".to_owned(),
document_id: id.to_owned(),
})
}
fn response_limit() -> Result<NonZeroUsize, Box<dyn std::error::Error>> {
NonZeroUsize::new(4096).ok_or_else(|| "zero response limit".into())
}
fn with_discard_storage(test: impl FnOnce(Storage, Version) -> TestResult) -> TestResult {
with_storage(|mut storage| {
let initial = storage.capture(&selection())?.version;
let declared = storage.mutate(&Mutation {
request_id: "declare-private".to_owned(),
branch: "main".to_owned(),
expected: initial,
change: Change::Collection(CollectionWrite {
collection: "private".to_owned(),
schema: None,
}),
})?;
let written = storage.mutate(&Mutation {
request_id: "write-private".to_owned(),
branch: "main".to_owned(),
expected: declared.version,
change: Change::Put(DocumentWrite {
collection: "private".to_owned(),
document_id: "hidden".to_owned(),
document: RawValue::from_string(r#"{"value":1}"#.to_owned())?,
}),
})?;
let fork = storage.mutate(&Mutation {
request_id: "fork-work".to_owned(),
branch: "main".to_owned(),
expected: written.version,
change: Change::Fork(Fork {
name: "work".to_owned(),
from: written.version,
}),
})?;
test(storage, fork.version)
})
}
fn work_selection() -> Selection {
Selection {
branch: "work".to_owned(),
version: None,
}
}
fn discard_request(expected: Version) -> Mutation {
Mutation {
request_id: "discard-work".to_owned(),
branch: "work".to_owned(),
expected,
change: Change::Discard(BranchDiscard {
name: "work".to_owned(),
}),
}
}
fn limited_source() -> Arc<TestSource> {
Arc::new(TestSource::denying(&[("private", "hidden")]).allowing_collection("records"))
}
fn dispatch_path(
service: &Service,
operation: Operation,
bounded: bool,
) -> Result<serde_json::Value, ResponseFailure> {
if bounded {
service.dispatch_bounded(operation, NonZeroUsize::MIN.saturating_add(4095))
} else {
service.dispatch(operation)
}
}
fn assert_discard_refusal(failure: ResponseFailure) -> TestResult {
let answer = serde_json::to_value(failure)?;
assert_eq!(answer["code"], "discard_access_refused");
assert_eq!(answer["outcome"], "not_applied");
assert_eq!(answer["request_id"], "discard-work");
assert_eq!(answer["receipt_branch"], "work");
Ok(())
}
#[test]
fn limited_caller_cannot_discard_a_listed_readable_branch() -> TestResult {
for bounded in [false, true] {
with_discard_storage(|storage, expected| {
let source = limited_source();
let service = Service::with_access_source(
storage,
NonZeroU32::MIN.saturating_add(7),
None,
source.clone(),
)?;
let before = super::view(&service.catalogue.capture(&work_selection())?);
let refused = dispatch_path(
&service,
Operation::Mutate(discard_request(expected)),
bounded,
)
.err()
.ok_or("limited caller discarded the branch")?;
assert_discard_refusal(refused)?;
assert_eq!(source.calls(), 1);
assert_eq!(
super::view(&service.catalogue.capture(&work_selection())?),
before
);
let branches = dispatch_path(
&service,
Operation::Branches(BranchPage {
after: None,
limit: NonZeroU32::MIN.saturating_add(7),
}),
bounded,
)?;
let work = branches["entries"]
.as_array()
.ok_or("branch listing has no entries")?
.iter()
.find(|entry| entry["branch"] == "work")
.ok_or("refused discard hid the branch")?;
assert_eq!(work, &before);
let read = dispatch_path(
&service,
Operation::Get(DocumentRead {
selection: work_selection(),
collection: "records".to_owned(),
document_id: "one".to_owned(),
}),
bounded,
)?;
assert_eq!(read["document"]["raw_json"], r#"{"sensitive":"marker"}"#);
assert_eq!(source.calls(), 3);
Ok(())
})?;
}
Ok(())
}
#[test]
fn limited_caller_cannot_replay_a_completed_discard() -> TestResult {
for bounded in [false, true] {
with_discard_storage(|mut storage, expected| {
storage.mutate(&discard_request(expected))?;
let source = limited_source();
let service =
Service::with_access_source(storage, NonZeroU32::MIN, None, source.clone())?;
let refused = dispatch_path(
&service,
Operation::Mutate(discard_request(expected)),
bounded,
)
.err()
.ok_or("limited caller replayed a completed discard")?;
assert_discard_refusal(refused)?;
assert_eq!(source.calls(), 1);
assert_eq!(
service
.catalogue
.capture(&work_selection())
.err()
.ok_or("completed discard became visible")?
.code,
"branch_missing"
);
Ok(())
})?;
}
Ok(())
}
#[test]
fn whole_store_caller_discards_and_replays_the_original_receipt() -> TestResult {
for bounded in [false, true] {
with_discard_storage(|storage, expected| {
let source = Arc::new(TestSource::denying(&[]).allowing_collection("records"));
let service =
Service::with_access_source(storage, NonZeroU32::MIN, None, source.clone())?;
let receipt = dispatch_path(
&service,
Operation::Mutate(discard_request(expected)),
bounded,
)?;
assert_eq!(receipt["branch"], "work");
assert_eq!(receipt["request_id"], "discard-work");
assert_eq!(receipt["version"], serde_json::to_value(expected)?);
let replay = dispatch_path(
&service,
Operation::Mutate(discard_request(expected)),
bounded,
)?;
assert_eq!(replay, receipt);
assert_eq!(source.calls(), 2);
assert_eq!(
service
.catalogue
.capture(&work_selection())
.err()
.ok_or("discarded branch remains visible")?
.code,
"branch_missing"
);
Ok(())
})?;
}
Ok(())
}
#[test]
fn denied_discard_is_refused_before_writer_admission() -> TestResult {
for bounded in [false, true] {
with_discard_storage(|storage, expected| {
let source = limited_source();
let service =
Service::with_access_source(storage, NonZeroU32::MIN, None, source.clone())?;
let admitted = service.admission()?;
let refused = dispatch_path(
&service,
Operation::Mutate(discard_request(expected)),
bounded,
)
.err()
.ok_or("limited caller discarded during admission")?;
drop(admitted);
assert_discard_refusal(refused)?;
assert_eq!(source.calls(), 1);
Ok(())
})?;
}
Ok(())
}
#[test]
fn discard_guard_does_not_change_other_mutation_kinds() -> TestResult {
for bounded in [false, true] {
with_storage(|storage| {
let expected = storage.capture(&selection())?.version;
let source = limited_source();
let service =
Service::with_access_source(storage, NonZeroU32::MIN, None, source.clone())?;
let receipt = dispatch_path(
&service,
Operation::Mutate(Mutation {
request_id: "put-visible".to_owned(),
branch: "main".to_owned(),
expected,
change: Change::Put(DocumentWrite {
collection: "records".to_owned(),
document_id: "visible".to_owned(),
document: RawValue::from_string(r#"{"value":2}"#.to_owned())?,
}),
}),
bounded,
)?;
assert_eq!(receipt["request_id"], "put-visible");
assert_eq!(source.calls(), 1);
let read = dispatch_path(&service, get("visible"), bounded)?;
assert_eq!(read["document"]["raw_json"], r#"{"value":2}"#);
Ok(())
})?;
}
Ok(())
}
#[test]
fn denied_document_is_byte_equal_to_an_absent_document() -> TestResult {
with_storage(|storage| {
let source = Arc::new(TestSource::denying(&[("records", "one")]));
let service = Service::with_access_source(storage, NonZeroU32::MIN, None, source)?;
let denied = service.dispatch_bounded(get("one"), response_limit()?)?;
let absent = service.dispatch_bounded(get("absent"), response_limit()?)?;
assert!(denied["document"].is_null());
assert_eq!(serde_json::to_vec(&denied)?, serde_json::to_vec(&absent)?);
Ok(())
})
}
#[test]
fn standing_alone_source_allows_the_document() -> TestResult {
with_storage(|storage| {
let service = Service::new(storage, NonZeroU32::MIN)?;
let answer = service.dispatch_bounded(get("one"), response_limit()?)?;
assert_eq!(answer["document"]["raw_json"], r#"{"sensitive":"marker"}"#);
Ok(())
})
}
#[test]
fn denial_matches_both_collection_and_document() -> TestResult {
with_storage(|storage| {
let source = Arc::new(TestSource::denying(&[
("other", "one"),
("records", "other"),
]));
let service = Service::with_access_source(storage, NonZeroU32::MIN, None, source)?;
let answer = service.dispatch_bounded(get("one"), response_limit()?)?;
assert_eq!(answer["document"]["raw_json"], r#"{"sensitive":"marker"}"#);
Ok(())
})
}
#[test]
fn each_dispatch_asks_the_source_once() -> TestResult {
with_storage(|storage| {
let source = Arc::new(TestSource::denying(&[("records", "one")]));
let service = Service::with_access_source(storage, NonZeroU32::MIN, None, source.clone())?;
assert_eq!(source.calls(), 0);
service.dispatch_bounded(get("one"), response_limit()?)?;
assert_eq!(source.calls(), 1);
service.dispatch_bounded(get("absent"), response_limit()?)?;
assert_eq!(source.calls(), 2);
let unbounded = service.dispatch(get("one"))?;
assert!(unbounded["document"].is_null());
assert_eq!(source.calls(), 3);
Ok(())
})
}
#[test]
fn source_failure_has_no_owner_fallback() -> TestResult {
with_storage(|storage| {
let source = Arc::new(TestSource::unavailable());
let service = Service::with_access_source(storage, NonZeroU32::MIN, None, source)?;
let failure = service
.dispatch_bounded(get("one"), response_limit()?)
.err()
.ok_or("unavailable source served a document")?;
assert_eq!(failure.code, "access_unavailable");
Ok(())
})
}
#[test]
fn denied_storage_point_read_does_not_read_the_tree() -> TestResult {
with_storage(|storage| {
let source = TestSource::denying(&[("records", "one")]);
let access = source.access()?;
let reader = storage.reader()?;
let mut captured = storage.capture(&selection())?;
captured.shards.clear();
let denied = reads::document(&access, &reader, &captured, "records", "one")?;
assert!(denied.is_none());
let allowed = access::request(&Standalone)?;
assert!(
reads::document(&allowed, &reader, &captured, "records", "one").is_err(),
"control: a read of a capture with no shards did not fail"
);
Ok(())
})
}
#[test]
fn malformed_identifier_is_still_refused_at_the_boundary() -> TestResult {
with_storage(|storage| {
let service = Service::new(storage, NonZeroU32::MIN)?;
let failure = service
.dispatch_bounded(get(""), response_limit()?)
.err()
.ok_or("empty document identifier was accepted")?;
assert_eq!(failure.failure().code, "invalid_identifier");
Ok(())
})
}
fn with_list_storage(
test: impl FnOnce(Storage, crate::serve::storage::Captured) -> TestResult,
) -> TestResult {
let directory = tempfile::tempdir()?;
let mut storage = Storage::create(
DatabaseConfig {
data_dir: directory.path().join("database"),
shard_count: 1,
distributed: None,
executor_threads: Some(1),
node_cache_budget: Some(NodeCacheBudget::Unlimited),
retention: haematite::Retention::KeepAll,
},
NodeCacheBudget::Unlimited,
fixture_long_wait_interval(),
)?;
let initial = storage.capture(&selection())?.version;
storage.mutate(&Mutation {
request_id: "declare".to_owned(),
branch: "main".to_owned(),
expected: initial,
change: Change::Collection(CollectionWrite {
collection: "records".to_owned(),
schema: None,
}),
})?;
for id in ["a", "c", "d", "e"] {
let expected = storage.capture(&selection())?.version;
storage.mutate(&Mutation {
request_id: id.to_owned(),
branch: "main".to_owned(),
expected,
change: Change::Put(DocumentWrite {
collection: "records".to_owned(),
document_id: id.to_owned(),
document: RawValue::from_string(format!(r#"{{"value":"{id}"}}"#))?,
}),
})?;
}
let absent = storage.capture(&selection())?;
storage.mutate(&Mutation {
request_id: "b".to_owned(),
branch: "main".to_owned(),
expected: absent.version,
change: Change::Put(DocumentWrite {
collection: "records".to_owned(),
document_id: "b".to_owned(),
document: RawValue::from_string(serde_json::to_string(
&serde_json::json!({"hidden": "x".repeat(4096)}),
)?)?,
}),
})?;
test(storage, absent)
}
fn row_limit(rows: usize) -> Result<NonZeroUsize, Box<dyn std::error::Error>> {
NonZeroUsize::new(rows).ok_or_else(|| "zero row limit".into())
}
fn list(
after: Option<String>,
version: Option<crate::serve::protocol::Version>,
rows: u32,
) -> Result<Operation, Box<dyn std::error::Error>> {
Ok(Operation::List(DocumentPage {
selection: Selection {
branch: "main".to_owned(),
version,
},
collection: "records".to_owned(),
prefix: None,
content: None,
after,
limit: NonZeroU32::new(rows).ok_or("zero page limit")?,
}))
}
fn ids(page: &reads::Page<reads::Document>) -> Vec<&str> {
page.entries.iter().map(|entry| entry.id.as_str()).collect()
}
fn list_payload(page: &reads::Page<reads::Document>) -> Result<Vec<u8>, serde_json::Error> {
serde_json::to_vec(&serde_json::json!({"entries": page.entries, "next_after": page.next_after}))
}
#[test]
fn denied_document_is_excluded_from_list_count_and_bytes() -> TestResult {
with_list_storage(|storage, absent| {
let reader = storage.reader()?;
let at = storage.capture(&selection())?;
let access = access::request(&TestSource::denying(&[("records", "b")]))?;
let page = reads::documents(&access, &reader, &at, "records", None, row_limit(5)?)?;
let expected = reads::documents(
&access::request(&Standalone)?,
&reader,
&absent,
"records",
None,
row_limit(5)?,
)?;
assert_eq!(ids(&page), ["a", "c", "d", "e"]);
assert_eq!(page.entries.len(), 4);
assert_eq!(
serde_json::to_vec(&page.entries)?,
serde_json::to_vec(&expected.entries)?
);
assert_eq!(
serde_json::to_vec(&page.entries)?.len(),
serde_json::to_vec(&expected.entries)?.len()
);
assert!(page.next_after.is_none());
Ok(())
})
}
#[test]
fn list_continuation_names_the_last_visible_document() -> TestResult {
with_list_storage(|storage, _| {
let source = Arc::new(TestSource::denying(&[("records", "b")]));
let service = Service::with_access_source(
storage,
NonZeroU32::new(5).ok_or("zero max page")?,
None,
source.clone(),
)?;
for bounded in [false, true] {
let first_operation = list(None, None, 2)?;
let first = if bounded {
service.dispatch_bounded(first_operation, response_limit()?)?
} else {
service.dispatch(first_operation)?
};
let entries = first["entries"].as_array().ok_or("missing entries")?;
assert_eq!(entries.len(), 2);
assert_eq!(entries[0]["id"], "a");
assert_eq!(entries[1]["id"], "c");
assert_eq!(first["next_after"], "c");
let version = serde_json::from_value(first["selection"]["version"].clone())?;
let second_operation = list(Some("c".to_owned()), Some(version), 2)?;
let second = if bounded {
service.dispatch_bounded(second_operation, response_limit()?)?
} else {
service.dispatch(second_operation)?
};
let entries = second["entries"].as_array().ok_or("missing entries")?;
assert_eq!(entries.len(), 2);
assert_eq!(entries[0]["id"], "d");
assert_eq!(entries[1]["id"], "e");
assert!(second["next_after"].is_null());
}
assert_eq!(source.calls(), 4);
Ok(())
})
}
#[test]
fn denied_document_bytes_do_not_use_the_page_budget() -> TestResult {
with_list_storage(|storage, absent| {
let reader = storage.reader()?;
let at = storage.capture(&selection())?;
let owner = access::request(&Standalone)?;
let expected = reads::documents(&owner, &reader, &absent, "records", None, row_limit(5)?)?;
let byte_limit = row_limit(
serde_json::to_vec(&expected.entries)?
.len()
.checked_sub(2)
.ok_or("missing entry array brackets")?,
)?;
let denied = access::request(&TestSource::denying(&[("records", "b")]))?;
let page = reads::documents_bounded(
&denied,
&reader,
&at,
"records",
None,
reads::PageLimits {
rows: row_limit(5)?,
bytes: byte_limit,
},
)?;
assert_eq!(ids(&page), ["a", "c", "d", "e"]);
assert_eq!(list_payload(&page)?, list_payload(&expected)?);
let control = reads::documents_bounded(
&owner,
&reader,
&at,
"records",
None,
reads::PageLimits {
rows: row_limit(5)?,
bytes: byte_limit,
},
)?;
assert_eq!(ids(&control), ["a"]);
assert_eq!(control.next_after.as_deref(), Some("a"));
Ok(())
})
}
#[test]
fn denied_list_payload_is_byte_equal_to_the_never_written_capture() -> TestResult {
with_list_storage(|storage, absent| {
let reader = storage.reader()?;
let at = storage.capture(&selection())?;
let owner = access::request(&Standalone)?;
let denied = access::request(&TestSource::denying(&[("records", "b")]))?;
for after in [None, Some("a"), Some("c")] {
let expected =
reads::documents(&owner, &reader, &absent, "records", after, row_limit(2)?)?;
let unbounded =
reads::documents(&denied, &reader, &at, "records", after, row_limit(2)?)?;
let bounded = reads::documents_bounded(
&denied,
&reader,
&at,
"records",
after,
reads::PageLimits {
rows: row_limit(2)?,
bytes: response_limit()?,
},
)?;
assert_eq!(list_payload(&unbounded)?, list_payload(&expected)?);
assert_eq!(list_payload(&bounded)?, list_payload(&expected)?);
}
Ok(())
})
}
#[test]
fn standing_alone_source_lists_all_five_documents() -> TestResult {
with_list_storage(|storage, _| {
let reader = storage.reader()?;
let at = storage.capture(&selection())?;
let owner = access::request(&Standalone)?;
let page = reads::documents(&owner, &reader, &at, "records", None, row_limit(5)?)?;
assert_eq!(ids(&page), ["a", "b", "c", "d", "e"]);
assert!(page.next_after.is_none());
let bounded = reads::documents_bounded(
&owner,
&reader,
&at,
"records",
None,
reads::PageLimits {
rows: row_limit(5)?,
bytes: row_limit(16384)?,
},
)?;
assert_eq!(list_payload(&bounded)?, list_payload(&page)?);
let service = Service::new(storage, NonZeroU32::new(5).ok_or("zero max page")?)?;
let reply = service.dispatch(list(None, None, 5)?)?;
assert_eq!(
reply["entries"].as_array().ok_or("missing entries")?.len(),
5
);
assert!(reply["next_after"].is_null());
let bounded_reply = service.dispatch_bounded(list(None, None, 5)?, row_limit(16384)?)?;
assert_eq!(
serde_json::to_vec(&bounded_reply)?,
serde_json::to_vec(&reply)?
);
Ok(())
})
}
#[test]
fn denied_run_over_the_page_scan_cap_is_refused_by_name() -> TestResult {
with_list_storage(|storage, _| {
let reader = storage.reader()?;
let at = storage.capture(&selection())?;
let denied = access::request(&TestSource::denying(&[
("records", "a"),
("records", "b"),
("records", "c"),
("records", "d"),
("records", "e"),
]))?;
let owner = access::request(&Standalone)?;
for bounded in [false, true] {
let result = if bounded {
reads::documents_bounded(
&denied,
&reader,
&at,
"records",
None,
reads::PageLimits {
rows: NonZeroUsize::MIN,
bytes: response_limit()?,
},
)
} else {
reads::documents(&denied, &reader, &at, "records", None, NonZeroUsize::MIN)
};
assert_eq!(
result
.err()
.ok_or("over-cap denied run returned a page")?
.code,
"access_scan_limit"
);
}
let control = reads::documents(&owner, &reader, &at, "records", None, NonZeroUsize::MIN)?;
assert_eq!(ids(&control), ["a"]);
assert_eq!(control.next_after.as_deref(), Some("a"));
Ok(())
})
}
#[path = "service_vector_access_tests.rs"]
mod vectors;
const fn fixture_long_wait_interval() -> std::time::Duration {
std::time::Duration::from_secs(1)
}