haematite_cli/serve/storage/

merge_coverage.rs

//! Affected owners are differences and incident edges at changed endpoint presence.
//! Unchanged orphan edges belong to the full candidate graph validation.

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 {
    /// List uncovered affected owners; caller-held pages cannot authorize publication.
    ///
    /// # Errors
    /// Refuses invalid positions, moved inputs or journal revisions and corrupt rows.
    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,
        })
    }

    /// Drive a server-owned position from the start; no caller page is a certificate.
    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",
        ));
    }
    // Changed owners are handled by the difference scan, so this phase skips them.
    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)
    }
}