rindle_cdc_apply/

connection.rs

//! The journal-aware connection-opening ritual every apply-plane (and engine) connection
//! goes through: journal-mode negotiation (design 306 D3/D5), the hardening/planner
//! pragmas, and the `regexp` registration that keeps every read surface answering the
//! same SQL the write master does.
//!
//! Lifted out of `rindle-replica`'s `parallel` module by the `rindle-cdc-apply`
//! extraction (design 309): the ritual is `rusqlite` + `rindle-regex`, engine-free, and
//! the headless apply store must open its connections EXACTLY the way the live
//! follower's coordinator does — same journal negotiation, same pragmas — or the
//! byte-identity argument between the two hosts breaks at the connection boundary.

use std::path::Path;

use rusqlite::{Connection, OpenFlags};

use rindle_writeplane::{set_foreign_keys, ForeignKeys, JournalMode, ReplicaError};

/// Cap how many rows `ANALYZE` / `PRAGMA optimize` samples per index, so a maintenance
/// tick's stats refresh on the live writer stays sub-millisecond regardless of table
/// size (see `rindle-replica`'s `maintenance` module, which restores this bound after
/// its full-`ANALYZE` escape hatch). Applied to every connection opened here.
pub const ANALYSIS_LIMIT: u32 = 400;

/// True for the WAL-family modes the engine runs on (either accepted — D3).
fn is_wal_family(mode: &str) -> bool {
    mode.eq_ignore_ascii_case("wal") || mode.eq_ignore_ascii_case("wal2")
}

/// Open one read-write connection (mirrors `Db::open`'s per-connection setup). Used for
/// the apply store's writer, the cluster coordinator's writer, and each IVM worker.
///
/// `foreign_keys` is the connection's enforcement posture and has no default here: an
/// **origin** write plane (the standalone daemon, the HCTree master) says
/// [`ForeignKeys::Enforced`], an **apply** plane (a follower, a headless replay) says
/// [`ForeignKeys::Unenforced`] because the rows it writes were already validated by the
/// authority whose cascades are themselves in the stream. See
/// [`rindle_writeplane::foreign_keys`] for the full argument, and
/// [`rindle_writeplane::foreign_key_check`] for the audit that works either way.
///
/// Journal handling (design 306 D3/D5): under the default `Wal` request a file
/// already in `wal` or `wal2` is accepted as-is; anything else (a fresh file's
/// `delete`, a rollback-mode file) is converted to the requested mode. An explicit
/// `Wal2` request additionally attempts to convert a plain-`wal` file and — because
/// the pragma silently reports `wal` when the switch cannot happen (an
/// un-checkpointed WAL, other connections; `docs/SQLITE_WAL2.md` §10.3) — **verifies
/// the outcome and fails loudly** rather than running a wal2-contracted daemon on a
/// plain-wal store (the post-condition the retired `open_wal2` enforced).
pub fn open_journal(
    path: &Path,
    journal: JournalMode,
    foreign_keys: ForeignKeys,
) -> Result<Connection, ReplicaError> {
    let conn = Connection::open(path).map_err(|e| ReplicaError::sqlite("open", e))?;
    let mut mode: String = conn
        .query_row("PRAGMA journal_mode", [], |r| r.get(0))
        .map_err(|e| ReplicaError::sqlite("read journal_mode", e))?;
    let wal2_required = journal == JournalMode::Wal2;
    if !is_wal_family(&mode) || (wal2_required && !mode.eq_ignore_ascii_case("wal2")) {
        mode = conn
            .query_row(journal.pragma(), [], |r| r.get(0))
            .map_err(|e| ReplicaError::sqlite("set journal_mode", e))?;
    }
    if !is_wal_family(&mode) {
        return Err(ReplicaError::Open(format!(
            "expected a wal/wal2 journal mode, got {mode:?} (use a file-backed path)"
        )));
    }
    if wal2_required && !mode.eq_ignore_ascii_case("wal2") {
        return Err(ReplicaError::Open(format!(
            "journal_mode=wal2 was requested but the store is {mode:?} and could not be \
             converted (an un-checkpointed wal file or a concurrent connection — \
             docs/SQLITE_WAL2.md §10.3); refusing to run outside the wal2 fleet contract"
        )));
    }
    let conn = configure_engine_connection(conn)?;
    set_foreign_keys(&conn, foreign_keys)?;
    Ok(conn)
}

/// Set `PRAGMA wal_autocheckpoint = pages` on one connection — the WAL-growth knob a
/// host that wants to own its checkpoint schedule passes at open
/// (`OpenOptions::wal_autocheckpoint`; design 410 §3.5).
///
/// `0` disables SQLite's automatic checkpoint entirely, so the WAL only ever folds back
/// on an explicit `PRAGMA wal_checkpoint` (the replica's `maintain`). That is the mobile
/// posture: a phone would otherwise pay an unpredictable multi-megabyte checkpoint
/// inside whichever commit happens to cross the threshold — mid-gesture, at 60 Hz — and
/// would rather take it at a moment of its choosing (gesture end, idle, backgrounding).
/// A host that does NOT set this keeps SQLite's default of 1000 pages, which is what
/// every server-side opener wants.
///
/// Connection-local (it is not persisted in the file) and only meaningful on the
/// connection that actually writes, so callers apply it to the writer.
pub fn set_wal_autocheckpoint(conn: &Connection, pages: u32) -> Result<(), ReplicaError> {
    conn.execute_batch(&format!("PRAGMA wal_autocheckpoint = {pages}"))
        .map_err(|e| ReplicaError::sqlite("wal_autocheckpoint", e))
}

/// Open a physically read-only connection for the public `read` surface. `PRAGMA
/// query_only` is only a mutable connection setting; opening with
/// `SQLITE_OPEN_READ_ONLY` is what prevents a callback from turning it off and mutating
/// the durable database behind CDC.
///
/// The writer must have opened the database first, so a WAL-family journal mode is
/// already persistent on the file. This opener verifies that (accepting `wal` or `wal2`
/// — D3) without trying to change it, then applies the same connection-local
/// planner/read pragmas as every other engine connection.
///
/// Takes no [`ForeignKeys`] posture, unlike its read-write peers: `PRAGMA foreign_keys`
/// governs whether a *write* is refused and whether a declared `ON DELETE`/`ON UPDATE`
/// action fires, and this connection can do neither. `PRAGMA foreign_key_check` — the
/// audit — reads the same answer here whatever the pragma says
/// ([`rindle_writeplane::foreign_key_check`]), which is precisely why the audit is the
/// read surface's tool and the pragma is not.
pub fn open_journal_read_only(path: &Path) -> Result<Connection, ReplicaError> {
    let conn = Connection::open_with_flags(
        path,
        OpenFlags::SQLITE_OPEN_READ_ONLY
            | OpenFlags::SQLITE_OPEN_NO_MUTEX
            | OpenFlags::SQLITE_OPEN_URI,
    )
    .map_err(|e| ReplicaError::sqlite("open read-only", e))?;
    // An attached schema gets its own open flags and could otherwise re-open this same file
    // read-write under another schema name. Keep the raw-Connection callback confined to `main`.
    conn.set_limit(rusqlite::limits::Limit::SQLITE_LIMIT_ATTACHED, 0);
    let mode: String = conn
        .query_row("PRAGMA journal_mode", [], |r| r.get(0))
        .map_err(|e| ReplicaError::sqlite("read journal_mode", e))?;
    if !is_wal_family(&mode) {
        return Err(ReplicaError::Open(format!(
            "expected a wal/wal2 journal mode, got {mode:?} (use a file-backed path)"
        )));
    }
    configure_engine_connection(conn)
}

/// Open the single connection of a **scratch** store: a database that is DERIVED from
/// somewhere else (today, `rindle-backup`'s producer scratch, rebuilt from the archive)
/// and is thrown away and rebuilt whenever it is not known to have been closed cleanly.
///
/// It trades the two durability properties the shared ritual above buys — and nothing
/// else; every other pragma is identical, so the store it materializes is the same store:
///
/// * `journal_mode = memory` — the rollback journal lives in RAM, so a commit writes each
///   page exactly once (no WAL, and therefore no checkpoint rewriting every page into the
///   main file). `ROLLBACK` still works, so a failed frame still unwinds cleanly; what is
///   lost is only crash recovery, since a process that dies mid-transaction leaves no hot
///   journal on disk to roll back. Connection-local and NOT persisted in the file, so
///   every reopen (a DDL bounce, the next quantum) must come back through here.
/// * `synchronous = OFF` — no fsync on the commit or the checkpoint path.
///
/// Neither is safe for a database anyone must be able to recover: after an OS crash or
/// power loss the file may be torn in ways `PRAGMA quick_check` cannot see (a lost leaf
/// page is structurally valid and silently stale). **The caller owes the other half of
/// the bargain**: it must record cleanliness out of band and rebuild from source rather
/// than reuse a scratch that is not known-clean — see `rindle_backup_sqlite::producer`'s
/// `ScratchState::clean`, which is written and fsynced only after a successful seal.
pub fn open_scratch(path: &Path, foreign_keys: ForeignKeys) -> Result<Connection, ReplicaError> {
    let conn = Connection::open(path).map_err(|e| ReplicaError::sqlite("open scratch", e))?;
    // Like the wal2 request above, the pragma reports the mode it actually ended up in
    // rather than failing, so verify it: a scratch that silently stayed in wal/wal2 would
    // quietly cost the double write this opener exists to avoid.
    let mode: String = conn
        .query_row("PRAGMA journal_mode = memory", [], |r| r.get(0))
        .map_err(|e| ReplicaError::sqlite("set journal_mode=memory", e))?;
    if !mode.eq_ignore_ascii_case("memory") {
        return Err(ReplicaError::Open(format!(
            "scratch journal_mode=memory was requested but the file is {mode:?} and could \
             not be converted (an un-checkpointed wal file or a concurrent connection)"
        )));
    }
    conn.execute_batch("PRAGMA synchronous = OFF;")
        .map_err(|e| ReplicaError::sqlite("synchronous=OFF", e))?;
    let conn = configure_shared_pragmas(conn)?;
    set_foreign_keys(&conn, foreign_keys)?;
    Ok(conn)
}

fn configure_engine_connection(conn: Connection) -> Result<Connection, ReplicaError> {
    // Lever A — follower durability (REPLICATOR-THROUGHPUT-AND-STREAMING-DESIGN.md §2.1): drop the
    // per-commit WAL fsync, matching `WriteMaster` on the relay side. `synchronous=NORMAL` is
    // crash-safe under wal/wal2 (no corruption ever; survives process death) and only relaxes
    // durability against power loss / OS crash — the last not-yet-checkpointed commits. A follower
    // is fully recoverable from the master's stream + its durable cursor, so that loss is already
    // covered out of band; `FULL` (the WAL default we'd otherwise inherit) would fsync every commit
    // for a guarantee this topology discards. `NORMAL` is a checkpoint-sync relaxation, orthogonal
    // to the wal/wal2 file layout. Set on every connection here — the coordinator's writer/reader
    // and each worker — since they all open through `open_journal`.
    conn.execute_batch("PRAGMA synchronous = NORMAL;")
        .map_err(|e| ReplicaError::sqlite("synchronous=NORMAL", e))?;
    configure_shared_pragmas(conn)
}

/// Everything on an engine connection that is independent of its durability posture —
/// shared verbatim by [`configure_engine_connection`] and [`open_scratch`], so the two
/// differ ONLY in journal mode and `synchronous` and answer the same SQL as each other
/// and as the write master.
fn configure_shared_pragmas(conn: Connection) -> Result<Connection, ReplicaError> {
    // Cap how many rows `ANALYZE` / `PRAGMA optimize` samples per index, so the maintenance tick's
    // stats refresh on the live writer stays sub-millisecond regardless of table size (see the
    // `maintenance` module). Per-connection; harmless on the readers/workers that never optimize.
    //
    // NOTE: we intentionally do NOT set `auto_vacuum=INCREMENTAL` or `busy_timeout` here. Both
    // perturb the verified concurrency core — incremental auto-vacuum changes the page allocation
    // BEGIN CONCURRENT's conflict detection / fault recovery were validated against, and a long
    // `busy_timeout` turns a fast-fail lock contention into a stall. The maintenance vacuum step is
    // therefore opt-in per database (`maintenance` module docs).
    conn.execute_batch(&format!("PRAGMA analysis_limit = {ANALYSIS_LIMIT}"))
        .map_err(|e| ReplicaError::sqlite("analysis_limit", e))?;
    // Bare `LIKE` must agree with the engine's in-memory matcher and `rindle-d2s` (both
    // case-sensitive; the IVM `TableSource` connection sets the same pragma) — otherwise a
    // mutator-session read or raw `/execute-sql-read` silently matches case-insensitively
    // while every maintained view matches case-sensitively. Per-connection; set on every
    // cluster connection since they all open through these openers.
    conn.execute_batch("PRAGMA case_sensitive_like = ON;")
        .map_err(|e| ReplicaError::sqlite("case_sensitive_like", e))?;
    // `regexp` for the same reason as the pragma above: SQLite has no built-in function for
    // the operator, so without this every read surface that reaches a store through these
    // openers answers `x REGEXP y` with "no such function" while the master's writer answers
    // it. Registered on every cluster connection since they all open through here.
    rindle_regex::register(&conn).map_err(|e| ReplicaError::sqlite("register regexp", e))?;
    // Query Planner Stability Guarantee. The engine reuses cached prepared statements
    // (`prepare_cached`) for the IVM leaf seeks, and re-runs each one — same shape, new
    // bound key — millions of times across derivation. Without QPSG, this SQLite build
    // lets the planner consult bound-parameter VALUES (STAT4 and other value-based
    // optimizations), which sets the statement's `expmask`; re-binding then EXPIRES the
    // cached VM, and the next `step()` does a full SQLite re-PARSE (`sqlite3Reprepare` →
    // `sqlite3RunParser`). Measured on the derivation fetch path that cost ~2x the
    // statement's actual execution — the dominant term in the follower's per-write derive,
    // and it scales O(writes x pipelines). QPSG makes plans value-independent (stable per
    // shape) so a cached seek is prepared once and reused — exactly what an IVM engine
    // wants; the equality/range index plans the engine emits are optimal regardless of the
    // bound value, so QPSG forfeits nothing here. Per-connection; set on every cluster
    // connection (workers + coordinator) since they all open through here.
    conn.set_db_config(
        rusqlite::config::DbConfig::SQLITE_DBCONFIG_ENABLE_QPSG,
        true,
    )
    .map_err(|e| ReplicaError::sqlite("enable QPSG", e))?;
    Ok(conn)
}