Merge `inputs` into one new segment at `output_level`. `flush_lsn` on the new ref is the max of the inputs' lsns (no new WAL writes happen during compaction — recovery already covers the inputs). When `audit_retain_ms` is `Some(w)`, out-of-horizon tile versions (those with `system_from_ms < now_ms - w`) are collapsed into a single synthetic ceiling tile per Hilbert prefix. Cell-level semantics ar
(
store: &ArrayStore,
inputs: &[String],
output_level: u8,
audit_retain_ms: Option<i64>,
now_ms: i64,
)
| 66 | /// are preserved: each coordinate's newest out-of-horizon state is |
| 67 | /// carried forward (unless GDPR-erased), so no cells are silently lost. |
| 68 | pub fn run( |
| 69 | store: &ArrayStore, |
| 70 | inputs: &[String], |
| 71 | output_level: u8, |
| 72 | audit_retain_ms: Option<i64>, |
| 73 | now_ms: i64, |
| 74 | ) -> Result<CompactionOutput, CompactionError> { |
| 75 | let schema = store.schema().clone(); |
| 76 | let schema_hash = store.schema_hash(); |
| 77 | let mut merged: BTreeMap<TileId, MergedTile> = BTreeMap::new(); |
| 78 | let mut max_flush_lsn: u64 = 0; |
| 79 | for id in inputs { |
| 80 | let manifest_ref = store |
| 81 | .manifest() |
| 82 | .segments |
| 83 | .iter() |
| 84 | .find(|s| &s.id == id) |
| 85 | .ok_or_else(|| CompactionError::Io { |
| 86 | detail: format!("compaction input not in manifest: {id}"), |
| 87 | })?; |
| 88 | max_flush_lsn = max_flush_lsn.max(manifest_ref.flush_lsn); |
| 89 | let handle = store.segments.get(id).ok_or_else(|| CompactionError::Io { |
| 90 | detail: format!("compaction input has no open handle: {id}"), |
| 91 | })?; |
| 92 | let reader = handle.reader(); |
| 93 | for (tile_idx, entry) in reader.tiles().iter().enumerate() { |
| 94 | let tile_id = entry.tile_id; |
| 95 | let payload = reader.read_tile(tile_idx)?; |
| 96 | merged |
| 97 | .entry(tile_id) |
| 98 | .or_insert_with(|| MergedTile::empty(&schema)) |
| 99 | .absorb(&schema, &payload)?; |
| 100 | } |
| 101 | } |
| 102 | |
| 103 | // Apply retention if configured. |
| 104 | let merged = match audit_retain_ms { |
| 105 | None => merged, |
| 106 | Some(retain_ms) => { |
| 107 | let horizon_ms = now_ms.saturating_sub(retain_ms); |
| 108 | apply_retention(merged, &schema, horizon_ms)? |
| 109 | } |
| 110 | }; |
| 111 | |
| 112 | let kek = store.kek().cloned(); |
| 113 | let id = next_segment_id_for_compaction(store, inputs); |
| 114 | let seg_path = store.root().join(&id); |
| 115 | let writer_bytes = |
| 116 | build_segment_bytes(&schema, schema_hash, kek.as_ref(), merged.into_iter())?; |
| 117 | write_atomic(&seg_path, &writer_bytes).map_err(|e| CompactionError::Io { |
| 118 | detail: format!("write merged segment {seg_path:?}: {e}"), |
| 119 | })?; |
| 120 | |
| 121 | // Reopen from the written bytes to pull tile bounds. |
| 122 | // When encryption is active, writer_bytes is an encrypted SEGA blob; |
| 123 | // use OwnedSegmentReader to decrypt before inspecting tiles. |
| 124 | let (min_tile, max_tile, tile_count) = { |
| 125 | let owned = nodedb_array::segment::reader::OwnedSegmentReader::open_with_kek( |
nothing calls this directly
no test coverage detected