Function plan

nodedb/src/engine/array/purge/plan.rs:57–290

Compute the purge plan for an array whose store is `store` and whose bitemporal horizon is `horizon_ms`. `horizon_ms` must already be the absolute system-time cutoff (i.e. `now_ms - audit_retain_ms`). This function does **not** recompute it.

(
    store: &ArrayStore,
    horizon_ms: i64,
    schema: &ArraySchema,
)

/// `horizon_ms` must already be the absolute system-time cutoff (i.e.
/// `now_ms - audit_retain_ms`). This function does **not** recompute it.
pub fn plan(
    store: &ArrayStore,
    horizon_ms: i64,
    schema: &ArraySchema,
) -> Result<PurgePlan, ArrayError> {
    if store.manifest().segments.is_empty() {
        return Ok(PurgePlan {
            segment_actions: Vec::new(),
            cells_carried_forward: 0,
        });
    }

    // ── Step 1: Collect all TileEntries tagged with their owning segment id and
    //   flush_lsn (used to choose the ceiling host). ────────────────────────────
    struct TaggedEntry {
        segment_id: String,
        flush_lsn: u64,
        tile_id: TileId,
    }

    let mut all_entries: Vec<TaggedEntry> = Vec::new();
    for seg_ref in &store.manifest().segments {
        let handle = match store.segments.get(&seg_ref.id) {
            Some(h) => h,
            None => continue, // segment in manifest but not in open handles — skip
        };
        let reader = handle.reader();
        for entry in reader.tiles() {
            all_entries.push(TaggedEntry {
                segment_id: seg_ref.id.clone(),
                flush_lsn: seg_ref.flush_lsn,
                tile_id: entry.tile_id,
            });
        }
    }

    if all_entries.is_empty() {
        return Ok(PurgePlan {
            segment_actions: Vec::new(),
            cells_carried_forward: 0,
        });
    }

    // ── Step 2: Check whether any tile is out-of-horizon. If none, no-op. ──────
    let any_outside = all_entries
        .iter()
        .any(|e| e.tile_id.system_from_ms < horizon_ms);
    if !any_outside {
        return Ok(PurgePlan {
            segment_actions: Vec::new(),
            cells_carried_forward: 0,
        });
    }

    // ── Step 3: Group entries by hilbert_prefix. ─────────────────────────────
    // For each prefix, build: inside list, outside list (newest→oldest), and
    // the earliest-flushed segment id for the ceiling host.
    struct PrefixGroup {