Function analyze_split

atomic-repository/src/repository/split.rs:158–244

Compute the reverse-dependency closure of `requested` within `source`. Bounds are `T: ViewTxnT` (which requires `GraphTxnT`), so this works on both read and write transactions and can be reused for dry-run previews.

(
    txn: &T,
    source: &ViewState,
    requested: &[Hash],
)

/// Bounds are `T: ViewTxnT` (which requires `GraphTxnT`), so this works on both
/// read and write transactions and can be reused for dry-run previews.
fn analyze_split<T: ViewTxnT>(
    txn: &T,
    source: &ViewState,
    requested: &[Hash],
) -> Result<SplitAnalysis, RepositoryError> {
    let db = |e: PristineError| RepositoryError::Database(e.to_string());

    // Resolve each requested hash to an internal id and confirm it is in the
    // source view's OWN change log.
    let mut requested_ids: HashSet<NodeId> = HashSet::new();
    let mut queue: VecDeque<(Hash, NodeId)> = VecDeque::new();
    for hash in requested {
        let id =
            txn.get_internal(hash)
                .map_err(db)?
                .ok_or_else(|| RepositoryError::ChangeNotFound {
                    hash: hash.to_base32(),
                })?;
        if txn.get_change_seq(source, id).map_err(db)?.is_none() {
            return Err(RepositoryError::ChangeNotInView {
                hash: hash.to_base32(),
                view: source.name.clone(),
            });
        }
        if requested_ids.insert(id) {
            queue.push_back((*hash, id));
        }
    }

    // BFS over reverse dependencies, staying inside the source view.
    let mut closure_ids: HashSet<NodeId> = requested_ids.clone();
    while let Some((hash, _id)) = queue.pop_front() {
        let dependents = txn.get_rev_change_deps(&hash).map_err(db)?;
        for dep_id in dependents {
            // Only dependents that actually live in the source view matter;
            // dependents in other views don't affect this view's coherence.
            if txn.get_change_seq(source, dep_id).map_err(db)?.is_none() {
                continue;
            }
            if closure_ids.insert(dep_id) {
                let dep_hash = txn.get_external(dep_id).map_err(db)?.ok_or_else(|| {
                    RepositoryError::ChangeNotFound {
                        hash: format!("id={}", dep_id.0),
                    }
                })?;
                queue.push_back((dep_hash, dep_id));
            }
        }
    }

    // Build ordered lists keyed by source sequence.
    let ordered = |ids: &HashSet<NodeId>| -> Result<Vec<SplitChange>, RepositoryError> {
        let mut v = Vec::with_capacity(ids.len());
        for &id in ids {
            let seq = txn
                .get_change_seq(source, id)
                .map_err(db)?
                .expect("closure member is in source view");