use std::collections::BTreeSet;
use std::num::NonZeroUsize;
use haematite::db::ReadOnlyDatabase;
use serde_json::value::RawValue;
use super::{TestResult, ids, with_documents};
use crate::serve::protocol::{Change, DocumentDelete, DocumentWrite, Fork, Mutation, Selection};
use crate::serve::storage::{Captured, MergeChoice, MergeChoose, MergeOwner, MergePin, Storage};
type PairResult<T> = Result<T, Box<dyn std::error::Error>>;
fn stored_ids(
reader: &ReadOnlyDatabase,
at: &Captured,
family: &str,
) -> PairResult<BTreeSet<String>> {
let prefix = format!("{family}\0records\0").into_bytes();
let end = super::super::prefix_end(&prefix)?;
let mut ids = BTreeSet::new();
for shard in &at.shards {
let page = reader.range_page_at(
shard.shard_id,
shard.head,
&prefix,
&end,
NonZeroUsize::new(64).ok_or("zero fixture row limit")?,
)?;
if page.next_from.is_some() {
return Err("pairing fixture exceeded the raw-key bound".into());
}
for (key, _) in page.entries {
let suffix = key
.strip_prefix(prefix.as_slice())
.ok_or("key outside raw family")?;
let id = std::str::from_utf8(suffix)?.to_owned();
if !ids.insert(id) {
return Err("duplicate document id across raw shards".into());
}
}
}
Ok(ids)
}
pub(super) fn assert_document_pairs(
storage: &Storage,
at: &Captured,
) -> PairResult<BTreeSet<String>> {
let reader = storage.reader()?;
let documents = stored_ids(&reader, at, "d")?;
let revisions = stored_ids(&reader, at, "m")?;
assert_eq!(
documents, revisions,
"unpaired document and revision keys at {:?}",
at.version
);
Ok(documents)
}
fn capture(storage: &Storage, branch: &str) -> PairResult<Captured> {
Ok(storage.capture(&Selection {
branch: branch.to_owned(),
version: None,
})?)
}
fn write(storage: &mut Storage, branch: &str, id: &str, change: Change) -> TestResult {
let expected = capture(storage, branch)?.version;
storage.mutate(&Mutation {
request_id: id.to_owned(),
branch: branch.to_owned(),
expected,
change,
})?;
Ok(())
}
fn put(value: u32) -> PairResult<Change> {
Ok(Change::Put(DocumentWrite {
collection: "records".to_owned(),
document_id: "a/1".to_owned(),
document: RawValue::from_string(format!(r#"{{"value":{value}}}"#))?,
}))
}
fn choose(storage: &mut Storage, id: &str, choice: MergeChoice) -> TestResult {
let owner = MergeOwner::Document {
collection: "records".to_owned(),
document_id: "a/1".to_owned(),
};
let mut after = None;
let missing = loop {
let page = storage.merge_coverage(
id,
after.as_deref(),
NonZeroUsize::new(64).ok_or("zero coverage row limit")?,
)?;
if let Some(missing) = page.missing.into_iter().find(|row| row.owner == owner) {
break missing;
}
if page.end {
return Err("fixture document conflict was not offered".into());
}
let next = page.next_after.ok_or("coverage continuation was absent")?;
if after.as_deref() == Some(next.as_str()) {
return Err("coverage did not advance".into());
}
after = Some(next);
};
storage.merge_choose(&MergeChoose {
merge_id: id.to_owned(),
owner: missing.owner,
token: missing.token,
choice,
})?;
Ok(())
}
fn candidate(storage: &Storage, id: &str) -> PairResult<Captured> {
let name = crate::serve::storage::merge_holders::holder_name(id, "candidate");
let record = storage
.refs
.get(&name)
.ok_or("candidate reference was absent")?;
Ok(Captured {
branch: name,
version: crate::serve::storage::version(record),
shards: record.shards.clone(),
})
}
fn merge_pairing(choice: MergeChoice, source_deleted: bool) -> TestResult {
with_documents(&ids(&["a/1", "a/2"]), "{}", |mut storage, origin| {
drop(assert_document_pairs(&storage, &origin)?);
write(
&mut storage,
"main",
"fork",
Change::Fork(Fork {
name: "feature".to_owned(),
from: origin.version,
}),
)?;
let source_change = if source_deleted {
Change::Delete(DocumentDelete {
collection: "records".to_owned(),
document_id: "a/1".to_owned(),
})
} else {
put(1)?
};
write(&mut storage, "feature", "source-change", source_change)?;
write(&mut storage, "main", "target-change", put(2)?)?;
let source = capture(&storage, "feature")?;
let target = capture(&storage, "main")?;
drop(assert_document_pairs(&storage, &source)?);
drop(assert_document_pairs(&storage, &target)?);
storage.merge_open(
"paired",
&MergePin {
branch: source.branch.clone(),
version: source.version,
},
&MergePin {
branch: target.branch.clone(),
version: target.version,
},
)?;
choose(&mut storage, "paired", choice)?;
let mut previous_page = None;
let merged = loop {
let page = storage.merge_candidate_page("paired")?;
let at = candidate(&storage, "paired")?;
drop(assert_document_pairs(&storage, &at)?);
if page.end {
break at;
}
if previous_page.is_some_and(|previous| page.page <= previous) {
return Err("candidate did not advance".into());
}
previous_page = Some(page.page);
};
let selected = match choice {
MergeChoice::TakeSource => &source,
MergeChoice::TakeTarget => &target,
MergeChoice::RemoveEdge => return Err("document cannot select an edge removal".into()),
};
let reader = storage.reader()?;
for key in [
crate::serve::storage::document_key("records", "a/1"),
crate::serve::storage::revision_key("records", "a/1"),
] {
assert_eq!(
crate::serve::storage::reads::get_bytes(&reader, &merged, &key)?,
crate::serve::storage::reads::get_bytes(&reader, selected, &key)?,
);
}
let expected = if source_deleted && choice == MergeChoice::TakeSource {
ids(&["a/2"])
} else {
ids(&["a/1", "a/2"])
}
.into_iter()
.collect::<BTreeSet<_>>();
assert_eq!(assert_document_pairs(&storage, &merged)?, expected);
Ok(())
})
}
#[test]
fn merge_source_choice_keeps_document_revision_pairs() -> TestResult {
merge_pairing(MergeChoice::TakeSource, false)
}
#[test]
fn merge_target_choice_keeps_document_revision_pairs() -> TestResult {
merge_pairing(MergeChoice::TakeTarget, false)
}
#[test]
fn merge_delete_winning_removes_both_raw_keys() -> TestResult {
merge_pairing(MergeChoice::TakeSource, true)
}