use std::collections::BTreeSet;
use super::{
Captured, Deserialize, Failure, HeldInputs, MergeChoice, MergeChoiceRow, MergeOwner, Node,
NonZeroUsize, Serialize, Storage, Triple, cursor, edges, identifier, validate_token,
};
use crate::serve::protocol::Edge;
use crate::serve::storage::{
reads,
review::{NODE_BUDGET, choice, compare, dangling},
};
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MissingChoice {
pub owner: MergeOwner,
pub token: String,
}
#[derive(Debug, Serialize)]
pub struct MergeCoverage {
pub missing: Vec<MissingChoice>,
pub next_after: Option<String>,
pub end: bool,
}
#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub(in crate::serve::storage) struct Position {
merge_id: String,
revision: u64,
comparison: Option<String>,
diff_done: bool,
incident: Option<Incident>,
pending: Vec<MissingChoice>,
}
#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Incident {
node: Node,
incoming: bool,
after: Option<String>,
}
pub(in crate::serve::storage) struct Step {
pub(in crate::serve::storage) missing: Vec<MissingChoice>,
pub(in crate::serve::storage) next: Option<Position>,
pub(in crate::serve::storage) keys: Vec<Vec<u8>>,
}
pub(super) const PAGE_OWNER_LIMIT: usize = 128;
struct Work {
missing: Vec<MissingChoice>,
keys: Vec<Vec<u8>>,
source_left: usize,
target_left: usize,
nodes_left: usize,
incident_left: usize,
}
impl Work {
const fn new() -> Self {
Self {
missing: Vec::new(),
keys: Vec::new(),
source_left: dangling::SCAN_ALLOWANCE,
target_left: dangling::SCAN_ALLOWANCE,
nodes_left: NODE_BUDGET,
incident_left: dangling::SCAN_ALLOWANCE,
}
}
const fn full(&self, limit: NonZeroUsize) -> bool {
self.missing.len() == limit.get() || self.keys.len() == PAGE_OWNER_LIMIT
}
const fn can_compare(&self) -> bool {
self.source_left > 0 && self.target_left > 0 && self.nodes_left >= 2
}
}
struct Context<'a> {
storage: &'a Storage,
inputs: HeldInputs,
service: Captured,
merge_id: &'a str,
}
struct IncidentOwner {
key: Vec<u8>,
missing: Vec<MissingChoice>,
}
struct Inspection {
missing: Vec<MissingChoice>,
incident: Option<Node>,
}
impl Storage {
pub fn merge_coverage(
&self,
merge_id: &str,
after: Option<&str>,
limit: NonZeroUsize,
) -> Result<MergeCoverage, Failure> {
let position = after.map(cursor::decode::<Position>).transpose()?;
let step = self.coverage_step(merge_id, position, limit)?;
let end = step.next.is_none();
let next_after = step.next.as_ref().map(cursor::encode).transpose()?;
Ok(MergeCoverage {
missing: step.missing,
next_after,
end,
})
}
pub(in crate::serve::storage) fn coverage_step(
&self,
merge_id: &str,
position: Option<Position>,
limit: NonZeroUsize,
) -> Result<Step, Failure> {
let plan = self.open_merge_plan(merge_id)?;
let inputs = self.choice_inputs(&plan)?;
let service = self.merge_service_selection()?;
let revision = self.choice_revision(&service, merge_id)?;
let position = match position {
Some(position) => {
validate_position(&position, merge_id, revision)?;
position
}
None => Position {
merge_id: merge_id.to_owned(),
revision,
comparison: None,
diff_done: false,
incident: None,
pending: Vec::new(),
},
};
walk(
&Context {
storage: self,
inputs,
service,
merge_id,
},
position,
limit,
)
}
}
fn validate_position(position: &Position, id: &str, revision: u64) -> Result<(), Failure> {
if position.merge_id != id || position.pending.len() > 3 {
return Err(Failure::new(
"invalid_continuation",
"coverage position belongs to another walk",
));
}
if position.revision != revision {
return Err(Failure::new(
"choice_journal_moved",
"journal changed during coverage listing",
));
}
for missing in &position.pending {
missing.owner.key()?;
validate_token(&missing.token)?;
}
if let Some(incident) = &position.incident {
identifier(&incident.node.collection)?;
identifier(&incident.node.document_id)?;
if incident
.after
.as_ref()
.is_some_and(|after| after.len() > 5 * 513)
{
return Err(Failure::new(
"invalid_continuation",
"incident cursor exceeds its key bound",
));
}
}
Ok(())
}
fn walk(
context: &Context<'_>,
mut position: Position,
limit: NonZeroUsize,
) -> Result<Step, Failure> {
let mut work = Work::new();
if !position.pending.is_empty() {
let take = limit.get().min(position.pending.len());
work.missing.extend(position.pending.drain(..take));
}
drain_incident(context, &mut position, &mut work, limit)?;
if !work.full(limit)
&& position.pending.is_empty()
&& position.incident.is_none()
&& !position.diff_done
{
difference_step(context, &mut position, &mut work, limit)?;
}
let ended = position.diff_done && position.incident.is_none() && position.pending.is_empty();
Ok(Step {
missing: work.missing,
next: if ended { None } else { Some(position) },
keys: work.keys,
})
}
fn drain_incident(
context: &Context<'_>,
position: &mut Position,
work: &mut Work,
limit: NonZeroUsize,
) -> Result<(), Failure> {
while position.incident.is_some()
&& position.pending.is_empty()
&& !work.full(limit)
&& work.incident_left > 0
{
incident_step(context, position, work, limit)?;
}
Ok(())
}
fn comparison_page(
context: &Context<'_>,
position: &Position,
work: &mut Work,
) -> Result<Option<compare::Page>, Failure> {
let scan = NonZeroUsize::new(work.source_left.min(work.target_left))
.ok_or_else(|| Failure::new("coverage_budget", "raw scan allowance is exhausted"))?;
let page = match compare::held_page(
&context.storage.reader,
&context.inputs.origin,
&context.inputs.target,
&context.inputs.source,
position.comparison.as_deref(),
scan,
work.nodes_left,
) {
Ok(page) => page,
Err(error) if error.code == "node_budget_too_small" && work.nodes_left < NODE_BUDGET => {
return Ok(None);
}
Err(error) => return Err(error),
};
work.source_left = work
.source_left
.checked_sub(page.source_scanned)
.ok_or_else(|| Failure::new("coverage_budget", "source scan exceeded its allowance"))?;
work.target_left = work
.target_left
.checked_sub(page.target_scanned)
.ok_or_else(|| Failure::new("coverage_budget", "target scan exceeded its allowance"))?;
work.nodes_left = work
.nodes_left
.checked_sub(page.nodes_loaded)
.ok_or_else(|| Failure::new("coverage_budget", "comparison exceeded its node allowance"))?;
Ok(Some(page))
}
fn difference_step(
context: &Context<'_>,
position: &mut Position,
work: &mut Work,
limit: NonZeroUsize,
) -> Result<(), Failure> {
while work.can_compare() && !work.full(limit) {
let Some(page) = comparison_page(context, position, work)? else {
return Ok(());
};
let keys: BTreeSet<_> = page
.source
.iter()
.chain(&page.target)
.map(haematite::ObservedChange::key)
.filter(|key| !key.starts_with(b"k\0"))
.collect();
for key in keys {
if let Some(owner) = owner_key(key)? {
let inspected = inspect(context, owner)?;
work.keys.push(key.to_vec());
append_missing(position, &mut work.missing, inspected.missing, limit);
if let Some(node) = inspected.incident {
position.incident = Some(Incident {
node,
incoming: false,
after: None,
});
}
position.comparison = Some(compare::after_key(&page, key)?);
drain_incident(context, position, work, limit)?;
} else {
unsupported_conflict(context, key)?;
if key.starts_with(b"c\0")
|| key.starts_with(b"n\0")
|| key.starts_with(b"v\0")
|| key.starts_with(b"w\0")
{
work.keys.push(key.to_vec());
}
position.comparison = Some(compare::after_key(&page, key)?);
}
if work.full(limit) || position.incident.is_some() || !position.pending.is_empty() {
return Ok(());
}
}
position.comparison = page.next_after;
if position.comparison.is_none() {
position.diff_done = true;
return Ok(());
}
}
Ok(())
}
fn append_missing(
position: &mut Position,
missing: &mut Vec<MissingChoice>,
found: Vec<MissingChoice>,
limit: NonZeroUsize,
) {
for owner in found {
if missing.len() < limit.get() {
missing.push(owner);
} else {
position.pending.push(owner);
}
}
}
fn incident_step(
context: &Context<'_>,
position: &mut Position,
work: &mut Work,
limit: NonZeroUsize,
) -> Result<(), Failure> {
let incident = position
.incident
.as_ref()
.ok_or_else(|| Failure::new("invalid_continuation", "incident work disappeared"))?;
let page = incident_page(context, incident, work, limit)?;
let incoming = incident.incoming;
let scanned = page.entries.len();
let charged = if scanned == 0 && page.next_after.is_some() {
1
} else {
scanned
};
work.incident_left = work
.incident_left
.checked_sub(charged)
.ok_or_else(|| Failure::new("coverage_budget", "incident scan exceeded its allowance"))?;
for edge in page.entries {
let last = incident_suffix(&edge, incoming);
if incoming && edge.from == edge.to {
continue;
}
let Some(owner) = inspect_incident_edge(context, edge)? else {
continue;
};
work.keys.push(owner.key);
append_missing(position, &mut work.missing, owner.missing, limit);
if work.full(limit) || !position.pending.is_empty() {
let incident = position
.incident
.as_mut()
.ok_or_else(|| Failure::new("invalid_continuation", "incident work disappeared"))?;
incident.after = Some(last);
return Ok(());
}
}
if let Some(after) = page.next_after {
let incident = position
.incident
.as_mut()
.ok_or_else(|| Failure::new("invalid_continuation", "incident work disappeared"))?;
incident.after = Some(after);
} else if incoming {
position.incident = None;
} else {
let incident = position
.incident
.as_mut()
.ok_or_else(|| Failure::new("invalid_continuation", "incident work disappeared"))?;
incident.incoming = true;
incident.after = None;
}
Ok(())
}
fn incident_page(
context: &Context<'_>,
incident: &Incident,
work: &Work,
limit: NonZeroUsize,
) -> Result<reads::Page<Edge>, Failure> {
let remaining = PAGE_OWNER_LIMIT
.checked_sub(work.keys.len())
.ok_or_else(|| Failure::new("coverage_budget", "owner batch exceeded its allowance"))?;
let rows = NonZeroUsize::new(limit.get().min(remaining).min(work.incident_left))
.ok_or_else(|| Failure::new("coverage_budget", "incident allowance is zero"))?;
let page = if incident.incoming {
edges::incoming(
&context.storage.reader,
&context.inputs.target,
&incident.node,
None,
incident.after.as_deref(),
rows,
None,
)?
} else {
edges::out(
&context.storage.reader,
&context.inputs.target,
&incident.node,
None,
incident.after.as_deref(),
rows,
None,
)?
};
Ok(page)
}
fn inspect_incident_edge(
context: &Context<'_>,
edge: Edge,
) -> Result<Option<IncidentOwner>, Failure> {
let owner = MergeOwner::Edge {
relation: edge.relation,
from: edge.from,
to: edge.to,
};
let key = owner.key()?;
let values = context.storage.choice_values(&context.inputs, &key)?;
let target = values.target.as_deref().ok_or_else(|| {
Failure::new(
"edge_index_inconsistent",
"incident index names an absent owner",
)
})?;
let properties: serde_json::Value = serde_json::from_slice(target).map_err(Failure::storage)?;
if properties != edge.properties {
return Err(Failure::new(
"edge_index_inconsistent",
"incident and owner properties differ",
));
}
if !values.unchanged() {
return Ok(None);
}
Ok(Some(IncidentOwner {
key,
missing: inspect(context, owner)?.missing,
}))
}
fn incident_suffix(edge: &Edge, incoming: bool) -> String {
let other = if incoming { &edge.from } else { &edge.to };
format!(
"{}\0{}\0{}",
edge.relation, other.collection, other.document_id
)
}
fn inspect(context: &Context<'_>, owner: MergeOwner) -> Result<Inspection, Failure> {
let key = owner.key()?;
let values = context.storage.choice_values(&context.inputs, &key)?;
let token = values.token(&context.inputs.pins, &key)?;
let row = context
.storage
.read_choice_row(&context.service, context.merge_id, &token)?;
if row.as_ref().is_some_and(|row| row.owner != owner) {
return Err(Failure::new(
"choice_row_corrupt",
"coverage row names another owner",
));
}
let conflict = values.ancestor != values.target
&& values.ancestor != values.source
&& values.target != values.source;
let mut missing = Vec::new();
let incident = match &owner {
MergeOwner::Document {
collection,
document_id,
} => {
let result = document_presence(&values, row.as_ref())?;
if result.is_none_or(|present| present != values.ancestor.is_some()) {
Some(Node {
collection: collection.clone(),
document_id: document_id.clone(),
})
} else {
None
}
}
MergeOwner::Edge { from, to, .. } => {
if edge_settled_absent(&values, row.as_ref()) {
return Ok(Inspection {
missing,
incident: None,
});
}
let from = endpoint(context, from, &mut missing)?;
let to = endpoint(context, to, &mut missing)?;
if from == Some(false) || to == Some(false) {
if row
.as_ref()
.is_some_and(|row| row.choice != MergeChoice::RemoveEdge)
{
return Err(Failure::new(
"choice_row_corrupt",
"settled dangling edge has a keeping choice",
));
}
if row.is_none() {
missing.push(MissingChoice {
owner: owner.clone(),
token: token.clone(),
});
}
} else if row
.as_ref()
.is_some_and(|row| row.choice == MergeChoice::RemoveEdge)
&& from == Some(true)
&& to == Some(true)
{
return Err(Failure::new(
"choice_row_corrupt",
"edge removal has no absent endpoint",
));
}
None
}
};
if conflict && row.is_none() && !missing.iter().any(|missing| missing.token == token) {
missing.push(MissingChoice { owner, token });
}
Ok(Inspection { missing, incident })
}
fn edge_settled_absent(values: &Triple, row: Option<&MergeChoiceRow>) -> bool {
match row.map(|row| row.choice) {
Some(MergeChoice::TakeSource) => values.source.is_none(),
Some(MergeChoice::TakeTarget) => values.target.is_none(),
None if values.ancestor == values.target => values.source.is_none(),
None if values.ancestor == values.source || values.target == values.source => {
values.target.is_none()
}
Some(MergeChoice::RemoveEdge) | None => false,
}
}
fn endpoint(
context: &Context<'_>,
node: &Node,
missing: &mut Vec<MissingChoice>,
) -> Result<Option<bool>, Failure> {
let owner = MergeOwner::Document {
collection: node.collection.clone(),
document_id: node.document_id.clone(),
};
let key = owner.key()?;
let values = context.storage.choice_values(&context.inputs, &key)?;
let token = values.token(&context.inputs.pins, &key)?;
let row = context
.storage
.read_choice_row(&context.service, context.merge_id, &token)?;
if row.as_ref().is_some_and(|row| row.owner != owner) {
return Err(Failure::new(
"choice_row_corrupt",
"endpoint row names another owner",
));
}
let presence = document_presence(&values, row.as_ref())?;
if presence.is_none() && !missing.iter().any(|missing| missing.token == token) {
missing.push(MissingChoice { owner, token });
}
Ok(presence)
}
fn document_presence(
values: &Triple,
row: Option<&MergeChoiceRow>,
) -> Result<Option<bool>, Failure> {
if let Some(row) = row {
return match row.choice {
MergeChoice::TakeSource => Ok(Some(values.source.is_some())),
MergeChoice::TakeTarget => Ok(Some(values.target.is_some())),
MergeChoice::RemoveEdge => Err(Failure::new(
"choice_row_corrupt",
"document row removes an edge",
)),
};
}
if values.target.is_some() == values.source.is_some() {
Ok(Some(values.target.is_some()))
} else if values.ancestor.is_none() {
Ok(Some(true))
} else {
Ok(None)
}
}
fn unsupported_conflict(context: &Context<'_>, key: &[u8]) -> Result<(), Failure> {
if key.starts_with(b"w\0") {
let values = context.storage.choice_values(&context.inputs, key)?;
choice::embedded_field(
key,
values.ancestor.as_deref(),
values.target.as_deref(),
values.source.as_deref(),
)?;
return Ok(());
}
let parts_count = if key.starts_with(b"c\0") || key.starts_with(b"n\0") {
2
} else if key.starts_with(b"v\0") {
4
} else {
return Ok(());
};
let text = std::str::from_utf8(key).map_err(Failure::storage)?;
let parts: Vec<_> = text.split('\0').collect();
if parts.len() != parts_count {
return Err(Failure::new(
"corrupt_key",
"unsupported owner key has invalid components",
));
}
for part in &parts[1..] {
identifier(part)?;
}
let values = context.storage.choice_values(&context.inputs, key)?;
if values.ancestor != values.target
&& values.ancestor != values.source
&& values.target != values.source
{
return Err(Failure::new(
"choice_owner_unsupported",
format!(
"family {} owner {} has conflicting values and no choice kind",
parts[0],
parts[1..].join("/"),
),
));
}
Ok(())
}
pub(super) fn owner_key(key: &[u8]) -> Result<Option<MergeOwner>, Failure> {
if key.starts_with(b"d\0") {
let text = std::str::from_utf8(key).map_err(Failure::storage)?;
let parts: Vec<_> = text.split('\0').collect();
let ["d", collection, document_id] = parts.as_slice() else {
return Err(Failure::new(
"corrupt_key",
"document key has invalid components",
));
};
let owner = MergeOwner::Document {
collection: (*collection).to_owned(),
document_id: (*document_id).to_owned(),
};
owner.key()?;
Ok(Some(owner))
} else if key.starts_with(b"e\0") {
let text = std::str::from_utf8(key).map_err(Failure::storage)?;
let edge = edges::decode_edge(b"", text, b"{}")?;
Ok(Some(MergeOwner::Edge {
relation: edge.relation,
from: edge.from,
to: edge.to,
}))
} else {
Ok(None)
}
}