haematite-cli 0.21.8

//! Two bounded comparisons settle one common three-version key range.

use std::num::NonZeroUsize;

use haematite::db::{ObservedChange, ReadOnlyDatabase};
use haematite::tree::DiffBudget;
use serde::{Deserialize, Serialize};

use super::super::{Captured, Catalogue, Failure, ForkProvenance, Held, branches, identifier};
use super::{NODE_BUDGET, cursor, movement, owner_changes_page};
use crate::serve::protocol::{ReviewPage, Selection, Version};

pub(in crate::serve::storage) struct Page {
    pub(in crate::serve::storage) selection: Captured,
    pub(in crate::serve::storage) from: ForkProvenance,
    pub(in crate::serve::storage) origin: Captured,
    pub(in crate::serve::storage) parent: Captured,
    pub(in crate::serve::storage) source: Vec<ObservedChange>,
    pub(in crate::serve::storage) target: Vec<ObservedChange>,
    pub(in crate::serve::storage) current_shard: Option<usize>,
    pub(in crate::serve::storage) next_after: Option<String>,
    pub(in crate::serve::storage) nodes_loaded: usize,
    pub(in crate::serve::storage) source_scanned: usize,
    pub(in crate::serve::storage) target_scanned: usize,
    /// The captures this page read, held until the page is dropped (STOR-003
    /// H1b2); empty where the caller holds the versions itself.
    pub(in crate::serve::storage) held: Vec<Held>,
}

#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Cursor {
    review: cursor::Cursor,
    parent: Version,
}

pub(super) struct Roots {
    pub(super) selection: Captured,
    pub(super) from: ForkProvenance,
    pub(super) origin: Captured,
    pub(super) parent: Captured,
    cursor: Option<Cursor>,
    start: usize,
    pub(in crate::serve::storage) held: Vec<Held>,
}

pub(super) fn page(
    reader: &ReadOnlyDatabase,
    catalogue: &Catalogue,
    request: &ReviewPage,
    limit: NonZeroUsize,
) -> Result<Page, Failure> {
    page_with_node_budget(reader, catalogue, request, limit, NODE_BUDGET)
}

pub(super) fn page_both(
    reader: &ReadOnlyDatabase,
    catalogue: &Catalogue,
    request: &ReviewPage,
    limit: NonZeroUsize,
) -> Result<Page, Failure> {
    compare_page(reader, catalogue, request, limit, NODE_BUDGET, true)
}

fn page_with_node_budget(
    reader: &ReadOnlyDatabase,
    catalogue: &Catalogue,
    request: &ReviewPage,
    limit: NonZeroUsize,
    node_budget: usize,
) -> Result<Page, Failure> {
    compare_page(reader, catalogue, request, limit, node_budget, false)
}

fn compare_page(
    reader: &ReadOnlyDatabase,
    catalogue: &Catalogue,
    request: &ReviewPage,
    limit: NonZeroUsize,
    node_budget: usize,
    both_sides: bool,
) -> Result<Page, Failure> {
    let roots = capture(reader, catalogue, request)?;
    compare_roots(reader, roots, limit, node_budget, both_sides)
}

pub(in crate::serve::storage) fn held_page(
    reader: &ReadOnlyDatabase,
    origin: &Captured,
    target: &Captured,
    source: &Captured,
    after: Option<&str>,
    limit: NonZeroUsize,
    node_budget: usize,
) -> Result<Page, Failure> {
    let cursor = after.map(cursor::decode::<Cursor>).transpose()?;
    let from = ForkProvenance {
        branch: origin.branch.clone(),
        version: origin.version,
    };
    if let Some(cursor) = &cursor {
        cursor.review.validate(&source.branch)?;
        if cursor.review.version != source.version
            || cursor.parent != target.version
            || cursor.review.from.branch != from.branch
            || cursor.review.from.version != from.version
        {
            return Err(Failure::new(
                "invalid_continuation",
                "comparison differs from held pins",
            ));
        }
    }
    for at in [origin, target] {
        if !at
            .shards
            .iter()
            .map(|shard| shard.shard_id)
            .eq(source.shards.iter().map(|shard| shard.shard_id))
        {
            return Err(Failure::new(
                "incomplete_roots",
                "held comparison shard sets differ",
            ));
        }
    }
    let start = match &cursor {
        Some(cursor) => source
            .shards
            .iter()
            .position(|shard| shard.shard_id == cursor.review.shard)
            .ok_or_else(|| {
                Failure::new("invalid_continuation", "comparison names an absent shard")
            })?,
        None => 0,
    };
    compare_roots(
        reader,
        Roots {
            selection: source.clone(),
            origin: origin.clone(),
            parent: target.clone(),
            from,
            cursor,
            start,
            // A held page's versions are held by the caller's merge pins.
            held: Vec::new(),
        },
        limit,
        node_budget,
        true,
    )
}

fn compare_roots(
    reader: &ReadOnlyDatabase,
    roots: Roots,
    limit: NonZeroUsize,
    node_budget: usize,
    both_sides: bool,
) -> Result<Page, Failure> {
    let Roots {
        selection,
        from,
        origin,
        parent,
        cursor,
        start,
        held,
    } = roots;
    let budget = DiffBudget {
        max_changes: limit,
        max_nodes: NonZeroUsize::new(node_budget / 2)
            .ok_or_else(|| Failure::new("node_budget", "comparison node budget must be nonzero"))?,
    };
    for (index, shard) in selection.shards.iter().enumerate().skip(start) {
        let before = &origin.shards[index];
        if before.head == shard.head && (!both_sides || before.head == parent.shards[index].head) {
            continue;
        }
        let after = cursor
            .as_ref()
            .filter(|cursor| cursor.review.shard == shard.shard_id)
            .and_then(|cursor| cursor.review.key.as_deref());
        let (mut source, source_scanned) = owner_changes_page(
            reader,
            shard.shard_id,
            before.head,
            shard.head,
            after,
            budget,
        )?;
        let (mut target, target_scanned) = owner_changes_page(
            reader,
            shard.shard_id,
            before.head,
            parent.shards[index].head,
            after,
            budget,
        )?;
        let end = settled_end(source.next_after, target.next_after);
        if end
            .as_deref()
            .is_some_and(|end| after.is_some_and(|after| end <= after))
        {
            return Err(Failure::new(
                "review_no_progress",
                "comparison did not settle a new key range",
            ));
        }
        if let Some(end) = end.as_deref() {
            source.entries.retain(|change| change.key() <= end);
            target.entries.retain(|change| change.key() <= end);
        }
        let next = match end {
            Some(key) => Some((shard.shard_id, Some(key))),
            None => selection
                .shards
                .iter()
                .find(|next| next.shard_id > shard.shard_id)
                .map(|next| (next.shard_id, None)),
        };
        let next_after = continuation(&selection, &from, parent.version, next)?;
        let current_shard = Some(shard.shard_id);
        return Ok(Page {
            selection,
            from,
            origin,
            parent,
            source: source.entries,
            target: target.entries,
            current_shard,
            next_after,
            nodes_loaded: source.nodes_loaded + target.nodes_loaded,
            source_scanned,
            target_scanned,
            held,
        });
    }
    Ok(Page {
        selection,
        from,
        origin,
        parent,
        source: Vec::new(),
        target: Vec::new(),
        current_shard: None,
        next_after: None,
        nodes_loaded: 0,
        source_scanned: 0,
        target_scanned: 0,
        held,
    })
}

pub(super) fn capture(
    reader: &ReadOnlyDatabase,
    catalogue: &Catalogue,
    request: &ReviewPage,
) -> Result<Roots, Failure> {
    identifier(&request.branch)?;
    let cursor = request
        .after
        .as_deref()
        .map(|text| {
            let cursor: Cursor = cursor::decode(text)?;
            cursor.review.validate(&request.branch)?;
            Ok::<_, Failure>(cursor)
        })
        .transpose()?;
    let mut selection = catalogue.capture_held(&Selection {
        branch: request.branch.clone(),
        version: cursor.as_ref().map(|cursor| cursor.review.version),
    })?;
    selection
        .shards
        .sort_unstable_by_key(|shard| shard.shard_id);
    let from = branches::read_provenance(reader, &selection)?;
    if cursor.as_ref().is_some_and(|cursor| {
        cursor.review.from.branch != from.branch || cursor.review.from.version != from.version
    }) {
        return Err(Failure::new(
            "invalid_continuation",
            "comparison origin differs from captured provenance",
        ));
    }
    let current_parent = catalogue.capture_held(&Selection {
        branch: from.branch.clone(),
        version: None,
    })?;
    movement(&from, &current_parent)?;
    let mut parent = match cursor.as_ref() {
        Some(cursor) => catalogue.capture_held(&Selection {
            branch: from.branch.clone(),
            version: Some(cursor.parent),
        })?,
        None => current_parent,
    };
    movement(&from, &parent)?;
    parent.shards.sort_unstable_by_key(|shard| shard.shard_id);
    let mut origin = catalogue.capture_held(&Selection {
        branch: from.branch.clone(),
        version: Some(from.version),
    })?;
    origin.shards.sort_unstable_by_key(|shard| shard.shard_id);
    for roots in [&origin, &parent] {
        if roots.shards.len() != selection.shards.len()
            || roots
                .shards
                .iter()
                .zip(&selection.shards)
                .any(|(a, b)| a.shard_id != b.shard_id)
        {
            return Err(Failure::new(
                "incomplete_roots",
                "comparison versions have different shard sets",
            ));
        }
    }
    let start = match cursor.as_ref() {
        Some(cursor) => selection
            .shards
            .iter()
            .position(|shard| shard.shard_id == cursor.review.shard)
            .ok_or_else(|| {
                Failure::new(
                    "invalid_continuation",
                    "comparison cursor names an absent shard",
                )
            })?,
        None => 0,
    };
    Ok(Roots {
        selection: Captured::clone(&selection),
        from,
        origin: Captured::clone(&origin),
        parent: Captured::clone(&parent),
        cursor,
        start,
        held: vec![selection, origin, parent],
    })
}

pub(super) fn first_cursor(page: &Page) -> Result<String, Failure> {
    let shard = page
        .selection
        .shards
        .first()
        .ok_or_else(|| Failure::new("incomplete_roots", "comparison does not hold any shard"))?
        .shard_id;
    continuation(
        &page.selection,
        &page.from,
        page.parent.version,
        Some((shard, None)),
    )?
    .ok_or_else(|| Failure::new("invalid_continuation", "comparison has no initial cursor"))
}

pub(in crate::serve::storage) fn after_key(page: &Page, key: &[u8]) -> Result<String, Failure> {
    let shard = page
        .current_shard
        .ok_or_else(|| Failure::new("invalid_continuation", "comparison did not settle a shard"))?;
    if ![&page.source, &page.target].iter().any(|changes| {
        changes
            .binary_search_by(|change| change.key().cmp(key))
            .is_ok()
    }) {
        return Err(Failure::new(
            "invalid_continuation",
            "selected key is outside the settled comparison page",
        ));
    }
    continuation(
        &page.selection,
        &page.from,
        page.parent.version,
        Some((shard, Some(key.to_vec()))),
    )?
    .ok_or_else(|| Failure::new("invalid_continuation", "comparison has no selected cursor"))
}

fn settled_end(source: Option<Vec<u8>>, target: Option<Vec<u8>>) -> Option<Vec<u8>> {
    match (source, target) {
        (Some(source), Some(target)) => Some(source.min(target)),
        (Some(key), None) | (None, Some(key)) => Some(key),
        (None, None) => None,
    }
}

fn continuation(
    selection: &Captured,
    from: &ForkProvenance,
    parent: Version,
    next: Option<(usize, Option<Vec<u8>>)>,
) -> Result<Option<String>, Failure> {
    next.map(|(shard, key)| {
        cursor::encode(&Cursor {
            review: cursor::Cursor {
                branch: selection.branch.clone(),
                version: selection.version,
                from: ForkProvenance {
                    branch: from.branch.clone(),
                    version: from.version,
                },
                shard,
                key,
            },
            parent,
        })
    })
    .transpose()
}

pub(super) fn sides(change: &ObservedChange) -> (Option<&[u8]>, Option<&[u8]>) {
    match change {
        ObservedChange::Added { value, .. } => (None, Some(value)),
        ObservedChange::Removed { value, .. } => (Some(value), None),
        ObservedChange::Modified {
            old_value,
            new_value,
            ..
        } => (Some(old_value), Some(new_value)),
    }
}

pub(super) fn target_value<'a>(
    change: &'a ObservedChange,
    target: &'a [ObservedChange],
) -> Result<Option<&'a [u8]>, Failure> {
    let (ancestor, _) = sides(change);
    match target.binary_search_by(|entry| entry.key().cmp(change.key())) {
        Ok(index) => {
            let (target_ancestor, value) = sides(&target[index]);
            if ancestor != target_ancestor {
                return Err(Failure::new(
                    "inconsistent_diff",
                    "comparisons disagree on the captured origin value",
                ));
            }
            Ok(value)
        }
        Err(_) => Ok(ancestor),
    }
}

#[cfg(test)]
#[path = "review_compare_tests.rs"]
mod tests;