Execute one GC pass across all known arrays.
(
op_log: &OriginOpLog,
snapshots: &OriginSnapshotStore,
ack_registry: &ArrayAckRegistry,
array_snapshot_hlcs: &RwLock<HashMap<String, Hlc>>,
)
| 74 | |
| 75 | /// Execute one GC pass across all known arrays. |
| 76 | fn run_gc( |
| 77 | op_log: &OriginOpLog, |
| 78 | snapshots: &OriginSnapshotStore, |
| 79 | ack_registry: &ArrayAckRegistry, |
| 80 | array_snapshot_hlcs: &RwLock<HashMap<String, Hlc>>, |
| 81 | ) { |
| 82 | let arrays = ack_registry.known_arrays(); |
| 83 | if arrays.is_empty() { |
| 84 | debug!("array_gc_task: no arrays with acks — skipping"); |
| 85 | return; |
| 86 | } |
| 87 | |
| 88 | for array in &arrays { |
| 89 | let ack_vector = ack_registry.ack_vector(array); |
| 90 | if ack_vector.min_ack_hlc().is_none() { |
| 91 | debug!(array = %array, "array_gc_task: no min_ack for array — skipping"); |
| 92 | continue; |
| 93 | } |
| 94 | |
| 95 | // Build a snapshot-for-array closure that reads the latest persisted |
| 96 | // snapshot and presents it as the GC snapshot. If no snapshot exists |
| 97 | // we return None — GC still drops ops below the frontier but writes |
| 98 | // no new snapshot (no data to compact for a newly-created array that |
| 99 | // has never had a snapshot taken). |
| 100 | let array_name = array.clone(); |
| 101 | let snapshots_ref = snapshots; |
| 102 | |
| 103 | let report = collapse_below(op_log, &ack_vector, snapshots_ref, |arr, frontier| { |
| 104 | // Try to reuse the latest stored snapshot for this array. |
| 105 | // In a full Origin implementation the snapshot would be built |
| 106 | // from live tile state; here we promote the latest stored |
| 107 | // snapshot's HLC to the new frontier so the GC boundary |
| 108 | // advances even between full compactions. |
| 109 | if let Some(mut snap) = snapshots_ref.latest_for_array(arr) { |
| 110 | snap.snapshot_hlc = frontier; |
| 111 | Ok(Some(snap)) |
| 112 | } else { |
| 113 | // No snapshot yet for this array; GC will drop ops without |
| 114 | // writing a replacement snapshot. New peers will get the |
| 115 | // op-stream from Hlc::ZERO (or catch up via future snapshots). |
| 116 | Ok(None) |
| 117 | } |
| 118 | }); |
| 119 | |
| 120 | match report { |
| 121 | Ok(r) => { |
| 122 | if r.ops_dropped > 0 || r.snapshots_written > 0 { |
| 123 | info!( |
| 124 | array = %array_name, |
| 125 | ops_dropped = r.ops_dropped, |
| 126 | snapshots_written = r.snapshots_written, |
| 127 | frontier = ?r.frontier, |
| 128 | "array_gc_task: GC run complete" |
| 129 | ); |
| 130 | } |
| 131 | |
| 132 | if let Some(frontier) = ack_vector.min_ack_hlc() { |
| 133 | // Evict obsoleted snapshots. |
nothing calls this directly
no test coverage detected