faucet_source_oracle_cdc/

stream.rs

//! The LogMiner [`Source`]. A fetch cycle runs on one blocking thread that
//! owns the mining session and hands each committed transaction to the async
//! stream as its own page (bookmarked), so a transaction is never split and
//! the pipeline persists progress only after the sink confirms it.

use std::collections::HashMap;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use async_trait::async_trait;
use faucet_common_oracle::oracle::Connection;
use faucet_common_oracle::{
    NLS_SESSION_SQL, OraclePool, Side, TypeFamily, blocking, call_timeout, checkout, connect_pool,
    ora_code, ora_err,
};
use faucet_core::check::{CheckContext, CheckReport, Probe};
use faucet_core::{FaucetError, Source, SourceLag, Stream, StreamPage};
use serde_json::Value;
use tokio::sync::mpsc::Sender;

use crate::config::{OracleCdcSourceConfig, StartPosition};
use crate::lag::{LagPlan, commit_scn, from_age};
use crate::logs::{LogFile, oldest_available_scn, plan_log_files};
use crate::miner::{Assembler, LogRow, Miner, TableMeta, to_envelope};
use crate::sql::{
    ADD_LOGFILE_SQL, ARCHIVED_LOGS_SQL, CONTAINER_SQL, CURRENT_SCN_SQL, DatabaseLogging,
    END_LOGMNR_SQL, LOGFILE_ADD, LOGFILE_NEW, LoggingReport, ONLINE_LOGS_SQL, PDB_INCARNATION_SQL,
    POSITION_LOCAL_SQL, POSITION_SQL, RESETLOGS_SQL, SCN_AGE_SQL, START_LOGMNR_SQL,
    SUPPLEMENTAL_SQL, THREADS_SQL, columns_sql, contents_sql, flush_sql, flush_table_ddl, in_pdb,
    log_groups_sql, logging_report, needs_logfiles,
};
use crate::state::Position;

/// Oracle LogMiner change-data-capture source.
pub struct OracleCdcSource {
    shared: Arc<Shared>,
    pending: Mutex<Option<Position>>,
    emitted: Mutex<Option<u64>>,
    lag_warned: AtomicBool,
    /// Backend-call context (#880): the run's deadline, cancel token and
    /// round-trip recorder.
    calls: faucet_core::CallSlot,
}

struct Shared {
    config: OracleCdcSourceConfig,
    pool: OraclePool,
    tables: Vec<(String, String)>,
    state_key: String,
    add_logfiles: bool,
}

fn src_err(context: &str, e: &faucet_common_oracle::oracle::Error) -> FaucetError {
    ora_err(Side::Source, context, e)
}

/// The current PDB incarnation SCN when the session is in a pluggable
/// database. A view the user may not read degrades to no check, with a warning.
fn pdb_incarnation(conn: &Connection) -> Option<u64> {
    let read = || -> Result<Option<u64>, faucet_common_oracle::oracle::Error> {
        let (cdb, container): (String, String) = conn.query_row_as(CONTAINER_SQL, &[])?;
        if !in_pdb(&cdb, &container) {
            return Ok(None);
        }
        let (scn,): (Option<u64>,) = conn.query_row_as(PDB_INCARNATION_SQL, &[])?;
        Ok(scn)
    };
    read().unwrap_or_else(|e| {
        tracing::warn!(
            connector = "oracle-cdc",
            error = %e,
            "V$PDB_INCARNATION is not readable, so a PDB point-in-time recovery or flashback \
             cannot be detected; grant SELECT on it to enable the check"
        );
        None
    })
}

/// A LogMiner start failure; a missing log (ORA-01291) is lost redo.
fn start_error(e: &faucet_common_oracle::oracle::Error, from: u64, to: u64) -> FaucetError {
    classify_start_error(faucet_common_oracle::ora_code(e), &e.to_string(), from, to)
}

fn classify_start_error(code: Option<i32>, message: &str, from: u64, to: u64) -> FaucetError {
    match code {
        Some(1291) => FaucetError::Source(missing_redo_message(from, to)),
        _ => FaucetError::Source(format!("oracle start LogMiner: {message}")),
    }
}

fn missing_redo_message(from: u64, to: u64) -> String {
    format!(
        "oracle-cdc: the redo for SCN {from}..={to} is not available to LogMiner (ORA-01291 \
         missing log file). Run the database in ARCHIVELOG mode and keep archived logs until \
         they are mined; changes in a recycled range cannot be recovered — re-snapshot, then \
         restart capture"
    )
}

fn i64_scn(scn: u64) -> i64 {
    i64::try_from(scn).unwrap_or(i64::MAX)
}

fn table_binds(tables: &[(String, String)]) -> Vec<String> {
    tables
        .iter()
        .flat_map(|(o, t)| [o.clone(), t.clone()])
        .collect()
}

impl Shared {
    fn conn(&self) -> Result<Connection, FaucetError> {
        checkout(
            &self.pool,
            Side::Source,
            call_timeout(self.config.statement_timeout_secs),
        )
    }

    fn load_meta(&self, conn: &Connection) -> Result<TableMeta, FaucetError> {
        let binds = table_binds(&self.tables);
        let refs: Vec<&dyn faucet_common_oracle::oracle::sql_type::ToSql> =
            binds.iter().map(|b| b as _).collect();
        let rows = conn
            .query_as::<(String, String, String, String, Option<i64>)>(
                &columns_sql(self.tables.len()),
                &refs,
            )
            .map_err(|e| src_err("column metadata", &e))?;
        let mut meta = TableMeta::new();
        for r in rows {
            let (o, t, c, ty, scale) = r.map_err(|e| src_err("column metadata", &e))?;
            let family = TypeFamily::from_data_type(&ty, scale);
            let digits = matches!(family, TypeFamily::Timestamp | TypeFamily::TimestampTz)
                .then(|| scale.and_then(|s| u8::try_from(s).ok()))
                .flatten();
            meta.entry((o, t))
                .or_default()
                .insert(c, crate::miner::ColumnType { family, digits });
        }
        Ok(meta)
    }

    fn position_now(&self, conn: &Connection) -> Result<Position, FaucetError> {
        let (current, oldest): (u64, Option<u64>) = match conn.query_row_as(POSITION_SQL, &[]) {
            Ok(row) => row,
            Err(e) => {
                tracing::warn!(
                    error = %e,
                    "oracle-cdc: GV$TRANSACTION is not readable; looking for open \
                     transactions on the connected instance only. On RAC, grant \
                     SELECT_CATALOG_ROLE so a transaction open on another instance \
                     is captured from its start"
                );
                conn.query_row_as(POSITION_LOCAL_SQL, &[])
                    .map_err(|e| src_err("current SCN", &e))?
            }
        };
        Ok(Position::resume_from(current, oldest))
    }

    fn list_logs(&self, conn: &Connection, lo: u64, hi: u64) -> Result<Vec<LogFile>, FaucetError> {
        let mut out = Vec::new();
        let (lo, hi) = (i64_scn(lo), i64_scn(hi));
        let window: [&dyn faucet_common_oracle::oracle::sql_type::ToSql; 2] = [&lo, &hi];
        for (sql, archived, params) in [
            (ARCHIVED_LOGS_SQL, true, &window[..]),
            (ONLINE_LOGS_SQL, false, &[][..]),
        ] {
            let rows = conn
                .query_as::<(String, i64, i64, u64, u64)>(sql, params)
                .map_err(|e| src_err("redo log listing", &e))?;
            for r in rows {
                let (name, thread, sequence, first_change, next_change) =
                    r.map_err(|e| src_err("redo log listing", &e))?;
                out.push(LogFile {
                    name,
                    thread,
                    sequence,
                    first_change,
                    next_change,
                    archived,
                });
            }
        }
        Ok(out)
    }

    fn initial_position(&self, conn: &Connection) -> Result<Position, FaucetError> {
        match self.config.start_position {
            StartPosition::Current => self.position_now(conn),
            StartPosition::Earliest => {
                let logs = self.list_logs(conn, 0, u64::MAX)?;
                let oldest = oldest_available_scn(&logs).ok_or_else(|| {
                    FaucetError::Source("oracle-cdc: no redo logs are visible".into())
                })?;
                Ok(Position::at(oldest.saturating_sub(1), oldest))
            }
        }
    }

    fn flush(&self, conn: &Connection, scn: u64) -> Result<(), FaucetError> {
        conn.execute(
            &flush_sql(&self.config.flush_table),
            &[&i64_scn(scn), &i64_scn(scn)],
        )
        .map_err(|e| src_err("redo flush", &e))?;
        conn.commit().map_err(|e| src_err("redo flush", &e))
    }

    /// Mine `(from, to]`, feeding every row to `on_row`. The session is always
    /// ended, even when a row handler fails.
    fn mine(
        &self,
        conn: &Connection,
        from: u64,
        to: u64,
        on_row: &mut dyn FnMut(LogRow) -> Result<bool, FaucetError>,
    ) -> Result<(), FaucetError> {
        let logs = plan_log_files(&self.list_logs(conn, from, to)?, from + 1, to)
            .map_err(|m| FaucetError::Source(format!("oracle-cdc: {m}")))?;
        if self.add_logfiles {
            for (i, log) in logs.iter().enumerate() {
                let opt = if i == 0 { LOGFILE_NEW } else { LOGFILE_ADD };
                conn.execute(ADD_LOGFILE_SQL, &[&log.name, &opt])
                    .map_err(|e| src_err("add log file", &e))?;
            }
        }
        conn.execute(START_LOGMNR_SQL, &[&i64_scn(from + 1), &i64_scn(to)])
            .map_err(|e| start_error(&e, from + 1, to))?;
        let result = self.read_contents(conn, from, to, on_row);
        let _ = conn.execute(END_LOGMNR_SQL, &[]);
        result
    }

    fn read_contents(
        &self,
        conn: &Connection,
        from: u64,
        to: u64,
        on_row: &mut dyn FnMut(LogRow) -> Result<bool, FaucetError>,
    ) -> Result<(), FaucetError> {
        let mut binds: Vec<Box<dyn faucet_common_oracle::oracle::sql_type::ToSql>> =
            vec![Box::new(i64_scn(from)), Box::new(i64_scn(to))];
        for b in table_binds(&self.tables) {
            binds.push(Box::new(b));
        }
        let refs: Vec<&dyn faucet_common_oracle::oracle::sql_type::ToSql> =
            binds.iter().map(|b| b.as_ref()).collect();
        let mut stmt = conn
            .statement(&contents_sql(self.tables.len()))
            .fetch_array_size(1000)
            .build()
            .map_err(|e| src_err("mine", &e))?;
        type Raw = (
            u64,
            i64,
            String,
            Option<String>,
            Option<String>,
            Option<String>,
            i64,
            Option<String>,
            i64,
            Option<String>,
            Option<String>,
        );
        let rows = stmt
            .query_as::<Raw>(&refs)
            .map_err(|e| src_err("mine", &e))?;
        let mut assembler = Assembler::default();
        for r in rows {
            let (scn, op, xid, owner, table, redo, csf, row_id, rollback, ts, info) =
                r.map_err(|e| src_err("mine", &e))?;
            let row = LogRow {
                scn,
                op,
                xid,
                owner,
                table,
                redo: redo.unwrap_or_default(),
                row_id,
                rollback: rollback == 1,
                timestamp: ts,
                info,
            };
            if let Some(row) = assembler.push(row, csf == 1)
                && !on_row(row)?
            {
                return Ok(());
            }
        }
        Ok(())
    }

    /// One fetch cycle: mine windows until idle, emitting a page per committed
    /// transaction. Returns early when the consumer goes away.
    fn capture(
        &self,
        start: Option<Position>,
        per_txn: bool,
        tx: Sender<StreamPage>,
        calls: faucet_core::CallContext,
    ) -> Result<(), FaucetError> {
        let runtime = tokio::runtime::Handle::current();
        let conn = self.conn()?;
        conn.execute(NLS_SESSION_SQL, &[])
            .map_err(|e| src_err("session setup", &e))?;
        let mut meta = self.load_meta(&conn)?;
        let send = |page: StreamPage| tx.blocking_send(page).is_ok();
        let (resetlogs,): (u64,) = conn
            .query_row_as(RESETLOGS_SQL, &[])
            .map_err(|e| src_err("incarnation check", &e))?;
        let pdb = pdb_incarnation(&conn);
        let mut position = match start {
            Some(mut p) => {
                p.bind_incarnation(resetlogs)?;
                p.bind_pdb_incarnation(pdb)?;
                p
            }
            None => {
                let mut p = self.initial_position(&conn)?;
                p.bind_incarnation(resetlogs)?;
                p.bind_pdb_incarnation(pdb)?;
                if per_txn
                    && !send(StreamPage {
                        records: Vec::new(),
                        bookmark: Some(p.to_value()),
                    })
                {
                    return Ok(());
                }
                p
            }
        };
        let mut miner = Miner::new(self.config.on_unsupported, self.config.max_staged_records)
            .with_max_staged_total(self.config.max_staged_records_total);
        let mut from = position.restart_scn.saturating_sub(1);
        let mut agg: Vec<Value> = Vec::new();
        let mut agg_since = Instant::now();
        let mut last_activity = Instant::now();
        let mut clamped_since: Option<Instant> = None;
        let mut clamp_warned = false;
        loop {
            let current = self.position_now(&conn)?.commit_scn;
            self.flush(&conn, current)?;
            let (threads, floor): (u64, Option<u64>) = conn
                .query_row_as(THREADS_SQL, &[])
                .map_err(|e| src_err("redo thread check", &e))?;
            match rac_clamp_lag(current, threads, floor) {
                Some(lag) if floor.is_some_and(|f| f <= from) => {
                    let since = *clamped_since.get_or_insert_with(Instant::now);
                    if !clamp_warned && since.elapsed() >= RAC_CLAMP_WARN_AFTER {
                        clamp_warned = true;
                        tracing::warn!(
                            open_threads = threads,
                            scns_held_back = lag,
                            held_secs = since.elapsed().as_secs(),
                            "oracle-cdc: capture is waiting for the lowest RAC thread checkpoint \
                             (V$THREAD.CHECKPOINT_CHANGE#), which advances only on a log switch \
                             or checkpoint — a quiet instance holds back every instance's \
                             changes. Set ARCHIVE_LAG_TARGET (e.g. 60) to bound the delay"
                        );
                    }
                }
                _ => clamped_since = None,
            }
            let to = window_end(current, from, self.config.max_scn_window, threads, floor);
            let reachable = window_end(current, from, u64::MAX, threads, floor);
            let mut captured = false;
            if to > from {
                let before = position.clone();
                let mut open = true;
                self.mine(&conn, from, to, &mut |row| {
                    let applied = miner.apply(&row)?;
                    // Reload at the DDL row so later commits in this window are typed by
                    // the new columns, not the ones loaded before it.
                    if !miner.take_ddl_tables().is_empty() {
                        meta = self.load_meta(&conn)?;
                    }
                    let Some(txn) = applied else {
                        return Ok(true);
                    };
                    if txn.events.is_empty() || position.already_emitted(txn.commit_scn, &txn.xid) {
                        return Ok(true);
                    }
                    let records: Vec<Value> = txn
                        .events
                        .iter()
                        .map(|e| to_envelope(e, &txn, &meta))
                        .collect();
                    position.record_commit(txn.commit_scn, &txn.xid, miner.oldest_open());
                    captured = true;
                    if per_txn {
                        open = send(StreamPage {
                            records,
                            bookmark: Some(position.to_value()),
                        });
                    } else {
                        agg.extend(records);
                        if agg_page_due(agg.len(), agg_since.elapsed(), self.config.idle_timeout) {
                            open = send(StreamPage {
                                records: std::mem::take(&mut agg),
                                bookmark: Some(position.to_value()),
                            });
                            agg_since = Instant::now();
                        }
                    }
                    Ok(open)
                })?;
                if !open {
                    return Ok(());
                }
                position.advance_to(to, miner.oldest_open());
                from = to;
                if per_txn
                    && position != before
                    && !send(StreamPage {
                        records: Vec::new(),
                        bookmark: Some(position.to_value()),
                    })
                {
                    return Ok(());
                }
            }
            if keeps_cycle_alive(captured, to, reachable) {
                last_activity = Instant::now();
            }
            if last_activity.elapsed() >= self.config.idle_timeout {
                break;
            }
            if to >= reachable {
                // The poll interval ends early on the run's cancel and deadline.
                runtime.block_on(calls.wait(self.config.poll_interval))?;
            }
        }
        if !per_txn {
            send(StreamPage {
                records: agg,
                bookmark: Some(position.to_value()),
            });
        }
        tracing::info!(state_key = %self.state_key, "oracle-cdc fetch cycle complete");
        Ok(())
    }

    fn check_supplemental(&self, conn: &Connection) -> Result<LoggingReport, FaucetError> {
        let (min, pk, all): (String, String, String) = conn
            .query_row_as(SUPPLEMENTAL_SQL, &[])
            .map_err(|e| src_err("supplemental logging check", &e))?;
        let binds = table_binds(&self.tables);
        let refs: Vec<&dyn faucet_common_oracle::oracle::sql_type::ToSql> =
            binds.iter().map(|b| b as _).collect();
        let groups = conn
            .query_as::<(String, String, String)>(&log_groups_sql(self.tables.len()), &refs)
            .map_err(|e| src_err("supplemental logging check", &e))?
            .collect::<Result<Vec<_>, _>>()
            .map_err(|e| src_err("supplemental logging check", &e))?;
        Ok(logging_report(
            &DatabaseLogging { min, pk, all },
            &self.tables,
            &groups,
        ))
    }
}

/// The end of the next mining window: `current`, at most `max_window` past
/// `from`, and — with several open redo threads (RAC) — no later than the
/// lowest thread checkpoint, since the flush reaches only the local instance
/// and a commit on another thread above that point may not be in the redo yet.
pub(crate) fn window_end(
    current: u64,
    from: u64,
    max_window: u64,
    open_threads: u64,
    checkpoint_floor: Option<u64>,
) -> u64 {
    let end = current.min(from.saturating_add(max_window));
    match checkpoint_floor {
        Some(floor) if open_threads > 1 => end.min(floor),
        _ => end,
    }
}

/// How long the RAC checkpoint floor may hold the window back before a warning.
pub(crate) const RAC_CLAMP_WARN_AFTER: Duration = Duration::from_secs(60);

/// The SCNs the RAC checkpoint floor is holding back from `current`, when it
/// is the limit (several open threads and a floor below `current`).
pub(crate) fn rac_clamp_lag(current: u64, open_threads: u64, floor: Option<u64>) -> Option<u64> {
    match floor {
        Some(f) if open_threads > 1 && f < current => Some(current - f),
        _ => None,
    }
}

/// Records a `batch_size: 0` cycle buffers before it emits a page anyway.
pub(crate) const AGG_PAGE_RECORDS: usize = 100_000;

/// Whether a `batch_size: 0` cycle must emit what it holds now: at
/// [`AGG_PAGE_RECORDS`] records, or once records have waited `idle_timeout` —
/// a database that never goes idle would otherwise grow the buffer without
/// bound and never persist a bookmark.
pub(crate) fn agg_page_due(len: usize, waited: Duration, idle_timeout: Duration) -> bool {
    len >= AGG_PAGE_RECORDS || (len > 0 && waited >= idle_timeout)
}

impl OracleCdcSource {
    /// Validate, connect, create the flush table and run the preflight: the
    /// captured tables must exist and minimal supplemental logging must be on.
    pub async fn new(config: OracleCdcSourceConfig) -> Result<Self, FaucetError> {
        config.validate()?;
        let pool = connect_pool(&config.connection, config.max_connections).await?;
        let tables = config.table_pairs();
        let state_key = config.resolved_state_key();
        let probe = Shared {
            config: config.clone(),
            pool: pool.clone(),
            tables: tables.clone(),
            state_key: state_key.clone(),
            add_logfiles: true,
        };
        let add_logfiles = blocking(move || {
            let conn = probe.conn()?;
            let (cdb, container): (String, String) = conn
                .query_row_as(CONTAINER_SQL, &[])
                .map_err(|e| src_err("container detection", &e))?;
            let meta = probe.load_meta(&conn)?;
            let missing: Vec<String> = probe
                .tables
                .iter()
                .filter(|t| !meta.contains_key(*t))
                .map(|(o, t)| format!("{o}.{t}"))
                .collect();
            if !missing.is_empty() {
                return Err(FaucetError::Config(format!(
                    "oracle-cdc: tables {missing:?} are not visible to this user (check the \
                     names' case and SELECT grants)"
                )));
            }
            let report = probe.check_supplemental(&conn)?;
            if !report.fatal.is_empty() {
                return Err(FaucetError::Config(format!(
                    "oracle-cdc: {}",
                    report.fatal.join("; ")
                )));
            }
            if !report.partial.is_empty() {
                tracing::warn!(
                    fix = ?report.partial,
                    "oracle-cdc: all-column supplemental logging is off, so update events carry \
                     only key and changed columns"
                );
            }
            conn.execute(&flush_table_ddl(&probe.config.flush_table), &[])
                .map_err(|e| src_err("create flush table", &e))?;
            Ok(needs_logfiles(&cdb, &container))
        })
        .await?;
        Ok(Self {
            shared: Arc::new(Shared {
                config,
                pool,
                tables,
                state_key,
                add_logfiles,
            }),
            pending: Mutex::new(None),
            emitted: Mutex::new(None),
            lag_warned: AtomicBool::new(false),
            calls: faucet_core::CallSlot::new(faucet_core::CallContext::new().without_timeout()),
        })
    }

    fn set_emitted(&self, scn: u64) {
        if let Ok(mut g) = self.emitted.lock() {
            *g = Some(scn);
        }
    }

    /// The commit SCN capture has reached: the last emitted bookmark, else
    /// the applied start bookmark.
    fn lag_position(&self) -> Option<u64> {
        self.emitted.lock().ok().and_then(|g| *g).or_else(|| {
            self.pending
                .lock()
                .ok()
                .and_then(|g| g.as_ref().map(|p| p.commit_scn))
        })
    }

    async fn measure_lag(&self, position: u64) -> Result<Option<SourceLag>, FaucetError> {
        let shared = self.shared.clone();
        let measured = blocking(move || {
            let conn = shared.conn()?;
            let current: u64 = conn
                .query_row_as(CURRENT_SCN_SQL, &[])
                .map_err(|e| src_err("lag current SCN", &e))?;
            match crate::lag::plan(position, current) {
                LagPlan::CaughtUp => Ok(Some(SourceLag::seconds(0.0))),
                LagPlan::Measure { current, position } => {
                    match conn
                        .query_row_as::<f64>(SCN_AGE_SQL, &[&i64_scn(current), &i64_scn(position)])
                    {
                        Ok(age) => Ok(Some(SourceLag::seconds(age))),
                        Err(e) => {
                            from_age(Err(ora_code(&e))).map_err(|_| src_err("lag SCN age", &e))
                        }
                    }
                }
            }
        });
        self.calls.call_once("query", measured).await
    }

    fn pages(
        &self,
        per_txn: bool,
    ) -> Pin<Box<dyn Stream<Item = Result<StreamPage, FaucetError>> + Send + '_>> {
        let start = self.pending.lock().expect("pending mutex").take();
        if let Some(p) = &start {
            self.set_emitted(p.commit_scn);
        }
        let shared = self.shared.clone();
        let calls = self.calls.get();
        Box::pin(async_stream::try_stream! {
            let (tx, mut rx) = tokio::sync::mpsc::channel::<StreamPage>(4);
            let capture_calls = calls.clone();
            let handle =
                tokio::task::spawn_blocking(move || shared.capture(start, per_txn, tx, capture_calls));
            // The run's deadline and cancel token end the wait for the next
            // page with an error (#880); dropping the receiver stops capture.
            let steps = calls.with_roundtrip_metering(false);
            while let Some(page) = steps.call_once("poll", async { Ok(rx.recv().await) }).await? {
                if let Some(scn) = page.bookmark.as_ref().and_then(commit_scn) {
                    self.set_emitted(scn);
                }
                yield page;
            }
            handle
                .await
                .map_err(|e| FaucetError::Source(format!("oracle-cdc capture task failed: {e}")))??;
        })
    }
}

#[async_trait]
impl Source for OracleCdcSource {
    fn set_call_context(&self, ctx: faucet_core::CallContext) {
        self.calls.install(ctx);
    }

    fn set_roundtrip_recorder(
        &self,
        recorder: std::sync::Arc<faucet_core::observability::RoundtripRecorder>,
    ) {
        self.calls.install_recorder(recorder);
    }

    async fn fetch_with_context(
        &self,
        _ctx: &HashMap<String, Value>,
    ) -> Result<Vec<Value>, FaucetError> {
        use futures::StreamExt;
        let mut pages = self.pages(false);
        let mut all = Vec::new();
        while let Some(page) = pages.next().await {
            all.extend(page?.records);
        }
        Ok(all)
    }

    fn stream_pages<'a>(
        &'a self,
        _ctx: &'a HashMap<String, Value>,
        _batch_size: usize,
    ) -> Pin<Box<dyn Stream<Item = Result<StreamPage, FaucetError>> + Send + 'a>> {
        self.pages(self.shared.config.batch_size != 0)
    }

    fn config_schema(&self) -> Value {
        serde_json::to_value(faucet_core::schema_for!(OracleCdcSourceConfig))
            .expect("schema serialization")
    }

    fn state_key(&self) -> Option<String> {
        Some(self.shared.state_key.clone())
    }

    fn legacy_state_keys(&self) -> Vec<String> {
        self.shared.config.legacy_state_keys()
    }

    async fn apply_start_bookmark(&self, bookmark: Value) -> Result<(), FaucetError> {
        *self.pending.lock().expect("pending mutex") = Some(Position::from_value(&bookmark)?);
        Ok(())
    }

    /// The current SCN (reaching back to open transactions), for anchoring CDC
    /// before a snapshot.
    async fn capture_resume_position(&self) -> Result<Option<Value>, FaucetError> {
        let shared = self.shared.clone();
        self.calls
            .call_once(
                "query",
                blocking(move || {
                    let conn = shared.conn()?;
                    Ok(Some(shared.position_now(&conn)?.to_value()))
                }),
            )
            .await
    }

    /// Seconds of commit time between the capture position and the current
    /// SCN (`SCN_TO_TIMESTAMP`); `None` before a position exists or when the
    /// position is older than the SCN-to-time mapping (ORA-08181). A failed
    /// query is logged once and reported as no lag.
    async fn lag(&self) -> Result<Option<SourceLag>, FaucetError> {
        let Some(position) = self.lag_position() else {
            return Ok(None);
        };
        match self.measure_lag(position).await {
            Ok(lag) => Ok(lag),
            Err(e) => {
                if !self.lag_warned.swap(true, Ordering::Relaxed) {
                    tracing::warn!(error = %e, "oracle-cdc: lag query failed; lag is not reported (logged once)");
                }
                Ok(None)
            }
        }
    }

    fn capabilities(&self) -> faucet_core::Capabilities {
        faucet_core::Capabilities::of_source(self)
            .to_builder()
            .reports_lag(true)
            .table_routing(true)
            .read_side_effects(true)
            .build()
    }

    fn supports_exactly_once(&self) -> bool {
        true
    }

    fn record_table(&self, record: &Value) -> Option<String> {
        schema_table(record)
    }

    fn position_le(&self, a: &Value, b: &Value) -> Option<bool> {
        crate::state::position_le(a, b)
    }

    fn record_covered(&self, record: &Value, position: &Value) -> Option<bool> {
        crate::state::record_covered(record, position)
    }

    fn connector_name(&self) -> &'static str {
        "oracle-cdc"
    }

    fn dataset_uri(&self) -> String {
        format!(
            "{}?tables={}",
            self.shared.config.connection.display_uri(),
            self.shared.config.tables.join(",")
        )
    }

    async fn check(&self, ctx: &CheckContext) -> Result<CheckReport, FaucetError> {
        let start = Instant::now();
        let shared = self.shared.clone();
        let run = blocking(move || {
            let conn = shared.conn()?;
            let report = shared.check_supplemental(&conn)?;
            let logs = shared.list_logs(&conn, 0, u64::MAX)?;
            Ok((report, logs.len()))
        });
        let probes = match tokio::time::timeout(ctx.timeout, run).await {
            Err(_) => vec![Probe::fail_hint(
                "connection",
                start.elapsed(),
                "timed out",
                "the database did not respond within the check timeout",
            )],
            Ok(Err(e)) => vec![Probe::fail_hint(
                "connection",
                start.elapsed(),
                e.to_string(),
                "check the connection settings and that the user can read V$DATABASE, V$LOG, \
                 V$ARCHIVED_LOG and ALL_LOG_GROUPS (grant SELECT_CATALOG_ROLE, LOGMINING)",
            )],
            Ok(Ok((report, logs))) => {
                let supplemental = if report.fatal.is_empty() {
                    Probe::pass("supplemental-logging", start.elapsed())
                } else {
                    Probe::fail_hint(
                        "supplemental-logging",
                        start.elapsed(),
                        "supplemental logging does not cover the captured tables",
                        report.fatal.join("; "),
                    )
                };
                let images = if report.partial.is_empty() {
                    Probe::pass("full-row-images", start.elapsed())
                } else {
                    Probe::skip(
                        "full-row-images",
                        format!(
                            "updates will carry only key and changed columns; to capture full \
                             rows run: {}",
                            report.partial.join(" ")
                        ),
                    )
                };
                let redo = if logs > 0 {
                    Probe::pass("redo-logs", start.elapsed())
                } else {
                    Probe::fail_hint(
                        "redo-logs",
                        start.elapsed(),
                        "no redo logs are visible",
                        "grant SELECT on V_$LOG, V_$LOGFILE and V_$ARCHIVED_LOG",
                    )
                };
                vec![
                    Probe::pass("connection", start.elapsed()),
                    supplemental,
                    images,
                    redo,
                ]
            }
        };
        Ok(CheckReport { probes })
    }
}

/// `OWNER.TABLE` of a change envelope, the name the `oracle` source's
/// discovery reports for the same table.
fn schema_table(record: &Value) -> Option<String> {
    let schema = record.get("schema")?.as_str()?;
    let table = record.get("table")?.as_str()?;
    Some(format!("{schema}.{table}"))
}

/// Whether a window resets the idle clock: it captured a commit, or mining has
/// not yet reached the current SCN. Each cycle restarts at `restart_scn` — pinned
/// by the oldest open transaction anywhere in the database — and re-mines redo
/// it already emitted; counting that catch-up as idle broke the cycle before it
/// reached new redo, every cycle, so capture never progressed (SQL-29).
fn keeps_cycle_alive(captured: bool, to: u64, current: u64) -> bool {
    captured || to < current
}

#[cfg(test)]
mod tests {

    #[test]
    fn catching_up_is_never_idle() {
        assert!(keeps_cycle_alive(false, 100, 200), "re-mining old redo");
        assert!(keeps_cycle_alive(true, 200, 200), "captured a commit");
        assert!(!keeps_cycle_alive(false, 200, 200), "caught up and quiet");
        assert!(!keeps_cycle_alive(false, 250, 200));
    }

    #[test]
    fn window_end_respects_the_rac_checkpoint_floor() {
        assert_eq!(window_end(1000, 100, 50, 1, Some(10)), 150);
        assert_eq!(window_end(120, 100, 50, 1, Some(10)), 120);
        assert_eq!(window_end(1000, 100, 50, 2, Some(130)), 130);
        assert_eq!(window_end(1000, 100, 50, 2, Some(500)), 150);
        assert_eq!(window_end(1000, 100, 50, 2, None), 150);
        assert_eq!(window_end(1000, 100, u64::MAX, 3, Some(700)), 700);
    }

    #[test]
    fn the_rac_floor_lag_is_reported_only_when_it_limits_the_window() {
        assert_eq!(rac_clamp_lag(1000, 2, Some(400)), Some(600));
        assert_eq!(rac_clamp_lag(1000, 1, Some(400)), None);
        assert_eq!(rac_clamp_lag(1000, 2, Some(1000)), None);
        assert_eq!(rac_clamp_lag(1000, 2, None), None);
    }

    #[test]
    fn agg_pages_are_bounded() {
        let idle = std::time::Duration::from_secs(30);
        assert!(!agg_page_due(0, idle * 2, idle));
        assert!(!agg_page_due(10, idle / 2, idle));
        assert!(agg_page_due(10, idle, idle));
        assert!(agg_page_due(
            AGG_PAGE_RECORDS,
            std::time::Duration::ZERO,
            idle
        ));
    }

    #[test]
    fn routes_by_owner_table() {
        assert_eq!(
            schema_table(&serde_json::json!({"schema": "APP", "table": "ORDERS"})),
            Some("APP.ORDERS".into())
        );
        assert_eq!(schema_table(&serde_json::json!({"table": "ORDERS"})), None);
    }
    use super::*;

    #[test]
    fn helpers() {
        assert_eq!(i64_scn(5), 5);
        assert_eq!(i64_scn(u64::MAX), i64::MAX);
        assert_eq!(
            table_binds(&[("A".into(), "T".into())]),
            vec!["A".to_string(), "T".to_string()]
        );
        assert!(missing_redo_message(1, 2).contains("SCN 1..=2"));
        assert!(
            classify_start_error(Some(1291), "x", 3, 4)
                .to_string()
                .contains("SCN 3..=4")
        );
        let other = classify_start_error(Some(1017), "ORA-01017", 3, 4).to_string();
        assert!(
            other.contains("ORA-01017") && !other.contains("ARCHIVELOG"),
            "{other}"
        );
    }
}