nordisk/nornir

src/code_room/lake_lane.rs

//! The lake's compaction, off the consumer and off the panes.
//!
//! The lake was written uncompressed, one data file per release per table
//! (17 GB on dwarf for ~1.7 GB of zstd, 2026-10-16). The consumer now writes
//! zstd; `mimir_ingest::EventLog::compact` rewrites what was written before
//! and merges the tail of small files, one replace snapshot per table. The
//! server runs it as a jera chain job on its own `lake` lane (low priority):
//! this is the job's body.

use mimir_ingest::{CompactOptions, CompactReport, MaterializeSweep, SplitOptions, SplitReport};

use super::CodeRoom;

impl CodeRoom {
    /// Is the workspace's lake in the legacy shape (ring rows in the source
    /// tables, no repo partition)? `None`: the lake is not open yet.
    pub fn lake_is_legacy(&self) -> Option<bool> {
        self.is_open().then(|| self.log().map(|l| !l.ring_split()).unwrap_or(false))
    }

    /// The one-shot migration (`EventLog::split_ring`): the ring copied to
    /// its own tables, verified, the lake switched; the next compaction
    /// partitions the source tables and drops their ring rows.
    pub fn run_lake_ring_split(&self) -> anyhow::Result<SplitReport> {
        let log = self.log().map_err(|r| anyhow::anyhow!("{}", r.note))?;
        let opts = SplitOptions::default();
        std::thread::scope(|s| s.spawn(|| log.block_on(log.split_ring(&opts))).join().unwrap_or_else(|p| std::panic::resume_unwind(p)))
    }

    /// The job: compact every table of the workspace's lake with the default
    /// options. Runs the fold-free rewrite on the lake's runtime from this
    /// (blocking) thread; the consumer goes on meanwhile — the commits
    /// serialize on the log's own lock.
    pub fn run_lake_compaction(&self) -> anyhow::Result<Vec<CompactReport>> {
        let log = self.log().map_err(|r| anyhow::anyhow!("{}", r.note))?;
        let opts = CompactOptions::default();
        std::thread::scope(|s| s.spawn(|| log.block_on(log.compact(&opts))).join().unwrap_or_else(|p| std::panic::resume_unwind(p)))
    }

    /// The job: refresh the READ-FAST zone for EVERY consumed repo stream —
    /// each repo whose anchor is behind its offset gets its interval rows
    /// written and its anchor moved
    /// (`mimir_ingest::EventLog::materialize_all`). Streams already at their
    /// tip are counted as skipped; a stream that fails is in the sweep's
    /// `failed` and never stops the streams behind it.
    pub fn run_lake_readfast(&self) -> anyhow::Result<MaterializeSweep> {
        let log = self.log().map_err(|r| anyhow::anyhow!("{}", r.note))?;
        std::thread::scope(|s| s.spawn(|| log.block_on(log.materialize_all())).join().unwrap_or_else(|p| std::panic::resume_unwind(p)))
    }
}