knut-bifrost 0.3.0

//! **Iceberg → table delivery**: the loop that reads an Iceberg table's change
//! windows and hands them to a [`TableSink`] (Doris), one snapshot at a time.
//!
//! Feature `skade`. The pieces it drives already existed — skade's
//! `read_changelog`, [`partition_change_rows`], [`deliver_window`] — and
//! nothing ran them end to end. This is that driver, and nothing more.
//!
//! # Where the table is
//!
//! [`IcebergLocator`] names the table three ways, one per kind of lake:
//!
//! | locator | how the table is found | feature |
//! |---|---|---|
//! | `Warehouse` | an embedded skade warehouse directory | `skade` |
//! | `S3` | its S3 root, **no catalog** — `version-hint.text`, else the newest `*.metadata.json` | `skade-s3` |
//! | `Catalog` | any `dyn skade::connect::CatalogConnector`, `ns.name` in it | `skade` |
//!
//! The catalogs bifrost's flags build a `Catalog` locator from:
//!
//! | connector | catalog | flag | feature |
//! |---|---|---|---|
//! | `IcebergRest` | an Iceberg REST catalog; MinIO AIStor's `/_iceberg` with SigV4 service `s3tables` | `--rest-uri` | `skade-rest` |
//! | `SkadeGrpc` | a skade-katalog gRPC front (`skade-katalog serve --grpc`) | `--grpc-uri` | `skade-grpc` |
//!
//! The cursor key names the catalog by its connector's `catalog_id`, which
//! for a REST catalog is the `rest:<uri>#<warehouse>` form the key always had.
//!
//! Every round loads the table again, so a new commit is seen on the next round
//! whichever way the table is found.
//!
//! # One snapshot per window
//!
//! A window is `(previous snapshot, next snapshot]`, never a wider range:
//!
//! * **memory** — `read_changelog` materialises the whole window; one commit at
//!   a time bounds it by the largest single commit, not by the backlog;
//! * **ordering** — inside one Iceberg commit an update is an equality delete
//!   plus an append of the same key; the sink ships deletes before upserts, so
//!   a one-commit window is always applied in the right order. A multi-commit
//!   window could hold insert-then-delete of one key, which the sink cannot
//!   order (`sink::doris` says so);
//! * **replay** — the label is derived from the snapshot pair, so a crash
//!   replays exactly the commit that was in flight.
//!
//! # The initial load, and what is refused after it
//!
//! * **No cursor** (first run): the whole **current** snapshot as inserts, one
//!   window `(None, current]` (`skade::read_snapshot_changelog`) — never a
//!   replay of history. It is held in memory in full; the log says so.
//! * **Then one snapshot per window**, whatever the writer did — appends,
//!   copy-on-write overwrites, merge-on-read deletes (v3 deletion vectors,
//!   which Spark 4.1 + Iceberg 1.12 writes for `MERGE INTO`), equality deletes.
//!   Each window carries its snapshot's data sequence number, which a Unique Key
//!   destination versions every row by.
//! * A **`replace`** (compaction) changes no row: delivered as an empty window,
//!   no read, no load, the cursor moves past it.
//! * A commit skade cannot read is refused and the cursor held
//!   (`refuse_full_reload` in `sink::table`).
//! * The **cursor's snapshot expired** from the table: a
//!   [`BifrostError::FullReloadRequired`] error, never a silent resync. The
//!   operator empties the destination, deletes the cursor, and restarts.
//!
//! # Near-live
//!
//! [`IcebergDelivery::run`] polls (default [`DEFAULT_POLL`]): every round
//! re-loads the table metadata through its [`IcebergLocator`] and drains the
//! new snapshots, oldest first. A stop signal is honoured between snapshots,
//! after the in-flight one's cursor commit.

use std::sync::Arc;
use std::time::{Duration, Instant};

use skade::connect::CatalogConnector;
use skade::iceberg::spec::Operation;
use skade::iceberg::table::Table as IceTable;

use crate::{
    error::{BifrostError, Result},
    sink::table::{SnapshotWindow, TableSink, TableWindow, deliver_window},
    state::SyncState,
    transform::partition_change_rows,
};

/// Where an Iceberg table is. See the module docs.
#[derive(Debug, Clone)]
pub enum IcebergLocator {
    /// An embedded skade warehouse at `dir`, table `table` (`ns.name` or `name`).
    Warehouse { dir: std::path::PathBuf, table: String },
    /// A table on S3/MinIO by its root, no catalog.
    #[cfg(feature = "skade-s3")]
    S3(skade::connect::S3Table),
    /// A table behind a catalog — an Iceberg REST catalog (`IcebergRest`), a
    /// skade-katalog gRPC front (`SkadeGrpc`), any [`CatalogConnector`] —
    /// `table` = `ns.name`.
    Catalog { connector: Arc<dyn CatalogConnector>, table: String },
}

impl IcebergLocator {
    /// The table `table` (`ns.name`) behind `connector`.
    pub fn catalog(connector: impl CatalogConnector + 'static, table: impl Into<String>) -> Self {
        IcebergLocator::Catalog { connector: Arc::new(connector), table: table.into() }
    }

    /// A stable, secret-free name for the table — the cursor key's source half.
    pub fn source_id(&self) -> String {
        match self {
            IcebergLocator::Warehouse { dir, table } => format!("skade:{}#{table}", dir.display()),
            #[cfg(feature = "skade-s3")]
            IcebergLocator::S3(t) => format!("s3:{}", t.location.trim_end_matches('/')),
            IcebergLocator::Catalog { connector, table } => format!("{}#{table}", connector.catalog_id()),
        }
    }

    /// Load the table at its current metadata.
    pub async fn load(&self) -> Result<IceTable> {
        let src = |e: skade::SkadeError| BifrostError::Source(format!("{}: {e}", self.source_id()));
        match self {
            IcebergLocator::Warehouse { dir, table } => {
                let wh = skade::Warehouse::open(dir).await.map_err(src)?;
                Ok(wh.table(table).await.map_err(src)?.inner().clone())
            }
            #[cfg(feature = "skade-s3")]
            IcebergLocator::S3(t) => t.load().await.map_err(src),
            IcebergLocator::Catalog { connector, table } => {
                let ident = skade::iceberg::TableIdent::from_strs(table.split('.')).map_err(|e| {
                    BifrostError::Source(format!("{}: bad table name {table:?}: {e}", self.source_id()))
                })?;
                let cat = connector.open().await.map_err(src)?;
                cat.load_table(&ident)
                    .await
                    .map_err(|e| BifrostError::Source(format!("{}: {e}", self.source_id())))
            }
        }
    }
}

/// Where a lake of **many** Iceberg tables is — an [`IcebergLocator`] per table,
/// on demand. `bifrost watch` streams every table a mapping names, so it holds a
/// lake and asks it for each table by the mapping's name:
///
/// | lake | the mapping's `table:` | feature |
/// |---|---|---|
/// | `Warehouse` | `name` or `ns.name` in the embedded skade warehouse | `skade` |
/// | `S3` | an `s3://…` table root verbatim, else `ns.name` → `<root>/ns/name` | `skade-s3` |
/// | `Catalog` | `ns.name` in the catalog, a bare `name` in its `namespace` — REST (MinIO AIStor: `/_iceberg`, SigV4 `s3tables`) or skade-katalog gRPC | `skade-rest` / `skade-grpc` |
#[derive(Debug, Clone)]
pub enum IcebergLake {
    /// An embedded skade warehouse directory.
    Warehouse(std::path::PathBuf),
    /// Iceberg tables on S3/MinIO read with no catalog: each table's current
    /// metadata is found under its root (`version-hint.text`, else the newest
    /// `*.metadata.json`).
    #[cfg(feature = "skade-s3")]
    S3 { s3: skade::S3Config, root: String },
    /// A catalog (Iceberg REST, skade-katalog gRPC, …); the tables' files are
    /// read the way its connector says. A bare table name (no `ns.`) is in
    /// `namespace`, if one is given.
    Catalog { connector: Arc<dyn CatalogConnector>, namespace: Option<String> },
}

impl IcebergLake {
    /// The locator for the mapping's table `table`.
    pub fn locate(&self, table: &str) -> IcebergLocator {
        match self {
            IcebergLake::Warehouse(dir) => IcebergLocator::Warehouse { dir: dir.clone(), table: table.to_string() },
            #[cfg(feature = "skade-s3")]
            IcebergLake::S3 { s3, root } => {
                let location = if bucket_of(table).is_some() {
                    table.to_string()
                } else {
                    format!("{}/{}", root.trim_end_matches('/'), table.replace('.', "/"))
                };
                IcebergLocator::S3(skade::connect::S3Table { s3: s3.clone(), location })
            }
            IcebergLake::Catalog { connector, namespace } => {
                let table = match namespace {
                    Some(ns) if !table.contains('.') => format!("{ns}.{table}"),
                    _ => table.to_string(),
                };
                IcebergLocator::Catalog { connector: connector.clone(), table }
            }
        }
    }

    /// Where the lake is, for logs. Never a secret.
    pub fn describe(&self) -> String {
        match self {
            IcebergLake::Warehouse(dir) => format!("skade warehouse {}", dir.display()),
            #[cfg(feature = "skade-s3")]
            IcebergLake::S3 { s3, root } => format!("s3 {root} via {}", s3.endpoint),
            IcebergLake::Catalog { connector, namespace } => match namespace {
                Some(ns) => format!("{} namespace {ns}", connector.describe()),
                None => connector.describe(),
            },
        }
    }
}

#[cfg(feature = "skade-s3")]
/// The bucket named in an `s3://bucket/…` URL (`s3a://` and `s3n://`, which Spark
/// writes, too).
pub fn bucket_of(url: &str) -> Option<&str> {
    ["s3://", "s3a://", "s3n://"]
        .iter()
        .find_map(|s| url.strip_prefix(s))
        .and_then(|r| r.split('/').next())
        .filter(|b| !b.is_empty())
}

#[cfg(feature = "skade-s3")]
fn env(name: &str) -> Option<String> {
    std::env::var(name).ok().filter(|v| !v.is_empty())
}

#[cfg(feature = "skade-s3")]
fn need_env(name: &str) -> Result<String> {
    env(name).ok_or_else(|| BifrostError::Other(format!("{name} is not set (secrets come from the environment, never a flag)")))
}

#[cfg(feature = "skade-s3")]
/// The CA to verify an `https://` S3/REST endpoint with: `flag` (a PEM file),
/// else `BIFROST_S3_CA_PEM` (the PEM text) written to `<state_dir>/s3-ca.pem`.
pub fn s3_ca(flag: Option<&std::path::Path>, state_dir: &std::path::Path) -> Result<Option<std::path::PathBuf>> {
    if let Some(p) = flag {
        return Ok(Some(p.to_path_buf()));
    }
    let Some(pem) = env("BIFROST_S3_CA_PEM") else { return Ok(None) };
    std::fs::create_dir_all(state_dir)?;
    let path = state_dir.join("s3-ca.pem");
    std::fs::write(&path, pem)?;
    Ok(Some(path))
}

#[cfg(feature = "skade-s3")]
/// The S3/MinIO config for a lake's files. The keys come from `MINIO_ACCESS_KEY`
/// and `MINIO_SECRET_KEY` only — a flag lands in `/proc/<pid>/cmdline`. They sign
/// the S3 reads and, for a REST catalog with SigV4 (AIStor), every catalog call.
pub fn s3_config(endpoint: String, region: String, bucket: String, ca: Option<std::path::PathBuf>) -> Result<skade::S3Config> {
    Ok(skade::S3Config {
        endpoint,
        region,
        bucket,
        access_key_id: need_env("MINIO_ACCESS_KEY")?,
        secret_access_key: need_env("MINIO_SECRET_KEY")?,
        path_style: true,
        tls: ca.map(|ca| skade::tls::ClientTls { ca_pem: Some(ca), ..Default::default() }),
    })
}

/// Knobs for [`IcebergDelivery`].
#[derive(Debug, Clone)]
pub struct DeliveryOptions {
    /// At most this many one-snapshot windows per [`IcebergDelivery::round`];
    /// the rest wait for the next round, so one round never runs unbounded.
    pub max_windows_per_round: usize,
    /// The row identity, when the Iceberg table declares no identifier fields
    /// (a plain Spark table) — the destination's key columns. Empty = what the
    /// changelog knows (identifier fields, equality-delete ids).
    pub key_columns: Vec<String>,
}

impl Default for DeliveryOptions {
    fn default() -> Self {
        Self { max_windows_per_round: 64, key_columns: Vec::new() }
    }
}

/// Default seconds between polls while healthy: near-live, and cheap — a poll
/// that finds no new snapshot reads one metadata file.
pub const DEFAULT_POLL: Duration = Duration::from_secs(5);

/// What one [`IcebergDelivery::round`] did.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RoundOutcome {
    /// Windows delivered (and cursor advances made) this round.
    pub windows: usize,
    pub rows_loaded: usize,
    pub rows_deleted: usize,
    /// Rows the store refused as an already-seen label: a replay, not an error.
    pub rows_duplicate: usize,
    /// The snapshot the cursor stands on after the round.
    pub at_snapshot: Option<i64>,
    /// Snapshots still to deliver after this round (0 = caught up).
    pub behind: usize,
    /// Wall time reading changelogs (table load + read_changelog + partition).
    pub read: Duration,
    /// Wall time inside the sink (encode + load + cursor commit).
    pub load: Duration,
}

/// The delivery loop: one Iceberg table into one table sink.
pub struct IcebergDelivery<S: TableSink> {
    locator: IcebergLocator,
    sink: S,
    state: SyncState,
    key: String,
    opts: DeliveryOptions,
    probed: bool,
}

impl<S: TableSink> IcebergDelivery<S> {
    /// `destination` names where the rows go (e.g. `doris:db.table`); it is
    /// part of the cursor key, so one source delivered to two destinations
    /// keeps two cursors.
    pub fn new(
        locator: IcebergLocator,
        destination: &str,
        sink: S,
        state: SyncState,
        opts: DeliveryOptions,
    ) -> Self {
        let key = format!("{} -> {destination}", locator.source_id());
        Self { locator, sink, state, key, opts, probed: false }
    }

    /// The cursor key this delivery reads and advances.
    pub fn cursor_key(&self) -> &str {
        &self.key
    }

    /// The snapshot the cursor stands on.
    pub fn cursor(&self) -> Option<i64> {
        self.state.get(&self.key).snapshot_id
    }

    /// Deliver what is waiting, up to [`DeliveryOptions::max_windows_per_round`]
    /// windows. Every delivered window has advanced the cursor before the next
    /// one is read; an error leaves the cursor on the last delivered window.
    pub async fn round(&mut self) -> Result<RoundOutcome> {
        let mut out = RoundOutcome::default();
        let t0 = Instant::now();
        let table = self.locator.load().await?;
        out.read += t0.elapsed();

        let from = self.cursor();
        out.at_snapshot = from;
        let Some(current) = table.metadata().current_snapshot().map(|s| s.snapshot_id()) else {
            return Ok(out); // no snapshot yet: nothing to deliver
        };
        if from == Some(current) {
            return Ok(out);
        }
        // First run: no cursor. The initial load is the whole current snapshot
        // as inserts — never a replay of history, which may hold overwrites,
        // deletes and compactions the destination never saw.
        let Some(from_id) = from else {
            tracing::warn!(
                target: "bifrost::delivery",
                event = "initial_full_snapshot",
                table = %self.key,
                to = current,
                "no cursor: the initial load is the whole current snapshot, held in memory in full",
            );
            self.deliver(&table, None, current, &mut out).await?;
            out.behind = 0;
            return Ok(out);
        };

        // The cursor's snapshot must still be in the table. Snapshot expiry can
        // remove it, and then the changes since it are unreadable — say so,
        // never resync silently.
        if table.metadata().snapshot_by_id(from_id).is_none() {
            return Err(BifrostError::FullReloadRequired(format!(
                "{}: the cursor's snapshot {from_id} has been expired from the table, so the \
                 changes since it can no longer be read. Empty the destination table, delete \
                 this table's cursor, and restart: the next run does the initial full load.",
                self.key
            )));
        }
        let chain = table
            .metadata()
            .snapshots_added_between(from, current)
            .map_err(|e| BifrostError::Source(format!("{}: snapshot lineage: {e}", self.key)))?;

        let mut prev = from;
        for (i, &snap) in chain.iter().enumerate() {
            if i == self.opts.max_windows_per_round {
                out.behind = chain.len() - i;
                break;
            }
            self.deliver(&table, prev, snap, &mut out).await?;
            prev = Some(snap);
        }
        Ok(out)
    }

    /// Read one window, probe the sink once, deliver, account.
    ///
    /// `from = None` is the initial load: the current snapshot's live rows.
    /// A `replace` snapshot (compaction) changes no row: it is delivered as an
    /// empty window, so the cursor moves past it without a read or a load.
    async fn deliver(&mut self, table: &IceTable, from: Option<i64>, to: i64, out: &mut RoundOutcome) -> Result<()> {
        let src = |e: skade::SkadeError| BifrostError::Source(format!("{e}"));
        let t = Instant::now();
        if !self.probed {
            let schema = skade::arrow_schema_of(table).map_err(src)?;
            self.sink.probe(&schema).await?;
            self.probed = true;
        }
        let snapshot = table.metadata().snapshot_by_id(to);
        let replace = from.is_some() && snapshot.is_some_and(|s| matches!(s.summary().operation, Operation::Replace));
        let sequence = snapshot.map(|s| s.sequence_number());

        let (rows, keys, needs_full_reload) = if replace {
            (crate::transform::ChangeRows::default(), Vec::new(), false)
        } else {
            let cl = match from {
                None => skade::read_snapshot_changelog(table, to).await.map_err(src)?,
                Some(_) => skade::read_changelog(table, from, to).await.map_err(src)?,
            };
            // An unreadable commit (a delete shape skade's planner does not
            // know): `deliver_window` refuses the window and holds the cursor.
            let needs_full_reload = cl.plan.needs_full_reload && from.is_some();
            let keys = if self.opts.key_columns.is_empty() { cl.key_columns().to_vec() } else { self.opts.key_columns.clone() };
            (partition_change_rows(&cl.rows)?, keys, needs_full_reload)
        };
        out.read += t.elapsed();

        let window = TableWindow {
            table: &self.key,
            snapshots: SnapshotWindow::new(from, to),
            upserts: &rows.upserts,
            deletes: &rows.deletes,
            needs_full_reload,
            sequence,
            key_columns: &keys,
        };
        let t = Instant::now();
        let ack = deliver_window(&mut self.sink, &mut self.state, &window).await?;
        out.load += t.elapsed();

        out.windows += 1;
        out.rows_loaded += ack.rows_loaded;
        out.rows_deleted += ack.rows_deleted;
        out.rows_duplicate += ack.rows_rejected_duplicate;
        out.at_snapshot = Some(ack.to_snapshot());
        Ok(())
    }

    /// Run rounds until `stop` resolves: `poll` between rounds while healthy,
    /// doubling up to `max_backoff` while rounds fail. A round in flight when
    /// `stop` fires is finished first — its cursor commit included — so a
    /// SIGTERM never leaves a loaded-but-uncommitted window behind.
    pub async fn run(
        &mut self,
        poll: Duration,
        max_backoff: Duration,
        stop: impl std::future::Future<Output = ()>,
    ) -> Result<()> {
        let mut backoff = crate::metrics::Backoff::new(poll, max_backoff);
        tokio::pin!(stop);
        loop {
            let result = self.round().await;
            let failed = match &result {
                Ok(o) => {
                    if o.windows > 0 {
                        tracing::info!(
                            target: "bifrost::delivery",
                            event = "round",
                            table = %self.key,
                            windows = o.windows,
                            loaded = o.rows_loaded,
                            deleted = o.rows_deleted,
                            duplicate = o.rows_duplicate,
                            at = ?o.at_snapshot,
                            behind = o.behind,
                            read_ms = o.read.as_millis() as u64,
                            load_ms = o.load.as_millis() as u64,
                            "delivered",
                        );
                    }
                    false
                }
                Err(e) => {
                    tracing::error!(target: "bifrost::delivery", event = "round_failed", table = %self.key, error = %e, "round failed, cursor held");
                    true
                }
            };
            let behind = matches!(&result, Ok(o) if o.behind > 0);
            let sleep = if behind { Duration::ZERO } else { backoff.next_sleep(failed) };
            tokio::select! {
                _ = &mut stop => return Ok(()),
                _ = tokio::time::sleep(sleep) => {}
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, Mutex};

    use arrow_array::{Int64Array, StringArray};
    use arrow_schema::{DataType, Field, Schema};
    use async_trait::async_trait;

    use super::*;
    use arrow_array::RecordBatch;

    use crate::sink::table::{AckToken, LoadAck};

    /// What the sink saw of one window.
    #[derive(Clone, Debug, PartialEq, Eq)]
    struct Seen {
        from: Option<i64>,
        to: i64,
        upserts: usize,
        deletes: usize,
        sequence: Option<i64>,
        keys: Vec<String>,
    }

    /// A sink that records every window it is handed.
    #[derive(Clone, Default)]
    struct Recording {
        windows: Arc<Mutex<Vec<Seen>>>,
        probes: Arc<Mutex<usize>>,
        fail_at: Option<i64>,
    }

    impl Recording {
        fn seen(&self) -> Vec<Seen> {
            self.windows.lock().unwrap().clone()
        }
    }

    #[async_trait]
    impl TableSink for Recording {
        async fn probe(&mut self, _: &Schema) -> Result<()> {
            *self.probes.lock().unwrap() += 1;
            Ok(())
        }
        async fn apply(&mut self, w: &TableWindow<'_>, token: AckToken) -> Result<LoadAck> {
            if self.fail_at == Some(w.snapshots.to) {
                return Err(BifrostError::Other("sink down".into()));
            }
            self.windows.lock().unwrap().push(Seen {
                from: w.snapshots.from,
                to: w.snapshots.to,
                upserts: w.upsert_rows(),
                deletes: w.delete_rows(),
                sequence: w.sequence,
                keys: w.key_columns.to_vec(),
            });
            Ok(token.ack(w.snapshots.to, vec![]).with_counts(w.upsert_rows(), w.delete_rows(), 0))
        }
    }

    fn people(ids: &[i64]) -> skade::arrow_array::RecordBatch {
        let schema = Arc::new(Schema::new(vec![
            Field::new("id", DataType::Int64, false),
            Field::new("name", DataType::Utf8, true),
        ]));
        let names: Vec<String> = ids.iter().map(|i| format!("p{i}")).collect();
        RecordBatch::try_new(
            schema,
            vec![Arc::new(Int64Array::from(ids.to_vec())), Arc::new(StringArray::from(names))],
        )
        .unwrap()
    }

    /// The embedded warehouse's catalog (redb) admits one open handle at a
    /// time; the delivery opens its own every round. Retry while it is held,
    /// and keep every handle short-lived.
    async fn table(dir: &std::path::Path) -> skade::Table {
        let deadline = Instant::now() + Duration::from_secs(10);
        loop {
            match skade::Warehouse::open(dir).await {
                Ok(wh) => return wh.table("people").await.unwrap(),
                Err(e) if Instant::now() < deadline => {
                    let _ = e;
                    tokio::time::sleep(Duration::from_millis(5)).await;
                }
                Err(e) => panic!("warehouse stayed locked: {e}"),
            }
        }
    }

    async fn warehouse_with(commits: &[&[i64]]) -> (tempfile::TempDir, Vec<i64>) {
        let dir = tempfile::tempdir().unwrap();
        {
            let wh = skade::Warehouse::open(dir.path()).await.unwrap();
            wh.create_table("people", &people(&[]).schema()).await.unwrap();
        }
        let snaps = append(dir.path(), commits).await;
        (dir, snaps)
    }

    async fn append(dir: &std::path::Path, commits: &[&[i64]]) -> Vec<i64> {
        let mut t = table(dir).await;
        let mut snaps = Vec::new();
        for ids in commits {
            t.append(&[people(ids)]).await.unwrap();
            snaps.push(t.current_snapshot_id().unwrap());
        }
        snaps
    }

    async fn seq_of(dir: &std::path::Path, snap: i64) -> i64 {
        let t = table(dir).await;
        t.inner().metadata().snapshot_by_id(snap).unwrap().sequence_number()
    }

    fn delivery(dir: &std::path::Path, sink: Recording, opts: DeliveryOptions) -> IcebergDelivery<Recording> {
        let state = SyncState::open(dir.join("cursor.json")).unwrap();
        let loc = IcebergLocator::Warehouse { dir: dir.to_path_buf(), table: "people".into() };
        IcebergDelivery::new(loc, "test:db.people", sink, state, opts)
    }

    fn keyed() -> DeliveryOptions {
        DeliveryOptions { key_columns: vec!["id".into()], ..DeliveryOptions::default() }
    }

    #[tokio::test]
    async fn the_first_run_loads_the_current_snapshot_then_one_window_per_snapshot() {
        let (dir, snaps) = warehouse_with(&[&[1, 2], &[3], &[4, 5, 6]]).await;
        let sink = Recording::default();
        let mut d = delivery(dir.path(), sink.clone(), keyed());
        let o = d.round().await.unwrap();
        assert_eq!((o.windows, o.rows_loaded), (1, 6), "the initial load is one window: the live rows");
        assert_eq!(d.cursor(), Some(snaps[2]));
        let s0 = seq_of(dir.path(), snaps[2]).await;
        assert_eq!(
            sink.seen(),
            vec![Seen { from: None, to: snaps[2], upserts: 6, deletes: 0, sequence: Some(s0), keys: vec!["id".into()] }]
        );
        assert_eq!(*sink.probes.lock().unwrap(), 1, "probe runs once, before the first window");

        // Then strictly one window per new snapshot, oldest first, each with its
        // own sequence number.
        let more = append(dir.path(), &[&[7], &[8, 9]]).await;
        let o = d.round().await.unwrap();
        assert_eq!(o.windows, 2);
        let w = sink.seen();
        assert_eq!((w[1].from, w[1].to, w[1].upserts), (Some(snaps[2]), more[0], 1));
        assert_eq!((w[2].from, w[2].to, w[2].upserts), (Some(more[0]), more[1], 2));
        assert!(w[1].sequence < w[2].sequence && w[0].sequence < w[1].sequence, "{w:?}");
        assert_eq!(d.cursor(), Some(more[1]));

        // Caught up: nothing more, cursor unchanged.
        assert_eq!(d.round().await.unwrap().windows, 0);
    }

    #[tokio::test]
    async fn a_round_is_bounded_and_says_how_far_behind_it_is() {
        let (dir, _) = warehouse_with(&[&[0]]).await;
        let mut d = delivery(dir.path(), Recording::default(), DeliveryOptions { max_windows_per_round: 2, ..DeliveryOptions::default() });
        d.round().await.unwrap();
        let snaps = append(dir.path(), &[&[1], &[2], &[3]]).await;
        let o = d.round().await.unwrap();
        assert_eq!((o.windows, o.behind), (2, 1));
        assert_eq!(d.cursor(), Some(snaps[1]));
        let o = d.round().await.unwrap();
        assert_eq!((o.windows, o.behind), (1, 0));
        assert_eq!(d.cursor(), Some(snaps[2]));
    }

    #[tokio::test]
    async fn a_failed_window_holds_the_cursor_on_the_last_delivered_one() {
        let (dir, first) = warehouse_with(&[&[0]]).await;
        delivery(dir.path(), Recording::default(), DeliveryOptions::default()).round().await.unwrap();
        let snaps = append(dir.path(), &[&[1], &[2], &[3]]).await;
        let sink = Recording { fail_at: Some(snaps[1]), ..Recording::default() };
        let mut d = delivery(dir.path(), sink, DeliveryOptions::default());
        assert_eq!(d.cursor(), Some(first[0]));
        assert!(d.round().await.is_err());
        assert_eq!(d.cursor(), Some(snaps[0]), "window 1 landed, window 2 failed: cursor on 1");

        // The cursor survives a restart (it is on disk), and delivery resumes at 2.
        let sink = Recording::default();
        let mut d = delivery(dir.path(), sink.clone(), DeliveryOptions::default());
        assert_eq!(d.cursor(), Some(snaps[0]));
        d.round().await.unwrap();
        let w = sink.seen();
        assert_eq!(w.first().map(|w| (w.from, w.to)), Some((Some(snaps[0]), snaps[1])));
        assert_eq!(d.cursor(), Some(snaps[2]));
    }

    /// A v3 merge-on-read delete (a deletion vector — what Spark 4.1 + Iceberg
    /// 1.12 writes for MERGE INTO) is delivered as delete rows, keyed.
    #[tokio::test]
    async fn a_deletion_vector_is_delivered_as_keyed_deletes() {
        let (dir, _) = warehouse_with(&[&[1, 2, 3]]).await;
        let sink = Recording::default();
        let mut d = delivery(dir.path(), sink.clone(), keyed());
        d.round().await.unwrap();
        {
            let mut t = table(dir.path()).await;
            let f = t.live_data_files().await.unwrap().remove(0);
            t.delete_positions(&f, &[1]).await.unwrap(); // id 2
            t.append(&[people(&[2])]).await.unwrap(); // …re-inserted: an update
        }
        let o = d.round().await.unwrap();
        assert_eq!(o.windows, 2);
        let w = sink.seen();
        assert_eq!((w[1].upserts, w[1].deletes, w[1].keys.clone()), (0, 1, vec!["id".to_string()]));
        assert_eq!((w[2].upserts, w[2].deletes), (1, 0));
    }

    /// A compaction (`replace`) changes no row: an empty window, the cursor
    /// moves past it, nothing is read or loaded.
    #[tokio::test]
    async fn a_replace_snapshot_is_skipped_as_an_empty_window() {
        let (dir, _) = warehouse_with(&[&[1], &[2]]).await;
        let sink = Recording::default();
        let mut d = delivery(dir.path(), sink.clone(), DeliveryOptions::default());
        d.round().await.unwrap();
        let replaced = {
            let mut t = table(dir.path()).await;
            let files = t.live_data_files().await.unwrap();
            t.rewrite_files(&files, vec![vec![people(&[1, 2])]], None).await.unwrap();
            t.current_snapshot_id().unwrap()
        };
        let o = d.round().await.unwrap();
        assert_eq!(o.windows, 1);
        let w = sink.seen();
        assert_eq!((w[1].to, w[1].upserts, w[1].deletes), (replaced, 0, 0));
        assert_eq!(d.cursor(), Some(replaced));
    }

    /// The cursor's snapshot expired from the table: an explicit error, the
    /// cursor untouched, nothing delivered.
    #[tokio::test]
    async fn an_expired_cursor_requires_a_full_reload() {
        let (dir, _) = warehouse_with(&[&[1]]).await;
        let sink = Recording::default();
        let mut d = delivery(dir.path(), sink.clone(), DeliveryOptions::default());
        let mut c = d.state.get(&d.key);
        c.snapshot_id = Some(4242); // a snapshot this table does not have (any more)
        d.state.advance_staged(&d.key.clone(), c).await.unwrap();
        let err = d.round().await.unwrap_err();
        assert!(matches!(err, BifrostError::FullReloadRequired(_)), "{err}");
        assert_eq!(d.cursor(), Some(4242));
        assert!(sink.seen().is_empty());
    }

    /// A `Catalog` locator — any `dyn CatalogConnector` — delivers exactly what
    /// the warehouse locator does; here the same warehouse opened as `SkadeLocal`.
    #[tokio::test]
    async fn a_catalog_locator_delivers_what_the_warehouse_locator_does() {
        let (dir, _) = warehouse_with(&[&[1, 2], &[3]]).await;
        let (sa, sb) = (Recording::default(), Recording::default());
        let state = |n: &str| SyncState::open(dir.path().join(n)).unwrap();
        let wh = IcebergLocator::Warehouse { dir: dir.path().to_path_buf(), table: "people".into() };
        let cat = IcebergLocator::catalog(skade::connect::SkadeLocal::new(dir.path()), "main.people");
        let mut a = IcebergDelivery::new(wh, "test:db.people", sa.clone(), state("a.json"), keyed());
        let mut b = IcebergDelivery::new(cat, "test:db.people", sb.clone(), state("b.json"), keyed());
        assert_eq!(b.cursor_key(), format!("skade:{}#main.people -> test:db.people", dir.path().display()));
        a.round().await.unwrap();
        b.round().await.unwrap();
        append(dir.path(), &[&[4], &[5, 6]]).await;
        assert_eq!(a.round().await.unwrap().windows, 2);
        assert_eq!(b.round().await.unwrap().windows, 2);
        assert_eq!(sb.seen(), sa.seen());
        assert_eq!(b.cursor(), a.cursor());
    }

    /// The cursor key of a REST table is the form it had before the locator
    /// took any connector, so a running delivery keeps its cursor.
    #[cfg(feature = "skade-rest")]
    #[test]
    fn a_rest_table_keeps_its_cursor_key() {
        let rest = skade::connect::IcebergRest {
            uri: "https://minio:9000/_iceberg".into(),
            warehouse: Some("red".into()),
            prefix_or_ref: None,
            token: None,
            tls: None,
            s3: skade::S3Config::default(),
            read_only: true,
            sigv4: None,
            ..Default::default()
        };
        assert_eq!(IcebergLocator::catalog(rest, "bench.people").source_id(), "rest:https://minio:9000/_iceberg#red#bench.people");
    }

    /// A table behind a skade-katalog gRPC front is keyed by the front.
    #[cfg(feature = "skade-grpc")]
    #[test]
    fn a_grpc_table_is_keyed_by_its_front() {
        let grpc = skade::connect::SkadeGrpc::local_files("katalog:8182");
        assert_eq!(IcebergLocator::catalog(grpc, "db.people").source_id(), "grpc:http://katalog:8182#db.people");
    }

    /// Near-live: the poll loop picks up a snapshot committed while it runs,
    /// and stops cleanly between snapshots.
    #[tokio::test]
    async fn the_poll_loop_picks_up_a_new_snapshot() {
        let (dir, _) = warehouse_with(&[&[1]]).await;
        let sink = Recording::default();
        let mut d = delivery(dir.path(), sink.clone(), DeliveryOptions::default());
        let (stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>();
        let path = dir.path().to_path_buf();
        let watcher = sink.clone();
        let writer = async move {
            // Wait for the initial load, commit a new snapshot, wait for it.
            while watcher.seen().is_empty() {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
            let snap = append(&path, &[&[2, 3]]).await[0];
            let deadline = Instant::now() + Duration::from_secs(10);
            while !watcher.seen().iter().any(|w| w.to == snap) {
                assert!(Instant::now() < deadline, "the poll loop never delivered the new snapshot");
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
            stop_tx.send(()).unwrap();
            snap
        };
        let (run, snap) = tokio::join!(
            d.run(Duration::from_millis(20), Duration::from_millis(200), async {
                let _ = stop_rx.await;
            }),
            writer
        );
        run.unwrap();
        assert_eq!(d.cursor(), Some(snap));
        assert_eq!(sink.seen().last().map(|w| w.upserts), Some(2));
    }
}