use std::num::{NonZeroU32, NonZeroUsize};
use std::sync::Arc;
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::{Value, json, value::RawValue};
use super::{page_read_counts, reset_page_read_counts, take_document_ranges};
use crate::serve::access::{Source, TestSource};
use crate::serve::protocol::{
Change, CollectionWrite, DocumentImport, DocumentPage, ImportedDocument, Mutation, Operation,
Selection,
};
use crate::serve::service::Service;
use crate::serve::storage::{Captured, Storage};
type TestResult = Result<(), Box<dyn std::error::Error>>;
fn with_documents(
ids: &[String],
raw: &str,
test: impl FnOnce(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 selection = Selection {
branch: "main".to_owned(),
version: None,
};
let declared = storage.mutate(&Mutation {
request_id: "declare".to_owned(),
branch: "main".to_owned(),
expected: storage.capture(&selection)?.version,
change: Change::Collection(CollectionWrite {
collection: "records".to_owned(),
schema: None,
}),
})?;
if !ids.is_empty() {
let documents = ids
.iter()
.map(|id| {
Ok(ImportedDocument {
document_id: id.clone(),
document: RawValue::from_string(raw.to_owned())?,
})
})
.collect::<Result<Vec<_>, serde_json::Error>>()?;
storage.mutate_bounded(
&Mutation {
request_id: "seed".to_owned(),
branch: "main".to_owned(),
expected: declared.version,
change: Change::PutMany(DocumentImport {
collection: "records".to_owned(),
documents,
}),
},
NonZeroU32::new(16).ok_or("zero import page bound")?,
)?;
}
let at = storage.capture(&selection)?;
test(storage, at)
}
fn ids(values: &[&str]) -> Vec<String> {
values.iter().map(|value| (*value).to_owned()).collect()
}
fn request(
at: &Captured,
prefix: Option<&str>,
after: Option<&str>,
limit: u32,
) -> Result<Operation, Box<dyn std::error::Error>> {
let mut value = json!({
"selection": {"branch": at.branch, "version": at.version},
"collection": "records",
"after": after,
"limit": limit,
});
if let Some(prefix) = prefix {
value
.as_object_mut()
.ok_or("document page was not an object")?
.insert("prefix".to_owned(), Value::String(prefix.to_owned()));
}
let page: DocumentPage = serde_json::from_value(value)?;
Ok(Operation::List(page))
}
fn list(
service: &Service,
at: &Captured,
prefix: Option<&str>,
after: Option<&str>,
limit: u32,
bytes: Option<usize>,
) -> Result<Value, Box<dyn std::error::Error>> {
let operation = request(at, prefix, after, limit)?;
Ok(match bytes {
Some(bytes) => service.dispatch_bounded(
operation,
NonZeroUsize::new(bytes).ok_or("zero byte budget")?,
)?,
None => service.dispatch(operation)?,
})
}
fn answer_ids(answer: &Value) -> Result<Vec<&str>, Box<dyn std::error::Error>> {
answer["entries"]
.as_array()
.ok_or_else(|| "missing entries".into())
.and_then(|entries| {
entries
.iter()
.map(|entry| entry["id"].as_str().ok_or_else(|| "missing id".into()))
.collect()
})
}
#[test]
fn prefix_excludes_neighbor_ids_in_both_paths() -> TestResult {
with_documents(
&ids(&["a", "a/1", "a/2", "ab", "b/1"]),
"{}",
|storage, at| {
let service = Service::new(storage, NonZeroU32::new(10).ok_or("zero max page")?)?;
for bytes in [None, Some(4096)] {
let answer = list(&service, &at, Some("a/"), None, 10, bytes)?;
assert_eq!(answer_ids(&answer)?, ["a/1", "a/2"]);
assert!(answer.get("next_after").is_some());
assert!(answer["next_after"].is_null());
}
Ok(())
},
)
}
#[test]
fn an_empty_prefix_range_is_exhausted_before_collection_end() -> TestResult {
with_documents(&ids(&["a/1", "z/1"]), "{}", |storage, at| {
let service = Service::new(storage, NonZeroU32::MIN)?;
for bytes in [None, Some(4096)] {
let answer = list(&service, &at, Some("m/"), None, 1, bytes)?;
assert!(answer_ids(&answer)?.is_empty());
assert!(answer.get("next_after").is_some());
assert!(answer["next_after"].is_null());
}
Ok(())
})
}
#[test]
fn prefix_pages_return_each_id_once_and_end_with_null() -> TestResult {
with_documents(&ids(&["a/1", "a/2", "a/3", "b/1"]), "{}", |storage, at| {
let service = Service::new(storage, NonZeroU32::MIN)?;
for bytes in [None, Some(4096)] {
let mut after = None;
for (index, expected) in ["a/1", "a/2", "a/3"].into_iter().enumerate() {
let answer = list(&service, &at, Some("a/"), after.as_deref(), 1, bytes)?;
assert_eq!(answer_ids(&answer)?, [expected]);
assert_eq!(
answer["next_after"],
if index == 2 {
Value::Null
} else {
Value::String(expected.to_owned())
}
);
assert_eq!(
answer["selection"]["version"],
serde_json::to_value(at.version)?
);
after = answer["next_after"].as_str().map(str::to_owned);
}
}
Ok(())
})
}
#[test]
fn an_after_outside_the_prefix_refuses_before_reads() -> TestResult {
with_documents(&ids(&["a/1", "b/1"]), "{}", |storage, at| {
let service = Service::new(storage, NonZeroU32::MIN)?;
for bytes in [None, Some(4096)] {
reset_page_read_counts();
drop(take_document_ranges());
let failure = list(&service, &at, Some("a/"), Some("b/0"), 1, bytes)
.err()
.ok_or("outside continuation was accepted")?;
assert!(
failure
.to_string()
.contains("document_after_outside_prefix")
);
assert_eq!(page_read_counts(), super::PageReadCounts::default());
assert!(take_document_ranges().is_empty());
}
Ok(())
})
}
#[test]
fn invalid_prefixes_refuse_before_reads() -> TestResult {
with_documents(&[], "{}", |storage, at| {
let service = Service::new(storage, NonZeroU32::MIN)?;
for prefix in [String::new(), "x".repeat(513), "a\0b".to_owned()] {
for bytes in [None, Some(4096)] {
reset_page_read_counts();
drop(take_document_ranges());
let failure = list(&service, &at, Some(&prefix), None, 1, bytes)
.err()
.ok_or("invalid prefix was accepted")?;
assert!(failure.to_string().contains("invalid_identifier"));
assert_eq!(page_read_counts(), super::PageReadCounts::default());
assert!(take_document_ranges().is_empty());
}
}
Ok(())
})
}
#[test]
fn a_hidden_prefixed_document_has_the_existing_visibility_behavior() -> TestResult {
with_documents(&ids(&["a/0", "a/1", "a/2", "z/0"]), "{}", |storage, at| {
let source = Arc::new(TestSource::denying(&[("records", "a/1")]));
let access = source.access()?;
let reader = storage.reader()?;
let legacy = super::documents(
&access,
&reader,
&at,
"records",
None,
NonZeroUsize::new(10).ok_or("zero row limit")?,
)?;
let expected = legacy
.entries
.into_iter()
.filter(|entry| entry.id.starts_with("a/"))
.map(serde_json::to_value)
.collect::<Result<Vec<_>, _>>()?;
let service = Service::with_access_source(
storage,
NonZeroU32::new(10).ok_or("zero max page")?,
None,
source,
)?;
for bytes in [None, Some(4096)] {
reset_page_read_counts();
let answer = list(&service, &at, Some("a/"), None, 10, bytes)?;
assert_eq!(answer["entries"], Value::Array(expected.clone()));
assert_eq!(answer_ids(&answer)?, ["a/0", "a/2"]);
assert!(answer["next_after"].is_null());
assert_eq!(page_read_counts().points, 2);
}
Ok(())
})
}
#[test]
fn prefix_seeks_and_range_call_counts_ignore_the_other_collection_ids() -> TestResult {
let mut keys: Vec<_> = (0..997).map(|index| format!("a/{index:04}")).collect();
keys.extend(ids(&["m/1", "m/2", "m/3"]));
assert_eq!(keys.len(), 1000);
with_documents(&keys, "{}", |storage, at| {
let service = Service::new(storage, NonZeroU32::new(3).ok_or("zero max page")?)?;
let first: &[u8] = b"d\0records\0m/";
let end: &[u8] = b"d\0records\0m0";
for bytes in [None, Some(4096)] {
reset_page_read_counts();
drop(take_document_ranges());
let answer = list(&service, &at, Some("m/"), None, 3, bytes)?;
assert_eq!(answer_ids(&answer)?, ["m/1", "m/2", "m/3"]);
assert!(answer["next_after"].is_null());
let counts = page_read_counts();
assert!(counts.ranges <= 4);
let ranges = take_document_ranges();
assert!(!ranges.is_empty());
assert_eq!(ranges.len(), counts.ranges);
assert_eq!(ranges[0].0.as_slice(), first);
for (from, to) in ranges {
assert!(from.as_slice() >= first);
assert!(from.as_slice() < end);
assert!(to.as_slice() <= end);
assert_ne!(from.as_slice(), &b"d\0records\0"[..]);
}
}
Ok(())
})
}
#[test]
fn a_valid_prefix_with_no_rows_still_seeks_within_its_range() -> TestResult {
with_documents(&ids(&["a/1", "z/1"]), "{}", |storage, at| {
let service = Service::new(storage, NonZeroU32::MIN)?;
for bytes in [None, Some(4096)] {
drop(take_document_ranges());
let answer = list(&service, &at, Some("m/"), None, 1, bytes)?;
assert!(answer_ids(&answer)?.is_empty());
let ranges = take_document_ranges();
assert_eq!(
ranges,
[(b"d\0records\0m/".to_vec(), b"d\0records\0m0".to_vec())]
);
}
Ok(())
})
}
#[test]
fn byte_limited_prefix_pages_keep_the_captured_version_and_continuation() -> TestResult {
let raw = serde_json::to_string(&json!({"value": "x".repeat(1024)}))?;
with_documents(&ids(&["a/1", "a/2", "b/1"]), &raw, |storage, at| {
let service = Service::new(storage, NonZeroU32::new(10).ok_or("zero max page")?)?;
let first = list(&service, &at, Some("a/"), None, 1, None)?;
let bytes = serde_json::to_vec(&first)?.len();
let bounded = list(&service, &at, Some("a/"), None, 10, Some(bytes))?;
assert_eq!(bounded, first);
let final_page = list(&service, &at, Some("a/"), Some("a/1"), 10, Some(bytes))?;
assert_eq!(answer_ids(&final_page)?, ["a/2"]);
assert!(final_page["next_after"].is_null());
assert_eq!(
final_page["selection"]["version"],
serde_json::to_value(at.version)?
);
Ok(())
})
}
#[test]
fn a_prefixed_row_larger_than_the_byte_budget_refuses() -> TestResult {
with_documents(&ids(&["a/1"]), "{}", |storage, at| {
let service = Service::new(storage, NonZeroU32::MIN)?;
let failure = list(&service, &at, Some("a/"), None, 1, Some(1))
.err()
.ok_or("oversized row was returned")?;
assert!(failure.to_string().contains("response_limit"));
Ok(())
})
}
#[test]
fn a_prefix_matches_bytes_without_a_delimiter_rule() -> TestResult {
with_documents(&ids(&["ab", "ab/1", "abc", "ac"]), "{}", |storage, at| {
let service = Service::new(storage, NonZeroU32::new(10).ok_or("zero max page")?)?;
let answer = list(&service, &at, Some("ab"), None, 10, None)?;
assert_eq!(answer_ids(&answer)?, ["ab", "ab/1", "abc"]);
assert!(answer["next_after"].is_null());
Ok(())
})
}
#[test]
fn multibyte_prefix_and_512_byte_boundary_are_supported() -> TestResult {
let prefix = "é".repeat(256);
with_documents(
&[prefix.clone(), "other".to_owned()],
"{}",
|storage, at| {
let service = Service::new(storage, NonZeroU32::MIN)?;
for bytes in [None, Some(4096)] {
let answer = list(&service, &at, Some(&prefix), None, 1, bytes)?;
assert_eq!(answer_ids(&answer)?, [prefix.as_str()]);
assert!(answer["next_after"].is_null());
}
Ok(())
},
)
}
#[test]
fn omitted_prefix_round_trips_without_a_new_wire_member() -> TestResult {
with_documents(&ids(&["a/1", "b/1"]), "{}", |storage, at| {
let operation = request(&at, None, None, 2)?;
let Operation::List(page) = operation else {
return Err("decoded another operation".into());
};
let encoded = serde_json::to_value(&page)?;
assert!(encoded.get("prefix").is_none());
let access = TestSource::denying(&[]).access()?;
let reader = storage.reader()?;
let expected = super::documents(
&access,
&reader,
&at,
"records",
None,
NonZeroUsize::new(2).ok_or("zero row limit")?,
)?;
let service = Service::new(storage, NonZeroU32::new(2).ok_or("zero max page")?)?;
let answer = service.dispatch(Operation::List(page))?;
assert_eq!(answer["entries"], serde_json::to_value(expected.entries)?);
assert_eq!(
answer["next_after"],
serde_json::to_value(expected.next_after)?
);
Ok(())
})
}
fn content_request(
at: &Captured,
prefix: Option<&str>,
after: Option<&str>,
limit: u32,
content: Option<&str>,
) -> Result<Operation, Box<dyn std::error::Error>> {
let Operation::List(page) = request(at, prefix, after, limit)? else {
return Err("decoded another operation".into());
};
let mut value = serde_json::to_value(page)?;
if let Some(content) = content {
value
.as_object_mut()
.ok_or("document page was not an object")?
.insert("content".to_owned(), Value::String(content.to_owned()));
}
let page: DocumentPage = serde_json::from_value(value)?;
Ok(Operation::List(page))
}
fn content_list(
service: &Service,
at: &Captured,
prefix: Option<&str>,
after: Option<&str>,
limit: u32,
content: Option<&str>,
bytes: Option<usize>,
) -> Result<Value, Box<dyn std::error::Error>> {
let operation = content_request(at, prefix, after, limit, content)?;
Ok(match bytes {
Some(bytes) => service.dispatch_bounded(
operation,
NonZeroUsize::new(bytes).ok_or("zero byte budget")?,
)?,
None => service.dispatch(operation)?,
})
}
fn without_bodies(mut answer: Value) -> Result<Value, Box<dyn std::error::Error>> {
for entry in answer["entries"].as_array_mut().ok_or("missing entries")? {
let entry = entry.as_object_mut().ok_or("entry was not an object")?;
entry.remove("raw_json").ok_or("full entry had no body")?;
}
Ok(answer)
}
#[test]
fn content_none_matches_full_without_any_body_member_in_both_paths() -> TestResult {
with_documents(
&ids(&["a/1", "a/2", "b/1"]),
r#"{"body":"private"}"#,
|storage, at| {
let service = Service::new(storage, NonZeroU32::new(10).ok_or("zero max page")?)?;
for bytes in [None, Some(4096)] {
let full = content_list(&service, &at, Some("a/"), None, 10, Some("full"), bytes)?;
let light = content_list(&service, &at, Some("a/"), None, 10, Some("none"), bytes)?;
assert_eq!(light, without_bodies(full)?);
for entry in light["entries"].as_array().ok_or("missing entries")? {
assert!(entry.get("raw_json").is_none());
assert!(entry.get("revision").is_some());
}
}
Ok(())
},
)
}
#[test]
fn explicit_full_and_absent_content_keep_the_existing_answer_bytes() -> TestResult {
with_documents(
&ids(&["a/1", "a/2", "b/1"]),
r#"{ "body": [1,2] }"#,
|storage, at| {
let reader = storage.reader()?;
let access = TestSource::denying(&[]).access()?;
let service = Service::new(storage, NonZeroU32::new(10).ok_or("zero max page")?)?;
let Operation::List(page) = content_request(&at, Some("a/"), None, 10, None)? else {
return Err("decoded another operation".into());
};
assert!(serde_json::to_value(page)?.get("content").is_none());
for bytes in [None, Some(4096)] {
let rows = match bytes {
Some(bytes) => super::documents_prefixed_bounded(
&access,
&reader,
&at,
"records",
Some("a/"),
None,
super::PageLimits {
rows: NonZeroUsize::new(10).ok_or("zero row limit")?,
bytes: NonZeroUsize::new(bytes).ok_or("zero byte budget")?,
},
)?,
None => super::documents_prefixed(
&access,
&reader,
&at,
"records",
Some("a/"),
None,
NonZeroUsize::new(10).ok_or("zero row limit")?,
)?,
};
let existing = json!({
"selection": {
"branch": at.branch,
"version": at.version,
"roots": at.shards.iter().map(|shard| json!({
"shard": shard.shard_id, "root": shard.head.to_string()
})).collect::<Vec<_>>(),
},
"entries": rows.entries,
"next_after": rows.next_after,
});
let absent = content_list(&service, &at, Some("a/"), None, 10, None, bytes)?;
let full = content_list(&service, &at, Some("a/"), None, 10, Some("full"), bytes)?;
assert_eq!(serde_json::to_vec(&absent)?, serde_json::to_vec(&existing)?);
assert_eq!(serde_json::to_vec(&full)?, serde_json::to_vec(&existing)?);
}
Ok(())
},
)
}
#[test]
fn unknown_content_refuses_before_page_or_content_reads() -> TestResult {
with_documents(&ids(&["a/1"]), "{}", |storage, at| {
let service = Service::new(storage, NonZeroU32::MIN)?;
for content in ["", "ids", "FULL"] {
for bytes in [None, Some(4096)] {
reset_page_read_counts();
drop(take_document_ranges());
super::take_content_reads();
let failure = content_list(&service, &at, None, None, 1, Some(content), bytes)
.err()
.ok_or("unknown content was accepted")?;
assert!(failure.to_string().contains("invalid_document_content"));
assert_eq!(page_read_counts(), super::PageReadCounts::default());
assert!(take_document_ranges().is_empty());
assert_eq!(super::take_content_reads(), 0);
}
}
Ok(())
})
}
#[test]
fn hidden_documents_are_absent_with_full_and_no_content() -> TestResult {
with_documents(&ids(&["a/0", "a/1", "a/2", "z/0"]), "{}", |storage, at| {
let source = Arc::new(TestSource::denying(&[("records", "a/1")]));
let service = Service::with_access_source(
storage,
NonZeroU32::new(10).ok_or("zero max page")?,
None,
source,
)?;
for bytes in [None, Some(4096)] {
let full = content_list(&service, &at, Some("a/"), None, 10, Some("full"), bytes)?;
let light = content_list(&service, &at, Some("a/"), None, 10, Some("none"), bytes)?;
assert_eq!(answer_ids(&full)?, ["a/0", "a/2"]);
assert_eq!(answer_ids(&light)?, ["a/0", "a/2"]);
assert_eq!(light, without_bodies(full)?);
}
Ok(())
})
}
#[test]
fn content_none_fits_a_budget_that_refuses_one_full_entry() -> TestResult {
let raw = serde_json::to_string(&json!({"body": "x".repeat(4096)}))?;
with_documents(&ids(&["a/1"]), &raw, |storage, at| {
let service = Service::new(storage, NonZeroU32::MIN)?;
let light = content_list(&service, &at, Some("a/"), None, 1, Some("none"), None)?;
let bytes = serde_json::to_vec(&light)?.len();
let bounded = content_list(
&service,
&at,
Some("a/"),
None,
1,
Some("none"),
Some(bytes),
)?;
assert_eq!(bounded, light);
let failure = content_list(
&service,
&at,
Some("a/"),
None,
1,
Some("full"),
Some(bytes),
)
.err()
.ok_or("full body fitted the metadata budget")?;
assert!(failure.to_string().contains("response_limit"));
assert_eq!(answer_ids(&bounded)?, ["a/1"]);
assert!(bounded["next_after"].is_null());
Ok(())
})
}
#[test]
fn content_none_reads_only_revision_metadata_and_no_document_bodies() -> TestResult {
with_documents(&ids(&["a/1", "a/2", "a/3", "b/1"]), "{}", |storage, at| {
let service = Service::new(storage, NonZeroU32::new(3).ok_or("zero max page")?)?;
for bytes in [None, Some(4096)] {
reset_page_read_counts();
super::take_content_reads();
let full = content_list(&service, &at, Some("a/"), None, 3, Some("full"), bytes)?;
assert_eq!(page_read_counts().ranges, 3);
assert_eq!(page_read_counts().points, 3);
assert_eq!(super::take_content_reads(), 3);
reset_page_read_counts();
let light = content_list(&service, &at, Some("a/"), None, 3, Some("none"), bytes)?;
assert_eq!(light, without_bodies(full)?);
assert_eq!(page_read_counts().ranges, 3);
assert_eq!(page_read_counts().points, 0);
assert_eq!(super::take_content_reads(), 0);
for entry in light["entries"].as_array().ok_or("missing entries")? {
assert!(entry.get("raw_json").is_none());
}
}
Ok(())
})
}
#[test]
fn content_none_keeps_the_prefix_seek_bounds_among_a_thousand_rows() -> TestResult {
let mut keys: Vec<_> = (0..997).map(|index| format!("a/{index:04}")).collect();
keys.extend(ids(&["m/1", "m/2", "m/3"]));
assert_eq!(keys.len(), 1000);
with_documents(&keys, "{}", |storage, at| {
let service = Service::new(storage, NonZeroU32::new(3).ok_or("zero max page")?)?;
let first: &[u8] = b"m\0records\0m/";
let end: &[u8] = b"m\0records\0m0";
for bytes in [None, Some(4096)] {
reset_page_read_counts();
drop(take_document_ranges());
super::take_content_reads();
let light = content_list(&service, &at, Some("m/"), None, 3, Some("none"), bytes)?;
assert_eq!(answer_ids(&light)?, ["m/1", "m/2", "m/3"]);
assert!(light["next_after"].is_null());
assert_eq!(super::take_content_reads(), 0);
let counts = page_read_counts();
assert_eq!(counts.points, 0);
assert!(counts.ranges <= 4);
let ranges = take_document_ranges();
assert_eq!(ranges.len(), counts.ranges);
assert!(!ranges.is_empty());
assert_eq!(ranges[0].0.as_slice(), first);
for (from, to) in ranges {
assert!(from.as_slice() >= first);
assert!(from.as_slice() < end);
assert!(to.as_slice() <= end);
assert_ne!(from.as_slice(), &b"m\0records\0"[..]);
}
}
Ok(())
})
}
#[test]
fn deleted_and_rewritten_documents_match_full_at_each_captured_version() -> TestResult {
with_documents(
&ids(&["a/1", "a/2", "a/3", "b/1"]),
"{}",
|mut storage, before| {
assert_eq!(
pairing_tests::assert_document_pairs(&storage, &before)?,
ids(&["a/1", "a/2", "a/3", "b/1"])
.into_iter()
.collect::<std::collections::BTreeSet<_>>(),
);
let deleted = storage.mutate(&Mutation {
request_id: "delete".to_owned(),
branch: "main".to_owned(),
expected: before.version,
change: Change::Delete(crate::serve::protocol::DocumentDelete {
collection: "records".to_owned(),
document_id: "a/2".to_owned(),
}),
})?;
let after_delete = storage.capture(&Selection {
branch: "main".to_owned(),
version: None,
})?;
assert_eq!(
pairing_tests::assert_document_pairs(&storage, &after_delete)?,
ids(&["a/1", "a/3", "b/1"])
.into_iter()
.collect::<std::collections::BTreeSet<_>>(),
);
storage.mutate(&Mutation {
request_id: "rewrite".to_owned(),
branch: "main".to_owned(),
expected: deleted.version,
change: Change::Put(crate::serve::protocol::DocumentWrite {
collection: "records".to_owned(),
document_id: "a/3".to_owned(),
document: RawValue::from_string(r#"{"rewritten":true}"#.to_owned())?,
}),
})?;
let after = storage.capture(&Selection {
branch: "main".to_owned(),
version: None,
})?;
assert_eq!(
pairing_tests::assert_document_pairs(&storage, &after)?,
ids(&["a/1", "a/3", "b/1"])
.into_iter()
.collect::<std::collections::BTreeSet<_>>(),
);
let service = Service::new(storage, NonZeroU32::new(10).ok_or("zero max page")?)?;
let old = content_list(&service, &before, Some("a/"), None, 10, Some("full"), None)?;
let current = content_list(&service, &after, Some("a/"), None, 10, Some("full"), None)?;
assert_eq!(answer_ids(&old)?, ["a/1", "a/2", "a/3"]);
assert_eq!(answer_ids(¤t)?, ["a/1", "a/3"]);
assert_ne!(
old["entries"][2]["revision"],
current["entries"][1]["revision"]
);
for (at, expected) in [
(&before, vec!["a/1", "a/2", "a/3"]),
(&after, vec!["a/1", "a/3"]),
] {
assert_light_matches_full_at(&service, at, &expected)?;
}
Ok(())
},
)
}
fn assert_light_matches_full_at(service: &Service, at: &Captured, expected: &[&str]) -> TestResult {
for bytes in [None, Some(4096)] {
let mut cursor = None;
for (index, id) in expected.iter().enumerate() {
let full = content_list(
service,
at,
Some("a/"),
cursor.as_deref(),
1,
Some("full"),
bytes,
)?;
let light = content_list(
service,
at,
Some("a/"),
cursor.as_deref(),
1,
Some("none"),
bytes,
)?;
assert_eq!(light, without_bodies(full)?);
assert_eq!(answer_ids(&light)?, [*id]);
assert_eq!(
light["selection"]["version"],
serde_json::to_value(at.version)?
);
assert_eq!(
light["next_after"],
if index + 1 == expected.len() {
Value::Null
} else {
Value::String((*id).to_owned())
}
);
cursor = light["next_after"].as_str().map(str::to_owned);
}
}
Ok(())
}
#[test]
fn content_none_byte_pages_keep_the_same_ids_revisions_and_cursor() -> TestResult {
with_documents(&[], "{}", |mut storage, at| {
let max_page = NonZeroU32::new(10).ok_or("zero max page")?;
let writer = format!("writer-{}", "x".repeat(256));
storage.mutate(&Mutation {
request_id: "writer-fork".to_owned(),
branch: "main".to_owned(),
expected: at.version,
change: Change::Fork(crate::serve::protocol::Fork {
name: writer.clone(),
from: at.version,
}),
})?;
let writer_selection = Selection {
branch: writer.clone(),
version: None,
};
let documents = ["a/1", "a/2", "a/3", "b/1"]
.into_iter()
.map(|id| {
Ok(ImportedDocument {
document_id: id.to_owned(),
document: RawValue::from_string("{}".to_owned())?,
})
})
.collect::<Result<Vec<_>, serde_json::Error>>()?;
let written = storage.mutate_bounded(
&Mutation {
request_id: "long-writer-seed".to_owned(),
branch: writer.clone(),
expected: storage.capture(&writer_selection)?.version,
change: Change::PutMany(DocumentImport {
collection: "records".to_owned(),
documents,
}),
},
max_page,
)?;
storage.mutate(&Mutation {
request_id: "read-fork".to_owned(),
branch: writer,
expected: written.version,
change: Change::Fork(crate::serve::protocol::Fork {
name: "read".to_owned(),
from: written.version,
}),
})?;
let at = storage.capture(&Selection {
branch: "read".to_owned(),
version: None,
})?;
let service = Service::new(storage, max_page)?;
let first = content_list(&service, &at, Some("a/"), None, 1, Some("none"), None)?;
let entry_bytes = serde_json::to_vec(&first["entries"][0])?.len();
let bytes = serde_json::to_vec(&first)?.len();
assert!(2 * entry_bytes + 3 > bytes);
let mut cursor = None;
for (index, id) in ["a/1", "a/2", "a/3"].into_iter().enumerate() {
let full = content_list(
&service,
&at,
Some("a/"),
cursor.as_deref(),
1,
Some("full"),
None,
)?;
let light = content_list(
&service,
&at,
Some("a/"),
cursor.as_deref(),
10,
Some("none"),
Some(bytes),
)?;
assert_eq!(light, without_bodies(full)?);
assert_eq!(answer_ids(&light)?, [id]);
assert_eq!(
light["next_after"],
if index == 2 {
Value::Null
} else {
Value::String(id.to_owned())
}
);
cursor = light["next_after"].as_str().map(str::to_owned);
}
Ok(())
})
}
#[path = "reads_pairing_tests.rs"]
mod pairing_tests;
const fn fixture_long_wait_interval() -> std::time::Duration {
const LONG_WAIT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
LONG_WAIT_INTERVAL
}