rindle_replica/

maintenance.rs

//! Online database maintenance — bounded, non-blocking upkeep the server runs on a timer
//! between commits, so planner stats stay semi-fresh and the file does not bloat under churn,
//! without ever taking a long lock the way a full `VACUUM` / `ANALYZE` would.
//!
//! A maintenance pass ([`run`]) does up to three independently-gated things on one connection, in
//! this order:
//!
//! 1. **`PRAGMA optimize`** — refresh stale `sqlite_stat*` so the cost model keeps choosing good
//!    plans on a write-heavy replica. `optimize` self-gates per table on a change counter (most
//!    ticks are near-no-ops), and the [`ANALYSIS_LIMIT`] set at connection open caps how many
//!    rows the ANALYZE it *does* run samples per index — so a refresh stays sub-millisecond
//!    regardless of table size. Its only writes land in `sqlite_stat*`, which the CDC capture
//!    hook skips, so running it on the hooked writer never fabricates a change event.
//! 2. **`PRAGMA incremental_vacuum(N)`** — return up to `N` freelist pages to the OS, but only once
//!    at least `freelist_threshold_pages` have accumulated, and only on a database already on
//!    incremental auto-vacuum (see below). Bounded work, no exclusive lock — unlike a full `VACUUM`
//!    (a whole-file rewrite under an exclusive lock).
//! 3. **`PRAGMA wal_checkpoint(PASSIVE)`** — fold WAL frames (including the vacuum's truncation)
//!    back into the main file, shrinking it on disk and bounding WAL growth. PASSIVE never blocks a
//!    reader or the writer; a long-lived reader just defers the rest to a later tick.
//!
//! ## The vacuum step is opt-in
//! The runtime does not change `auto_vacuum` at open. A database with `auto_vacuum=NONE`
//! skips this step and reuses freed pages internally, without returning them to the OS.
//! Converting an existing database to incremental auto-vacuum requires a `VACUUM` file
//! rewrite. If that behavior is needed, close the runtime and configure the database
//! out of band with `PRAGMA auto_vacuum=INCREMENTAL; VACUUM;` before reopening it.
//! The `optimize` and checkpoint steps run regardless. See
//! [`Cluster::maintain`](crate::Cluster::maintain).

use rusqlite::Connection;

use crate::ReplicaError;

/// Rows `ANALYZE` / `PRAGMA optimize` samples per index. Set as `PRAGMA analysis_limit` on every
/// connection at open (the connection ritual lives with the apply plane, which owns the const —
/// design 309) so a stats refresh on the live writer is bounded to a few hundred rows per
/// index — fast enough to slot between commits — instead of a full index scan. SQLite's own
/// recommended value for "good enough" stats; stale stats only ever revert a plan toward the
/// un-analyzed choice, never worse than no stats.
pub use crate::apply::ANALYSIS_LIMIT;

/// Which steps a maintenance pass performs and the thresholds that gate them. [`Default`] is the
/// server's tuning: optimize on, reclaim up to 1000 freelist pages once ≥1000 have built up
/// (~4 MiB at a 4 KiB page), passive checkpoint on.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct MaintenanceOptions {
    /// Run `PRAGMA optimize` to refresh stale planner statistics.
    pub run_optimize: bool,
    /// Max freelist pages to return to the OS per pass (the `N` in `incremental_vacuum(N)`).
    /// `0` disables the vacuum step.
    pub incremental_vacuum_pages: u32,
    /// Only run the vacuum step once the freelist has at least this many pages — so a quiet
    /// database is left untouched and the step is not churned for a handful of pages.
    pub freelist_threshold_pages: u32,
    /// Take a `PRAGMA wal_checkpoint(PASSIVE)` to bound WAL growth.
    pub wal_checkpoint: bool,
}

impl Default for MaintenanceOptions {
    fn default() -> MaintenanceOptions {
        MaintenanceOptions {
            run_optimize: true,
            incremental_vacuum_pages: 1000,
            freelist_threshold_pages: 1000,
            wal_checkpoint: true,
        }
    }
}

/// The three integers a `PRAGMA wal_checkpoint` returns. Moved to the apply plane
/// (`ApplyStore::checkpoint_truncate` is the TRUNCATE flavor's home — design 309);
/// re-exported here so `rindle_replica::WalCheckpoint` and the maintenance report are
/// unchanged.
pub use crate::apply::WalCheckpoint;

/// What a maintenance pass actually did — returned for observability/tests. A pass that ran while
/// a write transaction was open reports `skipped`.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct MaintenanceReport {
    /// The pass was a no-op because a write transaction was open (best-effort: maintenance never
    /// disturbs an in-flight txn). All other fields are at their defaults.
    pub skipped: bool,
    /// `PRAGMA optimize` ran.
    pub optimized: bool,
    /// Freelist page count observed before the vacuum step (0 if the step did not inspect it).
    pub freelist_before: u64,
    /// Pages actually returned to the OS this pass.
    pub pages_reclaimed: u64,
    /// The vacuum step ran (the freelist was over threshold on an incremental-vacuum database).
    pub vacuum_ran: bool,
    /// The checkpoint result, if the checkpoint step ran.
    pub wal_checkpoint: Option<WalCheckpoint>,
}

impl MaintenanceReport {
    pub(crate) fn skipped() -> MaintenanceReport {
        MaintenanceReport {
            skipped: true,
            ..MaintenanceReport::default()
        }
    }

    /// Whether the pass performed any work (ran and did at least one step).
    pub fn did_work(&self) -> bool {
        !self.skipped && (self.optimized || self.vacuum_ran || self.wal_checkpoint.is_some())
    }
}

/// Run a **full** `ANALYZE` on `conn` (deep planner statistics), temporarily lifting the
/// connection's [`ANALYSIS_LIMIT`] so SQLite builds the per-value `sqlite_stat4` histogram — not
/// just the `sqlite_stat1` row counts. The bounded limit is restored afterward (always, even if
/// `ANALYZE` fails) so the between-commits maintenance tick stays fast.
///
/// Why this exists (GitHub #68): with a non-zero `analysis_limit` set on every connection,
/// `ANALYZE` / `PRAGMA optimize` write `sqlite_stat1` but **zero** `sqlite_stat4` rows, and at the
/// time the planner needed `stat4` to make an `ORDER BY … LIMIT n` displacement re-fetch's
/// OR-of-ranges start bound seekable at all — without it, a full covering-index scan per insert.
///
/// **That is no longer the reason.** `rindle-sqlite`'s query builder now lifts a redundant
/// sargable bound on the leading sort column out of that OR-of-ANDs (`f6423231`, 2026-07-01), so
/// the displacement seeks with no `stat4` present — measured, and pinned by
/// `rindle-replica/tests/limit_push_cluster.rs`. `stat4` still sharpens the cost model generally
/// (see the planner designs) and `rindle-loadgen` still calls this, but do NOT reintroduce a claim
/// that the LIMIT push depends on it.
///
/// `stat4` is a slow-changing value-distribution histogram, so it is built deliberately/rarely
/// (after a bulk load, or on demand) rather than on every cheap maintenance tick. The caller must
/// hold no open write transaction.
pub(crate) fn analyze_full(conn: &Connection) -> Result<(), ReplicaError> {
    conn.execute_batch("PRAGMA analysis_limit = 0")
        .map_err(|e| ReplicaError::sqlite("analyze_full lift analysis_limit", e))?;
    // Restore the bounded limit no matter what ANALYZE does, so a later `PRAGMA optimize` tick
    // stays sub-millisecond regardless of table size.
    let analyzed = conn.execute_batch("ANALYZE");
    let restored = conn.execute_batch(&format!("PRAGMA analysis_limit = {ANALYSIS_LIMIT}"));
    analyzed.map_err(|e| ReplicaError::sqlite("analyze_full ANALYZE", e))?;
    restored.map_err(|e| ReplicaError::sqlite("analyze_full restore analysis_limit", e))?;
    Ok(())
}

/// Run one maintenance pass on `conn` per `opts`. The connection must be in autocommit (no open
/// transaction) — the callers ([`Cluster::maintain`](crate::Cluster::maintain) /
/// [`Db::maintain`](crate::Db::maintain)) guarantee this by skipping when a write txn is open.
pub(crate) fn run(
    conn: &Connection,
    opts: &MaintenanceOptions,
) -> Result<MaintenanceReport, ReplicaError> {
    let mut report = MaintenanceReport::default();

    if opts.run_optimize {
        conn.execute_batch("PRAGMA optimize")
            .map_err(|e| ReplicaError::sqlite("PRAGMA optimize", e))?;
        report.optimized = true;
    }

    if opts.incremental_vacuum_pages > 0 {
        report.pages_reclaimed = incremental_vacuum(conn, opts, &mut report)?;
    }

    // Checkpoint last so it folds the vacuum's truncation (and the tick's WAL frames) back into the
    // main file, shrinking it on disk and bounding WAL growth. PASSIVE never blocks a reader or the
    // writer; a long-lived reader just defers the rest to a later tick.
    if opts.wal_checkpoint {
        let (busy, log, ckpt): (i64, i64, i64) = conn
            .query_row("PRAGMA wal_checkpoint(PASSIVE)", [], |r| {
                Ok((r.get(0)?, r.get(1)?, r.get(2)?))
            })
            .map_err(|e| ReplicaError::sqlite("PRAGMA wal_checkpoint", e))?;
        report.wal_checkpoint = Some(WalCheckpoint {
            busy: busy != 0,
            log_frames: log,
            checkpointed_frames: ckpt,
        });
    }

    Ok(report)
}

/// Reclaim up to `incremental_vacuum_pages` freelist pages, gated on the freelist being over
/// `freelist_threshold_pages` and the database being on incremental auto-vacuum. Returns the page
/// count actually reclaimed (and records `freelist_before` / `vacuum_ran` on `report`).
fn incremental_vacuum(
    conn: &Connection,
    opts: &MaintenanceOptions,
    report: &mut MaintenanceReport,
) -> Result<u64, ReplicaError> {
    // auto_vacuum: 0 = NONE, 1 = FULL, 2 = INCREMENTAL. `incremental_vacuum` only does anything on a
    // non-NONE database. The runtime does not convert an existing database's page layout at open;
    // an operator configures this out of band. The default mode returns without work.
    let mode: i64 = conn
        .query_row("PRAGMA auto_vacuum", [], |r| r.get(0))
        .map_err(|e| ReplicaError::sqlite("PRAGMA auto_vacuum", e))?;
    if mode == 0 {
        return Ok(0);
    }
    let before: i64 = conn
        .query_row("PRAGMA freelist_count", [], |r| r.get(0))
        .map_err(|e| ReplicaError::sqlite("PRAGMA freelist_count", e))?;
    report.freelist_before = before.max(0) as u64;
    if report.freelist_before < u64::from(opts.freelist_threshold_pages) {
        return Ok(0);
    }
    // `PRAGMA incremental_vacuum(N)` emits ONE result row per freed page and must be stepped to
    // completion — `execute_batch` would step only once (freeing a single page). Counting the rows
    // is the exact reclaimed-page count, and the `(N)` bound caps the work per pass.
    let mut stmt = conn
        .prepare(&format!(
            "PRAGMA incremental_vacuum({})",
            opts.incremental_vacuum_pages
        ))
        .map_err(|e| ReplicaError::sqlite("prepare incremental_vacuum", e))?;
    let mut rows = stmt
        .query([])
        .map_err(|e| ReplicaError::sqlite("incremental_vacuum", e))?;
    let mut reclaimed = 0u64;
    while rows
        .next()
        .map_err(|e| ReplicaError::sqlite("incremental_vacuum step", e))?
        .is_some()
    {
        reclaimed += 1;
    }
    report.vacuum_ran = true;
    Ok(reclaimed)
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Open a lone WAL connection (the D5 default mode) with the same auto-vacuum/analysis
    /// pragmas the engine sets, so the reclaim mechanic can be tested without any worker
    /// holding a read mark on the freelist.
    fn open(path: &std::path::Path) -> Connection {
        let conn = Connection::open(path).unwrap();
        conn.execute_batch(&format!(
            "PRAGMA auto_vacuum=INCREMENTAL;\n\
             PRAGMA journal_mode=wal;\n\
             PRAGMA analysis_limit={ANALYSIS_LIMIT};"
        ))
        .unwrap();
        conn
    }

    #[test]
    fn incremental_vacuum_drains_the_freelist_when_stepped() {
        let dir = tempfile::tempdir().unwrap();
        let conn = open(&dir.path().join("m.db"));
        conn.execute_batch("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)")
            .unwrap();
        let pad = "x".repeat(200);
        for id in 0..5_000i64 {
            conn.execute(
                "INSERT INTO t(id, v) VALUES (?, ?)",
                rusqlite::params![id, format!("{pad}-{id}")],
            )
            .unwrap();
        }
        conn.execute("DELETE FROM t", []).unwrap();
        let freelist: i64 = conn
            .query_row("PRAGMA freelist_count", [], |r| r.get(0))
            .unwrap();
        assert!(
            freelist > 50,
            "sanity: mass delete built a freelist ({freelist})"
        );

        let opts = MaintenanceOptions {
            run_optimize: false,
            incremental_vacuum_pages: 1_000_000, // ≥ freelist, so one pass drains it
            freelist_threshold_pages: 1,
            wal_checkpoint: true,
        };

        // The vacuum step steps `incremental_vacuum` to completion (one row per page), so on a lone
        // connection (no pinned reader) a single pass returns the whole freelist to the OS.
        let r1 = run(&conn, &opts).unwrap();
        assert!(r1.vacuum_ran, "freelist over threshold");
        assert_eq!(
            r1.pages_reclaimed as i64, freelist,
            "stepped incremental_vacuum should reclaim the entire freelist in one pass"
        );

        let r2 = run(&conn, &opts).unwrap();
        assert_eq!(
            r2.pages_reclaimed, 0,
            "freelist already drained — a second pass reclaims nothing"
        );
    }

    #[test]
    fn incremental_vacuum_is_bounded_by_the_page_cap() {
        let dir = tempfile::tempdir().unwrap();
        let conn = open(&dir.path().join("cap.db"));
        conn.execute_batch("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)")
            .unwrap();
        let pad = "x".repeat(200);
        for id in 0..5_000i64 {
            conn.execute(
                "INSERT INTO t(id, v) VALUES (?, ?)",
                rusqlite::params![id, format!("{pad}-{id}")],
            )
            .unwrap();
        }
        conn.execute("DELETE FROM t", []).unwrap();
        let freelist: i64 = conn
            .query_row("PRAGMA freelist_count", [], |r| r.get(0))
            .unwrap();
        assert!(freelist > 100, "sanity ({freelist})");

        // A small cap bounds the work per pass — exactly `incremental_vacuum_pages` reclaimed.
        let r = run(
            &conn,
            &MaintenanceOptions {
                run_optimize: false,
                incremental_vacuum_pages: 25,
                freelist_threshold_pages: 1,
                wal_checkpoint: false,
            },
        )
        .unwrap();
        assert_eq!(
            r.pages_reclaimed, 25,
            "the page cap bounds reclaim per pass"
        );
    }

    /// The vacuum step no-ops on a database still at `auto_vacuum=NONE` (e.g. a file created before
    /// the engine set INCREMENTAL) rather than erroring — it just needs a one-time full VACUUM.
    #[test]
    fn vacuum_step_noops_on_auto_vacuum_none() {
        let dir = tempfile::tempdir().unwrap();
        let conn = Connection::open(dir.path().join("none.db")).unwrap();
        // Explicitly NONE, then create a table so the setting is locked in.
        conn.execute_batch(
            "PRAGMA auto_vacuum=NONE; PRAGMA journal_mode=wal2; CREATE TABLE t (id INTEGER PRIMARY KEY)",
        )
        .unwrap();
        let mode: i64 = conn
            .query_row("PRAGMA auto_vacuum", [], |r| r.get(0))
            .unwrap();
        assert_eq!(mode, 0, "sanity: db is auto_vacuum=NONE");

        let report = run(
            &conn,
            &MaintenanceOptions {
                run_optimize: false,
                incremental_vacuum_pages: 1_000,
                freelist_threshold_pages: 1,
                wal_checkpoint: false,
            },
        )
        .unwrap();
        assert!(
            !report.vacuum_ran,
            "incremental_vacuum must be skipped on an auto_vacuum=NONE file"
        );
    }
}