haematite-cli 0.21.8

//! One bounded diff page at captured fork and branch roots.

use std::num::NonZeroUsize;

use haematite::db::{ObservedChange, ObserverError, ReadOnlyDatabase};
use haematite::tree::{DiffBudget, DiffPage, Hash, diff::Error as DiffError};

use super::{Captured, Catalogue, Failure, ForkProvenance, Held, branches, identifier};
use crate::serve::protocol::{ReviewPage, Selection};

#[path = "review_choice.rs"]
pub(super) mod choice;
#[path = "review_compare.rs"]
pub(super) mod compare;
#[path = "review_counts.rs"]
pub(in crate::serve) mod counts;
#[path = "review_cursor.rs"]
pub(super) mod cursor;
#[path = "review_dangling.rs"]
pub(in crate::serve) mod dangling;
#[path = "review_prepare.rs"]
pub(in crate::serve) mod prepare;
#[path = "review_values.rs"]
mod values;

use cursor::Cursor;
use values::DiffLine;

pub(super) const NODE_BUDGET: usize = 65_536;

pub(in crate::serve) struct Page {
    pub(in crate::serve) selection: Captured,
    pub(in crate::serve) from: ForkProvenance,
    pub(in crate::serve) diff: Vec<DiffLine>,
    pub(in crate::serve) next_after: Option<String>,
    pub(in crate::serve) base_moved: Option<u64>,
}

/// The branch and fork versions a page reads, each held until the page is
/// built (STOR-003 H1b2).
struct ReviewRoots {
    selection: Held,
    from: ForkProvenance,
    parent: Held,
    cursor: Option<Cursor>,
    start: usize,
    base_moved: Option<u64>,
}

pub(in crate::serve) fn page(
    reader: &ReadOnlyDatabase,
    catalogue: &Catalogue,
    request: &ReviewPage,
    limit: NonZeroUsize,
) -> Result<Page, Failure> {
    let ReviewRoots {
        selection,
        from,
        parent,
        cursor,
        start,
        base_moved,
    } = capture(reader, catalogue, request)?;
    let budget = DiffBudget {
        max_changes: limit,
        max_nodes: NonZeroUsize::new(NODE_BUDGET)
            .ok_or_else(|| Failure::new("node_budget", "review node budget must be nonzero"))?,
    };
    for (shard, before) in selection.shards.iter().zip(&parent.shards).skip(start) {
        if before.head == shard.head {
            continue;
        }
        let after = cursor
            .as_ref()
            .filter(|cursor| cursor.shard == shard.shard_id)
            .and_then(|cursor| cursor.key.as_deref());
        let (result, _) = owner_changes_page(
            reader,
            shard.shard_id,
            before.head,
            shard.head,
            after,
            budget,
        )?;
        let diff = result
            .entries
            .into_iter()
            .map(|change| values::line(reader, &parent, &selection, change))
            .filter_map(Result::transpose)
            .collect::<Result<_, _>>()?;
        let next = match result.next_after {
            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, next)?;
        return Ok(Page {
            selection: selection.into_captured(),
            from,
            diff,
            next_after,
            base_moved,
        });
    }
    Ok(Page {
        selection: selection.into_captured(),
        from,
        diff: Vec::new(),
        next_after: None,
        base_moved,
    })
}

/// One shard's changes after `after`, at most `budget.max_changes` of them
/// (JT2). The engine stops the moment it holds its budget, so it is asked for
/// one more: a shard holding exactly the budget then finishes with no
/// `next_after`, and the review says it is done; one that gives the change
/// past the budget is cut back to the budget and goes on from the last key
/// it keeps. The count is of the changes scanned, metadata included, before
/// the owner's metadata is let go.
pub(super) fn owner_changes_page(
    reader: &ReadOnlyDatabase,
    shard: usize,
    before: Hash,
    after_root: Hash,
    after: Option<&[u8]>,
    budget: DiffBudget,
) -> Result<(DiffPage<ObservedChange>, usize), Failure> {
    let limit = budget.max_changes;
    let wider = DiffBudget {
        max_changes: limit.checked_add(1).ok_or_else(|| {
            Failure::new(
                "review_page_limit_overflow",
                "the page limit leaves no room to look one change past it",
            )
        })?,
        max_nodes: budget.max_nodes,
    };
    let page = reader
        .changes_page(shard, before, after_root, after, wider)
        .map_err(diff_failure)?;
    Ok(owner_changes(cut_to(page, limit)))
}

/// `page` with at most `limit` changes: the change past it, when the engine
/// gave one, is let go, and the page goes on from the last change it keeps.
fn cut_to(mut page: DiffPage<ObservedChange>, limit: NonZeroUsize) -> DiffPage<ObservedChange> {
    if page.entries.len() > limit.get() {
        page.entries.truncate(limit.get());
        page.next_after = page.entries.last().map(|change| change.key().to_vec());
    }
    page
}

fn owner_changes(mut page: DiffPage<ObservedChange>) -> (DiffPage<ObservedChange>, usize) {
    let scanned = page.entries.len();
    page.entries
        .retain(|change| !change.key().starts_with(b"k\0"));
    (page, scanned)
}

fn capture(
    reader: &ReadOnlyDatabase,
    catalogue: &Catalogue,
    request: &ReviewPage,
) -> Result<ReviewRoots, Failure> {
    identifier(&request.branch)?;
    let cursor = request
        .after
        .as_deref()
        .map(|text| Cursor::decode(text, &request.branch))
        .transpose()?;
    let mut selection = catalogue.capture_held(&Selection {
        branch: request.branch.clone(),
        version: cursor.as_ref().map(|cursor| cursor.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.from.branch != from.branch || cursor.from.version != from.version
    }) {
        return Err(Failure::new(
            "invalid_continuation",
            "review origin differs from its captured provenance",
        ));
    }
    let base_moved = parent_movement(catalogue, &from)?;
    let mut parent = catalogue.capture_held(&Selection {
        branch: from.branch.clone(),
        version: Some(from.version),
    })?;
    parent.shards.sort_unstable_by_key(|shard| shard.shard_id);
    if parent.shards.len() != selection.shards.len()
        || parent
            .shards
            .iter()
            .zip(&selection.shards)
            .any(|(parent, branch)| parent.shard_id != branch.shard_id)
    {
        return Err(Failure::new(
            "incomplete_roots",
            "review origin and branch have different shard sets",
        ));
    }
    let start = match cursor.as_ref() {
        Some(cursor) => selection
            .shards
            .iter()
            .position(|shard| shard.shard_id == cursor.shard)
            .ok_or_else(|| {
                Failure::new(
                    "invalid_continuation",
                    "review continuation names an absent shard",
                )
            })?,
        None => 0,
    };
    Ok(ReviewRoots {
        selection,
        from,
        parent,
        cursor,
        start,
        base_moved,
    })
}

/// A review reads the fork version, which is kept whatever became of the
/// parent, so a parent that was discarded or replaced leaves the review
/// readable and only its movement unknown. The parent's head is read for its
/// version only, never its nodes, so it is not held.
fn parent_movement(catalogue: &Catalogue, from: &ForkProvenance) -> Result<Option<u64>, Failure> {
    let head = catalogue.capture(&Selection {
        branch: from.branch.clone(),
        version: None,
    });
    match head.and_then(|head| movement(from, &head)) {
        Ok(moved) => Ok(Some(moved)),
        Err(failure)
            if failure.code == "branch_missing" || failure.code == "parent_generation_changed" =>
        {
            Ok(None)
        }
        Err(failure) => Err(failure),
    }
}

fn movement(from: &ForkProvenance, parent: &Captured) -> Result<u64, Failure> {
    if from.version.generation != parent.version.generation {
        return Err(Failure::new(
            "parent_generation_changed",
            "parent was replaced after the branch forked",
        ));
    }
    parent
        .version
        .sequence
        .checked_sub(from.version.sequence)
        .ok_or_else(|| {
            Failure::new(
                "parent_sequence_regressed",
                "parent head precedes the fork version",
            )
        })
}

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

pub(super) fn diff_failure(error: ObserverError) -> Failure {
    match error {
        ObserverError::Diff(error @ DiffError::NodeBudgetTooSmall { .. }) => {
            Failure::new("node_budget_too_small", error.to_string())
        }
        other => Failure::observer(other),
    }
}

#[cfg(test)]
mod owner_boundary_tests {
    use std::error::Error;
    use std::path::Path;

    #[test]
    fn owner_metadata_filter_preserves_raw_scan_cursor_and_node_cost() {
        use haematite::db::ObservedChange;
        use haematite::tree::DiffPage;

        let change = |key: &[u8]| ObservedChange::Added {
            key: key.to_vec(),
            value: b"1".to_vec(),
        };
        let (page, scanned) = super::owner_changes(DiffPage {
            entries: vec![
                change(b"d\0records\0one"),
                change(b"k\0d\0records"),
                change(b"k\0ready"),
            ],
            next_after: Some(b"k\0ready".to_vec()),
            nodes_loaded: 7,
        });
        assert_eq!(scanned, 3);
        assert_eq!(page.entries.len(), 1);
        assert_eq!(page.entries[0].key(), b"d\0records\0one");
        assert_eq!(page.next_after.as_deref(), Some(b"k\0ready".as_slice()));
        assert_eq!(page.nodes_loaded, 7);
        let (page, scanned) = super::owner_changes(DiffPage {
            entries: vec![change(b"k\0d\0records"), change(b"k\0ready")],
            next_after: Some(b"k\0ready".to_vec()),
            nodes_loaded: 5,
        });
        assert_eq!(scanned, 2);
        assert!(page.entries.is_empty());
        assert_eq!(page.next_after.as_deref(), Some(b"k\0ready".as_slice()));
        assert_eq!(page.nodes_loaded, 5);
    }

    #[test]
    fn review_page_and_compare_roots_share_the_only_owner_change_boundary()
    -> Result<(), Box<dyn Error>> {
        let review = include_str!("review.rs");
        let comparison = include_str!("review_compare.rs");
        let page = review
            .split("pub(in crate::serve) fn page(")
            .nth(1)
            .ok_or("review page entry path missing")?;
        let page = page
            .split("pub(super) fn owner_changes_page(")
            .next()
            .ok_or("review page body missing")?;
        assert!(page.contains("owner_changes_page("));
        let roots = comparison
            .split("fn compare_roots(")
            .nth(1)
            .ok_or("comparison entry path missing")?;
        let roots = roots.split("fn ").next().ok_or("comparison body missing")?;
        assert_eq!(roots.matches("owner_changes_page(").count(), 2);
        let needle = [".changes_", "page("].concat();
        let ranged = [".changes_", "page_in_range("].concat();
        let mut directories = vec![Path::new(env!("CARGO_MANIFEST_DIR")).join("src/serve/storage")];
        let mut calls = Vec::new();
        let mut admission_calls = 0;
        let mut embed_calls = 0;
        let mut merge_calls = 0;
        while let Some(directory) = directories.pop() {
            for entry in std::fs::read_dir(directory)? {
                let entry = entry?;
                let kind = entry.file_type()?;
                let path = entry.path();
                if kind.is_dir() {
                    directories.push(path);
                } else if kind.is_file() && path.extension().is_some_and(|ext| ext == "rs") {
                    let text = std::fs::read_to_string(&path)?;
                    let compact: String = text.chars().filter(|ch| !ch.is_whitespace()).collect();
                    let direct = compact.matches(&needle).count();
                    let ranged = compact.matches(&ranged).count();
                    if path.file_name().and_then(|name| name.to_str()) == Some("admission_reads.rs")
                    {
                        // Bounded d, m and cascade edge dependency intervals do not walk review owners.
                        assert_eq!(direct, 0, "admission added an unrestricted diff walk");
                        assert_eq!(
                            ranged, 1,
                            "admission changed its single ranged diff boundary"
                        );
                        admission_calls += ranged;
                    } else if path.file_name().and_then(|name| name.to_str())
                        == Some("embed_pending.rs")
                    {
                        // The embedding job's walk diffs one collection's
                        // revision keys (`m\0{collection}\0`), bounded below
                        // and above (STOR-003 V3): it walks no review owners.
                        assert_eq!(direct, 0, "the embedding walk added an unrestricted diff");
                        assert_eq!(ranged, 1, "the embedding walk changed its one ranged diff");
                        embed_calls += ranged;
                    } else if path.file_name().and_then(|name| name.to_str()) == Some("merges.rs") {
                        // A fast-forward merge carries every change the source
                        // made over the target's own tree, its derived rows
                        // with it, so it walks past the owner filter on purpose.
                        assert_eq!(direct, 1, "the merge changed its one change walk");
                        assert_eq!(ranged, 0, "the merge added a ranged diff");
                        merge_calls += direct;
                    } else {
                        for _ in 0..direct + ranged {
                            calls.push(path.clone());
                        }
                    }
                }
            }
        }
        assert_eq!(admission_calls, 1, "bounded admission path is missing");
        assert_eq!(embed_calls, 1, "bounded embedding walk is missing");
        assert_eq!(merge_calls, 1, "the merge's change walk is missing");
        assert_eq!(
            calls.len(),
            1,
            "owner walk bypasses shared metadata boundary: {calls:?}"
        );
        assert_eq!(
            calls[0].file_name().and_then(|name| name.to_str()),
            Some("review.rs")
        );
        Ok(())
    }
}