Function collapse_below

nodedb-array/src/sync/gc.rs:50–93

Compact ops below the min-ack frontier into snapshots, then prune the log. Algorithm: 1. Determine frontier = `acks.min_ack_hlc()`. If `None`, return early with an empty report — GC must not proceed without knowing every peer's progress. 2. Collect distinct array names whose ops have `hlc < frontier`. 3. For each such array, call `snapshot_for_array(name, frontier)`. - `Some(snap)` → write via `s

(
    log: &dyn OpLog,
    acks: &AckVector,
    sink: &dyn SnapshotSink,
    snapshot_for_array: impl Fn(&str, Hlc) -> ArrayResult<Option<TileSnapshot>>,
)

///    the count in `ops_dropped`.
/// 5. Return the report.
pub fn collapse_below(
    log: &dyn OpLog,
    acks: &AckVector,
    sink: &dyn SnapshotSink,
    snapshot_for_array: impl Fn(&str, Hlc) -> ArrayResult<Option<TileSnapshot>>,
) -> ArrayResult<GcReport> {
    let frontier = match acks.min_ack_hlc() {
        None => {
            return Ok(GcReport {
                snapshots_written: 0,
                ops_dropped: 0,
                frontier: Hlc::ZERO,
            });
        }
        Some(h) => h,
    };

    // Collect distinct array names with ops below the frontier.
    let mut arrays_to_snapshot: HashSet<String> = HashSet::new();
    for item in log.scan_from(Hlc::ZERO)? {
        let op = item?;
        if op.header.hlc < frontier {
            arrays_to_snapshot.insert(op.header.array.clone());
        }
    }

    // Write snapshots — abort on the first error; do NOT mutate the log yet.
    let mut snapshots_written: u64 = 0;
    for array in &arrays_to_snapshot {
        if let Some(snap) = snapshot_for_array(array, frontier)? {
            sink.write_snapshot(&snap)?;
            snapshots_written += 1;
        }
    }

    // All snapshots succeeded — now prune the log.
    let ops_dropped = log.drop_below(frontier)?;

    Ok(GcReport {
        snapshots_written,
        ops_dropped,
        frontier,
    })
}

// ─── MockSnapshotSink ────────────────────────────────────────────────────────