rindle_cdc_apply/

store.rs

//! [`ApplyStore`] — the connection pair + capture context behind the CDC apply plane:
//! the store half of what used to be `rindle-replica`'s `ClusterInner` (design 309 §3).
//!
//! One observed read-write **writer** (every durable mutation flows through it; the
//! `rindle-cdc` preupdate hook watches it), one physically read-only **reader** for
//! trusted ad-hoc reads, the committed-tx watermark, and the schema-DDL machinery
//! (`exec_ddl`, the marker-transaction primitives, checked public DDL). No engine, no
//! worker pool, no drain — the derivation host (`rindle_replica::Cluster`) composes
//! those ON TOP of this store; a headless applier composes nothing.

use std::cell::{Cell, RefCell};
use std::path::Path;
use std::rc::Rc;

use rusqlite::{Connection, OptionalExtension};

use rindle_cdc::CaptureCtx;
use rindle_writeplane::schema_envelope::{self, DdlActionRecorder, DdlApplyReport, DdlStep};
use rindle_writeplane::table_shape::ReplicatedTableSchema;
use rindle_writeplane::{
    sql, writeplane, ForeignKeys, JournalMode, ReplicaError, SqlStatementRequest, StatementResult,
    TxId, CLIENT_MUTATIONS_TABLE, ROOM_CLIENT_MUTATIONS_TABLE,
};

use super::connection::{open_journal, open_journal_read_only, open_scratch};

/// A standalone authority's checked DDL apply can fail either in the cluster/schema machinery or
/// in the shared write-plane policy checks. Keeping the latter typed preserves stable migration
/// error codes all the way to the host's HTTP renderer.
#[derive(Debug)]
pub enum DdlMigrationError {
    Replica(ReplicaError),
    Bookkeeping(writeplane::BookkeepingError),
}

impl std::fmt::Display for DdlMigrationError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            DdlMigrationError::Replica(error) => error.fmt(formatter),
            DdlMigrationError::Bookkeeping(error) => error.fmt(formatter),
        }
    }
}

impl std::error::Error for DdlMigrationError {}

impl From<ReplicaError> for DdlMigrationError {
    fn from(error: ReplicaError) -> Self {
        DdlMigrationError::Replica(error)
    }
}

impl From<writeplane::BookkeepingError> for DdlMigrationError {
    fn from(error: writeplane::BookkeepingError) -> Self {
        DdlMigrationError::Bookkeeping(error)
    }
}

/// A DDL/migration failure raised here has to surface through `writeplane`'s public SQL
/// error body. The impl lives beside [`DdlMigrationError`] because it is the local type —
/// `impl From<Local> for Foreign` is what the orphan rule permits.
impl From<DdlMigrationError> for writeplane::WritePlaneError {
    fn from(error: DdlMigrationError) -> Self {
        match error {
            DdlMigrationError::Replica(error) => error.into(),
            DdlMigrationError::Bookkeeping(error) => error.into(),
        }
    }
}

/// Enough entries for the bounded set of mutation shapes across a practical table registry.
/// Mutation SQL is constant per `(table, operation, column set)`, so steady-state follower apply
/// should compile each shape once rather than once per row.
const WRITER_STATEMENT_CACHE_CAPACITY: usize = 256;

/// The `BEGIN` flavor a headless apply store's writer transaction opens with — the same
/// flavor the live `rindle_replica::Cluster` passes (306 S5 left `BEGIN IMMEDIATE` as
/// the one writer flavor), so the headless applier and the live follower run
/// byte-identical transactions.
pub const DEFAULT_WRITER_BEGIN_SQL: &str = "BEGIN IMMEDIATE";

/// One `PRAGMA wal_checkpoint(TRUNCATE)` outcome ([`ApplyStore::checkpoint_truncate`]).
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct WalCheckpoint {
    /// True when the checkpoint could not run to completion (readers/writers held it off).
    pub busy: bool,
    /// Frames in the WAL at checkpoint time (-1 when unavailable).
    pub log_frames: i64,
    /// Frames successfully moved into the database (-1 when unavailable).
    pub checkpointed_frames: i64,
}

/// The apply plane's connection pair + capture context. `!Send` (the preupdate hook and
/// the transaction cells are single-thread state) — one per thread, like the replica
/// handles built over it.
pub struct ApplyStore {
    /// A physically **read-only** connection exposed via [`read`](Self::read) for trusted ad-hoc
    /// reads. Opening it with `SQLITE_OPEN_READ_ONLY` ensures even a callback that disables
    /// `PRAGMA query_only` cannot mutate behind CDC.
    ///
    /// **Declared before [`writer`](Self::writer) on purpose — this ordering is load-bearing.**
    /// Fields drop in declaration order, and SQLite runs the cleanup checkpoint that unlinks
    /// `-wal`/`-shm` only when the LAST connection to the database closes — which a read-only
    /// connection cannot do. So the read-only reader must never be the last one standing, or a
    /// closed database is left beside live sidecars. On a headless store these two are the ONLY
    /// connections, so the ordering carries the whole invariant (pinned by the `wal_sidecars`
    /// tests here and in `rindle-replica`).
    ///
    /// `None` on a **scratch** store ([`open_scratch`](Self::open_scratch)), which is
    /// single-connection: reads run on the writer. A scratch runs a rollback journal, where a
    /// second connection is not the free concurrency WAL makes it — a reader holding SHARED
    /// blocks the writer's commit escalation to EXCLUSIVE — and there is no one to serve it
    /// anyway, since the whole store is one thread's private derivation.
    reader: Option<Connection>,
    /// The single read-write connection. Every durable mutation goes through here;
    /// the preupdate hook observes them.
    pub(crate) writer: Connection,
    /// The preupdate-hook capture context (installed on `writer`).
    pub(crate) cdc: Rc<CaptureCtx>,
    /// The `BEGIN` flavor [`ApplyTxn::begin`](crate::ApplyTxn) opens the writer
    /// transaction with (see [`DEFAULT_WRITER_BEGIN_SQL`]).
    pub(crate) writer_begin_sql: &'static str,
    /// Last durably-committed global tx id (0 = nothing committed yet).
    pub(crate) committed_tx: Cell<u64>,
    /// Single-writer guard: true while an `ApplyTxn` is open.
    pub(crate) in_write: Cell<bool>,
}

impl Drop for ApplyStore {
    fn drop(&mut self) {
        // Remove the hook before the writer connection (and capture context) tear
        // down, so a stray late callback can't deref freed memory.
        rindle_cdc::uninstall(&self.writer);
    }
}

impl ApplyStore {
    /// Open the store over a **file-backed** SQLite database: the observed writer (with
    /// the capture hook installed), the read-only reader, the `__replica_meta`
    /// committed-tx watermark, and the journal negotiation (design 306 D3/D5 — a fresh
    /// file gets `journal`'s mode; an existing `wal`/`wal2` file keeps its mode unless
    /// `Wal2` is explicitly required). Asserts `sqlite3_threadsafe() != 0`, failing
    /// loud rather than silently degrading. `writer_begin_sql` is the `BEGIN` flavor
    /// every write transaction opens with ([`DEFAULT_WRITER_BEGIN_SQL`] unless a
    /// derivation host maps its own mode onto it). `foreign_keys` is the writer's
    /// enforcement posture: a headless replay and a live follower pass
    /// [`ForeignKeys::Unenforced`] (the stream's rows were validated upstream and its
    /// cascades already ran there), an origin write plane passes
    /// [`ForeignKeys::Enforced`] — see [`rindle_writeplane::foreign_keys`].
    pub fn open(
        path: &Path,
        journal: JournalMode,
        writer_begin_sql: &'static str,
        foreign_keys: ForeignKeys,
    ) -> Result<ApplyStore, ReplicaError> {
        if unsafe { rusqlite::ffi::sqlite3_threadsafe() } == 0 {
            return Err(ReplicaError::NotThreadsafe);
        }
        let writer = open_journal(path, journal, foreign_keys)?;
        writer.set_prepared_statement_cache_capacity(WRITER_STATEMENT_CACHE_CAPACITY);
        let reader = open_journal_read_only(path)?;
        Self::assemble(writer, Some(reader), writer_begin_sql)
    }

    /// Open a **scratch** store: one connection, `journal_mode=memory`, `synchronous=OFF`
    /// (see `connection::open_scratch` for exactly what that trades and what the caller
    /// owes in return). Everything else — the pragmas, the capture hook,
    /// the watermark, the transaction machinery — is identical to [`open`](Self::open), so a
    /// scratch materializes the same store as a durable one; only its crash posture differs.
    ///
    /// For a database DERIVED from an authority that can rebuild it (`rindle-backup`'s
    /// producer scratch, rebuilt from the archive). Never for one anybody restores from.
    pub fn open_scratch(
        path: &Path,
        writer_begin_sql: &'static str,
        foreign_keys: ForeignKeys,
    ) -> Result<ApplyStore, ReplicaError> {
        if unsafe { rusqlite::ffi::sqlite3_threadsafe() } == 0 {
            return Err(ReplicaError::NotThreadsafe);
        }
        let writer = open_scratch(path, foreign_keys)?;
        writer.set_prepared_statement_cache_capacity(WRITER_STATEMENT_CACHE_CAPACITY);
        Self::assemble(writer, None, writer_begin_sql)
    }

    /// The half of opening that is the same however the connections were made: the
    /// bookkeeping watermark, the capture hook, and the handle itself.
    fn assemble(
        writer: Connection,
        reader: Option<Connection>,
        writer_begin_sql: &'static str,
    ) -> Result<ApplyStore, ReplicaError> {
        // Bookkeeping watermark (its "__" name is skipped by CDC), then read it.
        writer
            .execute_batch(
                "CREATE TABLE IF NOT EXISTS __replica_meta (id INTEGER PRIMARY KEY, tx_id INTEGER NOT NULL)",
            )
            .map_err(|e| ReplicaError::sqlite("create __replica_meta", e))?;
        let committed: u64 = writer
            .query_row("SELECT tx_id FROM __replica_meta WHERE id = 0", [], |r| {
                r.get::<_, i64>(0)
            })
            .optional()
            .map_err(|e| ReplicaError::sqlite("read watermark", e))?
            .map(|v| v as u64)
            .unwrap_or(0);

        // Observe every row mutation on the writer.
        let cdc = CaptureCtx::new();
        rindle_cdc::install(&writer, &cdc);

        Ok(ApplyStore {
            reader,
            writer,
            cdc,
            writer_begin_sql,
            committed_tx: Cell::new(committed),
            in_write: Cell::new(false),
        })
    }

    /// The observed writer connection, for narrow host bookkeeping while no write
    /// transaction is open. Anything written here still passes through the capture hook
    /// — an application-table row landed outside the typed transaction machinery will
    /// trip the next commit's capture accounting, not silently diverge — but hosts
    /// should stay on the typed methods and keep this for unregistered bookkeeping.
    pub fn writer_connection(&self) -> &Connection {
        &self.writer
    }

    /// The capture context installed on the writer. Observation (`has_table`,
    /// `buffer_len`) is always safe; the mutating methods (`reset`/`drain`/`rewind`)
    /// belong to the transaction machinery — calling them under an open [`ApplyTxn`](crate::ApplyTxn)
    /// corrupts its capture accounting.
    pub fn capture(&self) -> &CaptureCtx {
        &self.cdc
    }

    /// The last durably-committed global tx id (0 if none yet).
    pub fn committed_tx_id(&self) -> TxId {
        TxId(self.committed_tx.get())
    }

    /// True while a write transaction ([`ApplyTxn`](crate::ApplyTxn)) is open.
    pub fn in_write(&self) -> bool {
        self.in_write.get()
    }

    /// Run an arbitrary read against a physically read-only connection (a wal2 reader sees
    /// the latest committed snapshot). Schema changes belong in [`Self::exec_ddl`], and row
    /// mutations in a write transaction.
    ///
    /// On a single-connection scratch store the read runs on the writer instead, and so sees
    /// the open write transaction's own uncommitted rows. Every caller here reads schema or
    /// bookkeeping it either just wrote or is about to depend on, so seeing more is never
    /// wrong; a future caller that needs the committed snapshot specifically must not assume
    /// this connection provides it.
    pub fn read<T>(
        &self,
        f: impl FnOnce(&Connection) -> rusqlite::Result<T>,
    ) -> Result<T, ReplicaError> {
        f(self.reader.as_ref().unwrap_or(&self.writer)).map_err(|e| ReplicaError::sqlite("read", e))
    }

    /// Walk every declared foreign key and report violating rows — the **opt-in audit**
    /// that stands in for enforcement on an apply-plane store (see
    /// [`rindle_writeplane::foreign_keys`]).
    ///
    /// `PRAGMA foreign_key_check` reports what is in the file, not what SQLite would have
    /// refused, so this answers the same question on a store opened
    /// [`ForeignKeys::Unenforced`] as on one opened [`ForeignKeys::Enforced`] — which is
    /// the whole reason the apply plane can turn enforcement off without giving up the
    /// ability to prove referential integrity. It is a full scan of the referencing
    /// tables: run it deliberately (an operator command, a post-restore gate, a soak
    /// assertion), never per commit. Runs on the read connection, so it neither takes the
    /// write lock nor disturbs an applier.
    ///
    /// Stops after `max_rows` violations, reporting
    /// [`rindle_writeplane::ForeignKeyAudit::truncated`]; pass
    /// [`rindle_writeplane::FOREIGN_KEY_AUDIT_ROW_CAP`] unless
    /// you have a reason not to, or `0` for no cap.
    pub fn foreign_key_audit(
        &self,
        max_rows: usize,
    ) -> Result<rindle_writeplane::ForeignKeyAudit, ReplicaError> {
        let conn = self.reader.as_ref().unwrap_or(&self.writer);
        rindle_writeplane::foreign_key_check(conn, max_rows)
    }

    /// The database's schema cookie (`PRAGMA schema_version`), read on the writer connection so
    /// an open mutation transaction's uncommitted DDL is visible. Hosts whose trusted SQL facade
    /// runs no statement classification compare it across a transaction to detect DDL that must
    /// invalidate reader pools and capture registration.
    pub fn schema_cookie(&self) -> Result<i64, ReplicaError> {
        self.writer
            .query_row("PRAGMA schema_version", [], |row| row.get(0))
            .map_err(|e| ReplicaError::sqlite("schema cookie", e))
    }

    /// Checkpoint and truncate the WAL through the writer. Snapshot/install code uses this
    /// after quiescing writes so the portable main file carries every committed frame. The
    /// public [`Self::read`] connection is physically read-only and deliberately cannot perform a
    /// checkpoint.
    pub fn checkpoint_truncate(&self) -> Result<WalCheckpoint, ReplicaError> {
        if self.in_write.get() {
            return Err(ReplicaError::Open(
                "cannot checkpoint while a write transaction is open".into(),
            ));
        }
        let (busy, log_frames, checkpointed_frames): (i64, i64, i64) = self
            .writer
            .query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |row| {
                Ok((row.get(0)?, row.get(1)?, row.get(2)?))
            })
            .map_err(|error| ReplicaError::sqlite("PRAGMA wal_checkpoint(TRUNCATE)", error))?;
        Ok(WalCheckpoint {
            busy: busy != 0,
            log_frames,
            checkpointed_frames,
        })
    }

    /// The engine-free half of base-table registration: introspect + validate the
    /// table's replicated shape, ensure the PK UNIQUE index row-identity point lookups
    /// require, and teach the capture hook the column types + primary key. Returns the
    /// introspected schema so a derivation host can build its sources from the SAME
    /// introspection (`rindle_replica::Cluster::register_table` lifts it to the
    /// engine's `ColumnDef` shape). Rejects BLOB columns / PK-less tables; the table
    /// must be **plain** (no triggers, no generated columns).
    pub fn register_table_capture(
        &self,
        table: &str,
    ) -> Result<ReplicatedTableSchema, ReplicaError> {
        // Registering a NEW table mid-write-txn would create its PK index inside the
        // uncommitted writer transaction — invisible to a derivation host's autocommit source
        // builds (they'd fail the PK-UNIQUE assert). Reject the nonsensical sequence.
        if self.in_write.get() {
            return Err(ReplicaError::Open(
                "cannot register a new table while a write transaction is open".into(),
            ));
        }
        let ts = schema_envelope::introspect_replicated_table(&self.writer, table)?;
        ensure_unique_pk_index_for(
            &self.writer,
            table,
            &ts.columns
                .iter()
                .map(|column| column.name.as_str())
                .collect::<Vec<_>>(),
            &ts.primary_key,
            ts.pk_is_rowid_alias,
            ts.without_rowid,
        )?;
        let col_types: Vec<_> = ts.columns.iter().map(|c| c.value_type).collect();
        self.cdc
            .set_table_with_primary_key(table, &col_types, &ts.primary_key);
        Ok(ts)
    }

    /// Create the client-mutations ledger tables (idempotent): the daemon-plane
    /// [`CLIENT_MUTATIONS_TABLE`] and the room-scoped [`ROOM_CLIENT_MUTATIONS_TABLE`].
    /// Both are **replicated data** (lmid-as-data rides the change stream), so a
    /// headless applier consuming a master's stream registers the former for capture
    /// exactly like the live follower does. The registration itself is the caller's
    /// next step (`register_table_capture`, or the derivation host's engine-inclusive
    /// registration). Rejected mid-write-txn.
    pub fn create_client_mutations_tables(&self) -> Result<(), ReplicaError> {
        if self.in_write.get() {
            return Err(ReplicaError::Open(
                "cannot enable client mutations while a write transaction is open".into(),
            ));
        }
        self.writer
            .execute_batch(&format!(
                "CREATE TABLE IF NOT EXISTS {CLIENT_MUTATIONS_TABLE} \
                 (client_id TEXT NOT NULL PRIMARY KEY, last_mutation_id INTEGER NOT NULL);\n\
                 CREATE TABLE IF NOT EXISTS {ROOM_CLIENT_MUTATIONS_TABLE} \
                 (doc TEXT NOT NULL, client_id TEXT NOT NULL, last_mutation_id INTEGER NOT NULL, \
                  PRIMARY KEY (doc, client_id))"
            ))
            .map_err(|e| ReplicaError::sqlite("create client mutations table", e))
    }

    /// Run schema DDL (`CREATE`/`ALTER`/`DROP`/`REINDEX`) against the writer connection — the
    /// supported way to define the plain base tables you then register and write through. The
    /// historical bounded `ANALYZE` call is also accepted; row-changing statements and
    /// row-producing DDL are rejected. The complete batch and schema-envelope validation commit
    /// atomically. Rejected while a write transaction is open (DDL there would be invisible to a
    /// derivation host's workers until commit).
    pub fn exec_ddl(&self, sql: &str) -> Result<(), ReplicaError> {
        if self.in_write.get() {
            return Err(ReplicaError::Open(
                "cannot run DDL while a write transaction is open".into(),
            ));
        }
        schema_envelope::ensure_embedded_ddl_batch(sql)?;
        schema_envelope::transactionally_apply_schema(
            &self.writer,
            &self.cdc,
            "exec_ddl",
            true,
            false,
            || {
                self.writer
                    .execute_batch(sql)
                    .map_err(|error| ReplicaError::sqlite("exec_ddl", error))
            },
        )
    }

    /// Apply schema `statements` AND stamp a durable idempotency marker (`key` → `marker_table`) in
    /// ONE ordinary transaction on the writer connection. This is the atomic unit a `BEGIN CONCURRENT`
    /// cursor advance can't give a follower's DDL (schema changes are illegal in a concurrent txn):
    /// because the marker lands in the SAME plain transaction as the DDL, `marker present` ⇔ `DDL
    /// applied`, exactly. A crash-window replay then dedups by `key` (a migration id / entry offset)
    /// instead of re-running the DDL and inferring "already applied?" from the error string. The
    /// `marker_table` MUST be UNREGISTERED bookkeeping (like `_rindle_source_offsets`) so the capture
    /// hook ignores its rows. The resulting schema is validated atomically before commit, and the
    /// consumer reboots its table sources afterward. Rejected while a write transaction is open
    /// (mirrors [`exec_ddl`]).
    ///
    /// # This seam trusts its statements
    ///
    /// Unlike [`exec_ddl`], this path does NOT run `ensure_embedded_ddl_batch` and passes
    /// `allow_row_changes = true`, so a historical create-copy-swap migration can land its data copy
    /// atomically with the marker. Copied rows come from an already-managed table and so satisfy the
    /// replicated value envelope by induction — but a table created inside the same batch is not yet
    /// CDC-registered, and `rindle-cdc` only counts uncaptured events for unregistered tables rather
    /// than validating their cells. **A literal invalid value in a statement given to this function
    /// is therefore not validated.**
    ///
    /// That is accepted deliberately: the only production caller is the follower's replicated-DDL
    /// apply path (`ApplyConsumer::apply_ddl_with_marker` and its `ClusterConsumer` delegate).
    /// Its statements originate from the write master's migrations, and the master's migrate
    /// surface is DDL-only. The guarantee is a reachability argument rather than an enforced check,
    /// so **if migrations ever admit DML, this seam needs a real boundary check** — validate cells
    /// for unregistered main tables in the capture hook, or register batch-created tables before
    /// the copy runs. See
    /// `follow-ups/222-224-sql-client-second-review.md` §2.6.
    ///
    /// [`exec_ddl`]: Self::exec_ddl
    pub fn exec_ddl_with_marker(
        &self,
        marker_table: &str,
        key: &str,
        statements: &[String],
    ) -> Result<DdlApplyReport, ReplicaError> {
        self.exec_ddl_with_marker_impl(
            marker_table,
            key,
            statements,
            false,
            |_conn, _step| Ok(()),
            |_conn, _changed_rows, _schemas, _report| Ok(()),
        )
    }

    /// [`exec_ddl_with_marker`](Self::exec_ddl_with_marker), with one caller-owned bookkeeping
    /// hook invoked after every real statement and its authorizer actions. The hook runs inside
    /// the SAME transaction as the DDL and marker. It may mutate unregistered bookkeeping tables,
    /// but must not issue schema DDL (the recorder and schema validator are still active).
    ///
    /// This is the follower's atomic desired-state seam: schema-derived metadata must never lag a
    /// committed marker and be reconstructed during crash replay, because startup repair can act
    /// on that stale metadata before replay gets a chance to clean it up.
    pub fn exec_ddl_with_marker_and_step_effects<F>(
        &self,
        marker_table: &str,
        key: &str,
        statements: &[String],
        mut apply_step_effects: F,
    ) -> Result<DdlApplyReport, ReplicaError>
    where
        F: FnMut(&Connection, &DdlStep) -> rusqlite::Result<()>,
    {
        self.exec_ddl_with_marker_impl(
            marker_table,
            key,
            statements,
            true,
            |conn, step| {
                apply_step_effects(conn, step)
                    .map_err(|error| ReplicaError::sqlite("ddl-marker atomic step effects", error))
            },
            |_conn, _changed_rows, _schemas, _report| Ok(()),
        )
    }

    /// Commit one public DDL unit on the standalone WAL2 authority. Every statement is first
    /// prepared under the shared public authorizer, then executed under the DDL action recorder;
    /// schema/capture/foreign-key validation, caller bookkeeping, and the next `TxId` watermark
    /// all land before the one SQLite COMMIT. The daemon reopens its engine after this boundary,
    /// so workers never continue with a pre-DDL schema at the newly persisted watermark.
    pub(crate) fn exec_public_ddl_unit<F, G>(
        &self,
        statements: &[SqlStatementRequest],
        declared_tables: &[String],
        ddl_only_message: &'static str,
        mut apply_step_effects: F,
        finish: G,
    ) -> Result<(Vec<StatementResult>, DdlApplyReport, TxId), DdlMigrationError>
    where
        F: FnMut(&Connection, &DdlStep) -> rusqlite::Result<()>,
        G: FnOnce(
            &Connection,
            TxId,
            &[StatementResult],
        ) -> Result<(), writeplane::BookkeepingError>,
    {
        if self.in_write.get() {
            return Err(ReplicaError::Open(
                "cannot run public DDL while a write transaction is open".into(),
            )
            .into());
        }
        let writer = &self.writer;
        let pre_tables = writeplane::pre_ddl_table_names(writer)?;
        let next_tx_id = TxId(self.committed_tx.get() + 1);
        let ran_drop_table = statements
            .iter()
            .any(|statement| writeplane::statement_is_drop_table(&statement.sql));

        let (results, report) = self.exec_ddl_unit_impl(
            "public ddl",
            statements,
            true,
            |conn, statement| {
                // SQLite allows one authorizer. Preflight under the denying public policy, then
                // replace it with the observing DDL recorder for the real execution.
                sql::preflight_public_ddl_statement(conn, statement)
                    .map_err(writeplane::BookkeepingError::from)?;
                conn.flush_prepared_statement_cache();
                let recorder = DdlActionRecorder::install(conn);
                let mark = recorder.mark();
                let result = writeplane::with_writer_statement_budget(conn, true, || {
                    sql::run_preflighted_ddl_statement(conn, statement)
                })
                .map_err(writeplane::BookkeepingError::from)?
                .map_err(writeplane::BookkeepingError::from)?;
                let (dropped_tables, altered_tables, created_indexes, dropped_indexes) =
                    recorder.delta_since(mark);
                drop(recorder);
                Ok((
                    result,
                    DdlStep {
                        sql: statement.sql.clone(),
                        dropped_tables,
                        altered_tables,
                        created_indexes,
                        dropped_indexes,
                    },
                ))
            },
            |conn, step| {
                apply_step_effects(conn, step).map_err(writeplane::BookkeepingError::Sqlite)?;
                Ok::<(), DdlMigrationError>(())
            },
            |_conn, _report, _results| Ok::<(), DdlMigrationError>(()),
            |conn, changed_rows, schemas, _report, results| {
                writeplane::validate_public_ddl_poststate(
                    conn,
                    changed_rows,
                    &pre_tables,
                    schemas,
                    declared_tables,
                    ddl_only_message,
                    ran_drop_table,
                )?;
                finish(conn, next_tx_id, results)?;
                self.persist_watermark_in_open_transaction(next_tx_id.0)?;
                Ok::<(), DdlMigrationError>(())
            },
        )?;
        self.committed_tx.set(next_tx_id.0);
        Ok((results, report, next_tx_id))
    }

    fn exec_ddl_with_marker_impl<E, F, V>(
        &self,
        marker_table: &str,
        key: &str,
        statements: &[String],
        step_effects_atomic: bool,
        mut apply_step_effects: F,
        validate: V,
    ) -> Result<DdlApplyReport, E>
    where
        E: From<ReplicaError>,
        F: FnMut(&Connection, &DdlStep) -> Result<(), E>,
        V: FnOnce(&Connection, usize, &[ReplicatedTableSchema], &DdlApplyReport) -> Result<(), E>,
    {
        if self.in_write.get() {
            return Err(ReplicaError::Open(
                "cannot run DDL while a write transaction is open".into(),
            )
            .into());
        }
        let w = &self.writer;
        // Execute at REAL statement boundaries so every observed action is attributed to its own
        // statement. Fall back to the whole slot only if the splitter refuses (over-limit/NUL —
        // such a slot fails execution anyway).
        let statements = statements
            .iter()
            .flat_map(|slot| sql::split_sql_script(slot).unwrap_or_else(|_| vec![slot.clone()]))
            .collect::<Vec<_>>();
        // Observe (never deny) the destructive actions the statements actually perform — the
        // authorizer is SQLite's own parser reporting each DROP/ALTER as it is prepared, so
        // comments, script-valued slots, and qualified names cannot hide one (design 227 third
        // review pass). Safe to install here: this path runs `protect_registered_tables=false`,
        // so no deny-guard competes for the connection's single authorizer slot.
        let recorder = DdlActionRecorder::install(w);
        let applied = self.exec_ddl_unit_impl(
            "ddl-marker",
            &statements,
            step_effects_atomic,
            |conn, sql| {
                let mark = recorder.mark();
                conn.execute_batch(sql)
                    .map_err(|error| E::from(ReplicaError::sqlite("ddl-marker apply", error)))?;
                let (dropped_tables, altered_tables, created_indexes, dropped_indexes) =
                    recorder.delta_since(mark);
                Ok((
                    (),
                    DdlStep {
                        sql: sql.clone(),
                        dropped_tables,
                        altered_tables,
                        created_indexes,
                        dropped_indexes,
                    },
                ))
            },
            &mut apply_step_effects,
            |conn, report, _results| {
                // Persist the ordered report IN the marker transaction: marker present iff DDL
                // and caller-owned step effects applied and the report is durable. Replay uses
                // it for bounce decisions and pre-atomic compatibility repair.
                let report_json = serde_json::to_string(report).map_err(|e| {
                    E::from(ReplicaError::Mutation(format!("ddl report encode: {e}")))
                })?;
                conn.execute(
                    &format!("INSERT OR IGNORE INTO {marker_table} (key, actions) VALUES (?1, ?2)"),
                    rusqlite::params![key, report_json],
                )
                .map_err(|e| E::from(ReplicaError::sqlite("ddl-marker apply", e)))?;
                Ok(())
            },
            |conn, changed_rows, schemas, report, _results| {
                validate(conn, changed_rows, schemas, report)
            },
        );
        drop(recorder);
        applied.map(|(_results, report)| report)
    }

    /// The one transaction/step-report spine shared by trusted marker DDL and guarded public DDL.
    /// Callers provide the statement runner (their authorizer/budget policy), the ordered
    /// bookkeeping effects, their in-transaction tail, and final post-state validation.
    #[allow(clippy::too_many_arguments)]
    fn exec_ddl_unit_impl<E, I, O, R, F, A, V>(
        &self,
        op: &'static str,
        statements: &[I],
        step_effects_atomic: bool,
        mut run_statement: R,
        mut apply_step_effects: F,
        after_statements: A,
        validate: V,
    ) -> Result<(Vec<O>, DdlApplyReport), E>
    where
        E: From<ReplicaError>,
        R: FnMut(&Connection, &I) -> Result<(O, DdlStep), E>,
        F: FnMut(&Connection, &DdlStep) -> Result<(), E>,
        A: FnOnce(&Connection, &DdlApplyReport, &[O]) -> Result<(), E>,
        V: FnOnce(
            &Connection,
            usize,
            &[ReplicatedTableSchema],
            &DdlApplyReport,
            &[O],
        ) -> Result<(), E>,
    {
        let writer = &self.writer;
        let results = RefCell::new(Vec::<O>::with_capacity(statements.len()));
        let steps = RefCell::new(Vec::<DdlStep>::with_capacity(statements.len()));
        schema_envelope::transactionally_apply_schema_checked(
            writer,
            &self.cdc,
            op,
            false,
            true,
            || {
                for statement in statements {
                    let (result, step) = run_statement(writer, statement)?;
                    apply_step_effects(writer, &step)?;
                    results.borrow_mut().push(result);
                    steps.borrow_mut().push(step);
                }
                let report = DdlApplyReport {
                    step_effects_atomic,
                    steps: steps.borrow().clone(),
                };
                after_statements(writer, &report, &results.borrow())
            },
            |conn, changed_rows, schemas| {
                let report = DdlApplyReport {
                    step_effects_atomic,
                    steps: steps.borrow().clone(),
                };
                validate(conn, changed_rows, schemas, &report, &results.borrow())
            },
        )?;
        Ok((
            results.into_inner(),
            DdlApplyReport {
                step_effects_atomic,
                steps: steps.into_inner(),
            },
        ))
    }

    /// Upsert the committed-tx watermark into `__replica_meta` (inside the writer txn,
    /// so it is durable atomically with the data), then COMMIT.
    pub(crate) fn persist_and_commit(&self, tx_id: u64) -> Result<(), ReplicaError> {
        self.persist_watermark_in_open_transaction(tx_id)?;
        self.writer
            .execute_batch("COMMIT")
            .map_err(|e| ReplicaError::sqlite("COMMIT", e))?;
        Ok(())
    }

    /// Persist only the watermark inside an already-open transaction. Ordinary DML commits call
    /// this immediately before their own COMMIT; checked public DDL calls it from the schema
    /// transaction's final validation hook so schema, outcome/journal metadata, and `TxId` are
    /// one durable atom.
    pub(crate) fn persist_watermark_in_open_transaction(
        &self,
        tx_id: u64,
    ) -> Result<(), ReplicaError> {
        self.writer
            .execute(
                "INSERT INTO __replica_meta(id, tx_id) VALUES(0, ?1) \
                 ON CONFLICT(id) DO UPDATE SET tx_id = excluded.tx_id",
                [tx_id as i64],
            )
            .map_err(|e| ReplicaError::sqlite("persist watermark", e))?;
        Ok(())
    }
}

/// Ensure `table` has a UNIQUE index covering exactly the PK columns, which row-identity
/// point lookups (the engine's `TableSource`, the follower's apply keys) require.
/// Idempotent. `column_names` is the table's columns in `cid` order; `primary_key` holds
/// indexes into it.
///
/// A rowid-alias `INTEGER PRIMARY KEY` is **skipped**: the rowid already enforces uniqueness and
/// SQLite point-looks-it-up natively (`SEARCH … USING INTEGER PRIMARY KEY (rowid=?)`), so a
/// synthetic index would be pure write amplification with no plan benefit. (Databases created by
/// older versions may still carry a now-unused `rindle_replica_pk_*` index; it is harmless and
/// left in place.)
pub fn ensure_unique_pk_index_for(
    conn: &Connection,
    table: &str,
    column_names: &[&str],
    primary_key: &[usize],
    pk_is_rowid_alias: bool,
    without_rowid: bool,
) -> Result<(), ReplicaError> {
    if (pk_is_rowid_alias && !without_rowid)
        || has_pk_unique_index(conn, table, column_names, primary_key)?
    {
        return Ok(());
    }
    let cols = primary_key
        .iter()
        .map(|&c| rindle_writeplane::quote_ident(column_names[c]))
        .collect::<Vec<_>>()
        .join(", ");
    let sql = format!(
        "CREATE UNIQUE INDEX IF NOT EXISTS {} ON {} ({cols})",
        rindle_writeplane::quote_ident(&format!("rindle_replica_pk_{table}")),
        rindle_writeplane::quote_ident(table),
    );
    conn.execute_batch(&sql)
        .map_err(|e| ReplicaError::sqlite("create pk unique index", e))?;
    Ok(())
}

/// Does an existing UNIQUE index cover *exactly* the PK column set?
fn has_pk_unique_index(
    conn: &Connection,
    table: &str,
    column_names: &[&str],
    primary_key: &[usize],
) -> Result<bool, ReplicaError> {
    let want: std::collections::BTreeSet<usize> = primary_key.iter().copied().collect();
    let name_to_id = |n: &str| -> Option<usize> { column_names.iter().position(|c| *c == n) };

    let mut list = conn
        .prepare("SELECT name, \"unique\" FROM pragma_index_list(?1)")
        .map_err(|e| ReplicaError::sqlite("prepare pragma_index_list", e))?;
    let indexes: Vec<(String, i64)> = list
        .query_map([table], |r| {
            Ok((r.get::<_, String>(0)?, r.get::<_, i64>(1)?))
        })
        .map_err(|e| ReplicaError::sqlite("run pragma_index_list", e))?
        .filter_map(Result::ok)
        .collect();

    for (idx, unique) in indexes {
        if unique == 0 {
            continue;
        }
        let mut info = conn
            .prepare("SELECT name FROM pragma_index_info(?1)")
            .map_err(|e| ReplicaError::sqlite("prepare pragma_index_info", e))?;
        let cols: std::collections::BTreeSet<usize> = info
            .query_map([&idx], |r| r.get::<_, String>(0))
            .map_err(|e| ReplicaError::sqlite("run pragma_index_info", e))?
            .filter_map(Result::ok)
            .filter_map(|n| name_to_id(&n))
            .collect();
        if cols == want {
            return Ok(true);
        }
    }
    Ok(false)
}