rindle_replica/

cluster.rs

//! A single writer/coordinator with independent live-query engines on worker threads.
//! [`Db`](crate::Db) delivers callbacks on its owner thread; [`Cluster`] distributes
//! query registrations and sends events through a bounded channel.
//!
//! ## Transaction and delivery order
//!
//! 1. [`write`](Cluster::write) opens the controlled writer with `BEGIN IMMEDIATE`.
//!    Captured row changes start the worker snapshot barrier. Each worker pins a
//!    read-only snapshot before the writer can commit.
//! 2. Captured chunks stream to workers while SQL runs. Workers derive with an
//!    in-memory batch overlay and can emit provisional `Changed` slices immediately.
//! 3. [`commit`](ClusterWriteTxn::commit) sends remaining chunks, waits for snapshot
//!    acknowledgments, commits SQL, and sends the commit verdict to the workers.
//! 4. After all slices and a successful verdict, each worker emits `Progressed`.
//!    Failed or aborted speculative derivation faults affected query registrations.
//!
//! A receiver must buffer changes until progress from the relevant workers confirms
//! them. A rollback after streamed changes requires discarding those changes and
//! re-registering faulted queries. See [`ClusterEvent`](crate::ClusterEvent).
//!
//! Registration hydrates a committed snapshot and is rejected while workers hold a
//! speculative view. Each query stays on one worker, preserving per-query order.
//! The coordinator is `!Send`; only owned captures and events cross thread boundaries.
//!
//! ## Composition (design 309)
//!
//! The store half — connections, capture, the write-transaction state machine, DDL —
//! lives in the engine-free apply plane ([`crate::apply`]): `Cluster` =
//! [`ApplyStore`] + [`WorkerPool`] + a schema-only [`Engine`], with the pool behind
//! the apply plane's [`CommitFanout`] seam ([`PoolFanout`]). Every store-half method
//! here DELEGATES, so the live follower and a headless applier share one
//! implementation.

use std::cell::{Cell, RefCell};
use std::path::Path;
use std::rc::Rc;
use std::sync::mpsc::Receiver;
use std::sync::Arc;

use rindle::value::{OwnedValue, Schema};
use rindle::{Ast, CaughtChange};
use rindle_cdc::Captured;
use rusqlite::{Connection, OptionalExtension};

use crate::analyze::AnalyzeReport;
use crate::apply::{
    ApplyStore, ApplyTxn, CommitFanout, CommitInfo, FanoutGate, FanoutStream,
    StreamBegin as ApplyStreamBegin,
};
use crate::engine::{analyze_standalone, AnalyzeSpec, Engine, OperatorStorage};
use crate::mutations::{
    realtime_lifecycle_ddl, upsert_lmid, MutationEnvelope, MutationOutcome, MutationReject,
    MutationSql, MutatorRegistry,
};
use crate::parallel::{
    event_channel_bound, open_journal, CommitGate as PoolCommitGate,
    StreamBegin as PoolStreamBegin, StreamTx as PoolStreamTx, WorkerPool,
};
use crate::schema;
use crate::{
    ClusterEvent, OpenOptions, QueryId, ReplicaError, SqlStatementRequest, StatementResult,
    StatementRunError, TxId,
};
use crate::{
    CLIENT_MUTATIONS_TABLE, ROOM_CLIENT_MUTATIONS_TABLE, ROOM_MUTATION_OUTCOMES_TABLE,
    ROOM_WATERMARK_TABLE, SCOPE_SESSIONS_TABLE,
};
use rindle_wire::family_key::Binding;

pub use crate::apply::DdlMigrationError;

/// The worker pool behind the apply plane's [`CommitFanout`] seam: the cluster's
/// derivation half. Owns the speculative-window bookkeeping too — the fence is read
/// only by the derivation surface ([`Cluster::read_snapshot`] /
/// [`Cluster::view_is_speculative`]), so it belongs on this side of the trait
/// (design 309 §3).
pub(crate) struct PoolFanout {
    pub(crate) pool: WorkerPool,
    pub(crate) n_workers: usize,
    /// True once the open transaction has PUSHED a speculative chunk to the workers
    /// (the capture pump crossed `PUSH_CHUNK_ROWS`) — the window in which a worker's
    /// assembled view has advanced past the last commit with uncommitted rows.
    /// Strictly narrower than the store's `in_write`: a transaction that never crosses
    /// the streaming threshold buffers entirely on the writer, so every worker's view
    /// still sits at the last commit and [`Cluster::read_snapshot`] stays serviceable
    /// throughout. Cleared by commit and by every rollback path (the stream/gate
    /// wrappers below own the transitions). `Rc` so the per-transaction wrappers share
    /// the one cell.
    ///
    /// Note the eager begin-barrier (`ApplyTxn::maybe_pump`,
    /// `follow-ups/cluster-barrier-wakeup.md` §6.1) opens the worker stream — and pins
    /// each worker's snapshot — at the transaction's first *captured change*, before
    /// anything is pushed. That wider window deliberately does NOT set this flag:
    /// batch-overlay derivation never writes through the worker connection, so a
    /// registration's cold hydrate inside the pinned (plain `BEGIN`) snapshot is a pure
    /// read of the committed post-(N-1) state — exactly the `hydrated_tx` it is stamped
    /// with. Only *pushed* chunks (folded into the sources' overlays, visible to a
    /// fresh hydrate) poison a baseline, and this flag starts precisely there.
    pub(crate) speculative: Rc<Cell<bool>>,
}

impl CommitFanout for PoolFanout {
    fn tx_begin(&self, tx_id: u64) -> ApplyStreamBegin {
        let wrap = |inner: PoolStreamTx| -> Box<dyn FanoutStream> {
            Box::new(PoolStream {
                inner: Some(inner),
                speculative: self.speculative.clone(),
            })
        };
        match self.pool.tx_begin(tx_id) {
            PoolStreamBegin::Ready(stream) => ApplyStreamBegin::Ready(wrap(stream)),
            PoolStreamBegin::Failed(stream) => ApplyStreamBegin::Failed(wrap(stream)),
        }
    }

    fn recover_failed_begin(&self) {
        // A worker died/stalled at the begin-barrier: reap + respawn the offender(s) so
        // the caller's retry can succeed.
        self.pool.reap();
    }
}

/// One in-flight transaction's fan-out over the pool ([`crate::parallel::StreamTx`]
/// behind the apply-plane trait), carrying the speculative-fence transitions.
struct PoolStream {
    /// `None` once finished (the gate owns the rest of the lifecycle).
    inner: Option<PoolStreamTx>,
    speculative: Rc<Cell<bool>>,
}

impl FanoutStream for PoolStream {
    fn tx_id(&self) -> u64 {
        self.inner.as_ref().expect("stream not finished").tx_id()
    }

    fn push(&mut self, changes: Arc<[Captured]>) {
        self.inner
            .as_mut()
            .expect("stream not finished")
            .push(changes);
        // The worker graphs now hold uncommitted rows: fence the one-shot view snapshot
        // until this txn's terminal marker (see `Cluster::view_is_speculative`).
        self.speculative.set(true);
    }

    fn await_acks(&mut self) -> bool {
        self.inner
            .as_mut()
            .expect("stream not finished")
            .await_acks()
    }

    fn finish(mut self: Box<Self>) -> Box<dyn FanoutGate> {
        let gate = self.inner.take().expect("stream not finished").finish();
        Box::new(PoolGate {
            gate,
            speculative: self.speculative.clone(),
        })
    }
}

impl Drop for PoolStream {
    fn drop(&mut self) {
        // Dropped without `finish` — an abort/rollback: the fence clears with the stream.
        // (The inner `StreamTx`'s own Drop releases the workers' held snapshots; a worker
        // that already received pushes tears down + re-hydrates.)
        if self.inner.is_some() {
            self.speculative.set(false);
        }
    }
}

struct PoolGate {
    gate: PoolCommitGate,
    speculative: Rc<Cell<bool>>,
}

impl FanoutGate for PoolGate {
    fn commit(self: Box<Self>) {
        self.speculative.set(false);
        self.gate.commit();
    }

    fn abort(self: Box<Self>) {
        self.speculative.set(false);
        self.gate.abort();
    }
}

/// Shared coordinator-thread state behind a [`Cluster`] (held via `Rc` so the
/// `Cluster` handle and its [`ClusterWriteTxn`]s reference one store + pool).
/// Field order is the teardown order: the store's own reader-before-writer ordering
/// (see [`ApplyStore`]) plus pool-joins-before-the-schema-engine-closes keep a
/// read-write connection last, so SQLite's cleanup checkpoint can unlink the
/// `-wal`/`-shm` sidecars (pinned by `tests/wal_sidecars.rs`).
struct ClusterInner {
    /// The engine-free store half: reader + observed writer + capture + the committed-tx
    /// watermark (design 309). Shared (`Rc`) with the write transactions and with
    /// [`ClusterConsumer`](crate::ClusterConsumer)'s apply surface.
    store: Rc<ApplyStore>,
    /// The worker pool behind the apply plane's commit fan-out seam, plus the
    /// speculative-window fence.
    fanout: Rc<PoolFanout>,
    /// A **schema-only** engine on the coordinator's own connection, `register_table`'d in
    /// lockstep with the workers (a source per table, no queries). The coordinator has no
    /// worker `Engine`, so this is where [`Cluster::view_schema`] resolves a query's view
    /// [`Schema`] — using the *exact same* `Engine::register_table` source-schema mapping
    /// the workers use, so the view schema lines up with the change stream. It never
    /// derives changes, so sharing the coordinator thread is sound. Doubles
    /// as the registered-tables record (idempotency via `has_source`).
    schema_engine: RefCell<Engine>,
}

/// The multi-threaded replica handle (cheap `Rc` clone). Single-thread-use on the
/// coordinator side; a server runs it on a dedicated owner thread and **continuously**
/// drains the [`ClusterEvent`] channel returned by [`open`](Cluster::open) on whatever
/// thread it likes — but it must never STOP draining while writes flow (the channel-out
/// is bounded; see `open`'s draining contract).
#[derive(Clone)]
pub struct Cluster {
    inner: Rc<ClusterInner>,
}

impl Cluster {
    /// Open a parallel replica over a **file-backed** SQLite database with
    /// `n_workers` IVM worker threads. Returns the handle and the channel-out
    /// `Receiver` for derived [`ClusterEvent`]s. A fresh file gets ordinary
    /// `journal_mode = wal` (design 306 D5); an existing `wal` or `wal2` file keeps
    /// its mode. Asserts `sqlite3_threadsafe() != 0`, like [`Db::open`].
    ///
    /// # Draining contract
    ///
    /// The consumer **MUST drain the returned `Receiver` continuously**, on its own
    /// thread, for the whole life of the `Cluster`. The channel-out is BOUNDED
    /// (FOLLOWER-LAG-SHED §4, rung 0b — `event_channel_bound(n_workers)` events, to bound
    /// memory under a slow consumer). A full event channel blocks a worker, which can
    /// block capture streaming or a later snapshot barrier on the writer. Reading
    /// events only after writes return can therefore deadlock. The drain loop must
    /// also avoid blocking on slow downstream subscribers. Commit return does not
    /// mean all worker events have been delivered; use `Progressed` for that boundary.
    ///
    /// Prefer [`ClusterConsumer`](crate::ClusterConsumer), which spawns and owns that
    /// continuous drain thread for you; reach for raw `open` only when you intend to run
    /// the drain loop yourself.
    ///
    /// [`Db::open`]: crate::Db::open
    pub fn open(
        path: impl AsRef<Path>,
        n_workers: usize,
    ) -> Result<(Cluster, Receiver<ClusterEvent>), ReplicaError> {
        Self::open_with(path, n_workers, OpenOptions::default())
    }

    /// Like [`open`](Self::open), but with explicit control over the cost-based join-flip
    /// planner. When `plan_queries` is true, every worker runs the planner (a
    /// `SqliteCostModel` over its own connection, cached) at query registration to annotate
    /// `flip` before lowering. Result-preserving; the plan is frozen per registration.
    /// The planner is server-side (table-source) only and `open` enables it by default;
    /// pass `false` here to opt out (e.g. to pin the unplanned path).
    pub fn open_with_planning(
        path: impl AsRef<Path>,
        n_workers: usize,
        plan_queries: bool,
    ) -> Result<(Cluster, Receiver<ClusterEvent>), ReplicaError> {
        Self::open_with(
            path,
            n_workers,
            OpenOptions {
                plan_queries,
                ..OpenOptions::default()
            },
        )
    }

    /// Like [`open_with_planning`](Self::open_with_planning), but also selects the
    /// operator-scratch-state backend for every worker `Engine`. Pass
    /// [`OperatorStorage::SqliteSpill`] to spill stateful-operator state to a per-worker
    /// on-disk temp database instead of RAM. The schema-only engine never builds stateful
    /// operators, so it always stays memory-backed. Fresh files get plain `wal`
    /// (design 306 D5); use [`open_with_journal`](Self::open_with_journal) for the
    /// daemon's wal2 opt-in.
    pub fn open_with_options(
        path: impl AsRef<Path>,
        n_workers: usize,
        plan_queries: bool,
        operator_storage: OperatorStorage,
    ) -> Result<(Cluster, Receiver<ClusterEvent>), ReplicaError> {
        Self::open_with(
            path,
            n_workers,
            OpenOptions {
                plan_queries,
                operator_storage,
                ..OpenOptions::default()
            },
        )
    }

    /// Like [`open`](Self::open), but with an explicit fresh-file [`JournalMode`] —
    /// the daemon's roles pass [`JournalMode::Wal2`] here (design 306 D5: wal2 is
    /// something the daemon requests, never something the library imposes). An
    /// existing file already in `wal` or `wal2` keeps its mode either way.
    ///
    /// [`JournalMode`]: crate::JournalMode
    /// [`JournalMode::Wal2`]: crate::JournalMode::Wal2
    pub fn open_with_journal(
        path: impl AsRef<Path>,
        n_workers: usize,
        journal: crate::JournalMode,
    ) -> Result<(Cluster, Receiver<ClusterEvent>), ReplicaError> {
        Self::open_with(
            path,
            n_workers,
            OpenOptions {
                journal,
                ..OpenOptions::default()
            },
        )
    }

    /// The full-combination opener: every [`OpenOptions`] field is honored, so any
    /// planner × operator-storage × journal combination is one call (e.g. a planned,
    /// spill-backed cluster on a wal2 daemon store). The named rungs above are
    /// conveniences over this.
    pub fn open_with(
        path: impl AsRef<Path>,
        n_workers: usize,
        opts: OpenOptions,
    ) -> Result<(Cluster, Receiver<ClusterEvent>), ReplicaError> {
        let OpenOptions {
            plan_queries,
            operator_storage,
            journal,
            wal_autocheckpoint,
            foreign_keys,
        } = opts;
        assert!(n_workers > 0, "a cluster needs at least one worker");
        let path = path.as_ref();
        // The engine-free store half: writer (+ capture hook) and reader, the journal
        // ritual, and the committed-tx watermark (design 309 §3). The writer begins
        // IMMEDIATE — the one BEGIN flavor left with WriteThenAbort gone (306 S5).
        let store = ApplyStore::open(
            path,
            journal,
            crate::apply::DEFAULT_WRITER_BEGIN_SQL,
            foreign_keys,
        )?;
        // Same meaning as on the single-thread `Db` (design 410 §3.5): the coordinator's
        // writer is the only connection that appends WAL frames, and the pragma is
        // connection-local. The workers open read-only-in-practice connections that never
        // checkpoint, so they need none of it.
        if let Some(pages) = wal_autocheckpoint {
            crate::parallel::set_wal_autocheckpoint(store.writer_connection(), pages)?;
        }
        // The schema-only engine's own connection (it only reads schema to build sources).
        let schema_conn = Rc::new(open_journal(path, journal, foreign_keys)?);

        // Spawn the workers + channel-out. The channel is BOUNDED (FOLLOWER-LAG-SHED §4,
        // rung 0b): a full queue blocks a worker's event send, so a slow drain
        // backpressures the workers — and, through the bounded `TxPush` credits, the
        // coordinator — instead of growing daemon memory without limit. Self-limiting and
        // uniform; the drain loop's sink never blocks (it evicts a slow client), so the
        // chain always makes progress (§4's no-deadlock argument).
        let (out_tx, out_rx) = std::sync::mpsc::sync_channel(event_channel_bound(n_workers));
        let pool = WorkerPool::spawn(
            path,
            n_workers,
            plan_queries,
            operator_storage,
            journal,
            foreign_keys,
            out_tx,
        )?;

        // The schema engine only resolves view schemas; it never registers user queries,
        // so planning is irrelevant and it builds no stateful operators — always off and
        // memory-backed.
        let schema_engine = Engine::new(schema_conn, false, OperatorStorage::Memory)?;
        let inner = Rc::new(ClusterInner {
            store: Rc::new(store),
            fanout: Rc::new(PoolFanout {
                pool,
                n_workers,
                speculative: Rc::new(Cell::new(false)),
            }),
            schema_engine: RefCell::new(schema_engine),
        });
        Ok((Cluster { inner }, out_rx))
    }

    /// The shared store half, for [`ClusterConsumer`](crate::ClusterConsumer)'s apply
    /// surface (`ApplyConsumer::from_parts`).
    pub(crate) fn store_rc(&self) -> Rc<ApplyStore> {
        self.inner.store.clone()
    }

    /// The pool's fan-out, for the same composition seam.
    pub(crate) fn fanout_rc(&self) -> Rc<dyn CommitFanout> {
        self.inner.fanout.clone()
    }

    /// Register a base table: discover its columns + PK, ensure the PK UNIQUE index
    /// `TableSource` requires, teach the capture hook the column types, and build
    /// the source on **every** worker (any may host a query referencing it).
    /// Idempotent. Rejects BLOB columns / PK-less tables.
    ///
    /// The table must be **plain** — SQL triggers and generated columns are rejected (see the
    /// crate-level "supported usage" docs). Foreign-key cascades are represented by the explicit
    /// row deltas observed by the preupdate hook.
    pub fn register_table(&self, table: &str) -> Result<(), ReplicaError> {
        if self.inner.schema_engine.borrow().has_source(table) {
            return Ok(());
        }
        // The engine-free half (introspection + PK index + capture registration) lives on
        // the store; lift the SAME introspection into the engine's `ColumnDef` shape and
        // build the source on every worker AND on the coordinator's schema-only engine (the
        // latter feeds `view_schema`). Build the schema engine first so `view_schema` is
        // available the moment the table is registered; the workers host the live queries.
        let ts = schema::lift_replicated(self.inner.store.register_table_capture(table)?);
        self.inner
            .schema_engine
            .borrow_mut()
            .register_table(table, ts.clone())?;
        self.inner.fanout.pool.register_table_all(table, ts)?;
        Ok(())
    }

    /// Register a live query under the caller-supplied [`QueryId`] tag. The query is
    /// sharded to a worker, built + hydrated there, and its `Hydrated` baseline emitted
    /// to the channel-out. Returns the **index of the worker that hosts it** (a `BuildError`
    /// surfaces synchronously here as `Err`). Later writes can emit provisional `Changed`
    /// slices before commit. That worker emits `Progressed` after successful commit and
    /// all its slices — so the returned index is what the async progress layer (the drain
    /// thread) keys a query's `cv` position by. The cluster owns the assignment; callers
    /// pass the returned index along rather than recomputing it, so a future scheduling
    /// change (work-stealing) stays internal to the cluster (§2.6).
    ///
    /// **Rejected while the open transaction is speculative**
    /// ([`view_is_speculative`](Self::view_is_speculative)). Registration COLD-hydrates on the
    /// worker, and a worker mid-stream holds the pushed chunks in its sources' overlays — so
    /// the baseline would be drawn from rows no one has committed, while tagged with the
    /// PRE-commit `hydrated_tx` read below. That is strictly worse than the equivalent
    /// one-shot leak: a one-shot is thrown away, but a subscription applies every later
    /// delta on top of its baseline, so a poisoned one breaks `view-after-write ==
    /// fresh-query` for the life of the subscription. (A stream that has pinned snapshots
    /// but pushed nothing is fine — the hydrate is a pure read of the committed post-(N-1)
    /// state; see the `speculative` field docs.)
    ///
    /// Refusing (rather than deferring internally) is forced by the same threading that shapes
    /// [`read_snapshot`](Self::read_snapshot): `WorkerPool::register_query_on` blocks this thread
    /// on the worker's reply, and only this thread can send the transaction's terminal marker — so
    /// a worker that held the registration until commit would deadlock against its own
    /// coordinator. Callers that can wait should re-issue after the commit.
    pub fn query(&self, query_id: QueryId, ast: Ast) -> Result<usize, ReplicaError> {
        if self.inner.fanout.speculative.get() {
            return Err(ReplicaError::Open(
                "cannot register a query while a speculative write transaction is streaming \
                 (its baseline would contain uncommitted rows); retry after commit"
                    .into(),
            ));
        }
        let worker = (query_id.0 % self.inner.fanout.n_workers as u64) as usize;
        let hydrated_tx = self.inner.store.committed_tx_id().0;
        // The sink id is returned for future per-query teardown; unused for now.
        self.inner
            .fanout
            .pool
            .register_query_on(worker, query_id, ast, hydrated_tx)?;
        Ok(worker)
    }

    /// Register a **parameterized query family** (design 310 §5) under `query_id`: one
    /// pipeline over `stripped` (the family's template AST) partitioned on `params`, with
    /// `bindings` bound and hydrated up front — each emits a
    /// [`Update::PartitionHydrated`](crate::Update::PartitionHydrated) instead of a
    /// `Hydrated`. Placed on one worker by the same `query_id % n` rule as
    /// [`query`](Self::query) (design §6, v1: one worker per family). Returns the worker.
    pub fn family(
        &self,
        query_id: QueryId,
        stripped: Ast,
        params: Vec<Box<str>>,
        bindings: Vec<Binding>,
    ) -> Result<usize, ReplicaError> {
        if self.inner.fanout.speculative.get() {
            return Err(ReplicaError::Open(
                "cannot register a query family while a speculative write transaction is \
                 streaming (its baseline would contain uncommitted rows); retry after commit"
                    .into(),
            ));
        }
        let worker = (query_id.0 % self.inner.fanout.n_workers as u64) as usize;
        let hydrated_tx = self.inner.store.committed_tx_id().0;
        self.inner.fanout.pool.register_family_on(
            worker,
            query_id,
            stripped,
            params,
            bindings,
            hydrated_tx,
        )?;
        Ok(worker)
    }

    /// Bind one more partition of the family registered under `query_id` (design 310
    /// §4.4): it hydrates on the hosting worker's latest committed snapshot and emits its
    /// `PartitionHydrated`. Refused, like [`query`](Self::query), while a speculative
    /// transaction is streaming.
    pub fn bind(&self, query_id: QueryId, binding: Binding) -> Result<(), ReplicaError> {
        if self.inner.fanout.speculative.get() {
            return Err(ReplicaError::Open(
                "cannot bind a family partition while a speculative write transaction is \
                 streaming; retry after commit"
                    .into(),
            ));
        }
        let worker = (query_id.0 % self.inner.fanout.n_workers as u64) as usize;
        let hydrated_tx = self.inner.store.committed_tx_id().0;
        self.inner
            .fanout
            .pool
            .bind_partition_on(worker, query_id, binding, hydrated_tx)
    }

    /// Unbind one partition of the family under `query_id`; nothing is emitted. `Ok(false)`
    /// if it was not bound.
    pub fn unbind(&self, query_id: QueryId, binding: Binding) -> Result<bool, ReplicaError> {
        let worker = (query_id.0 % self.inner.fanout.n_workers as u64) as usize;
        self.inner
            .fanout
            .pool
            .unbind_partition_on(worker, query_id, binding)
    }

    /// Analyze `ast` COLD against the current committed state and return where its time
    /// and rows go (`ANALYZE-QUERY-DESIGN.md`). Read-only and **fully self-contained**:
    /// it runs on the CALLING thread over a private connection
    /// ([`analyze_standalone`](crate::analyze_standalone)) — no worker engine, no live
    /// statement cache, no lease — so it never enters a worker's command queue and a
    /// genuinely slow cold hydrate cannot stall (or be stalled by) commits. `BuildError`
    /// (unknown table/column, unsupported shape) surfaces as `Err`.
    pub fn analyze_query(&self, ast: Ast) -> Result<AnalyzeReport, ReplicaError> {
        analyze_standalone(&self.analyze_spec(), &ast)
    }

    /// The `Send` snapshot for running [`analyze_standalone`](crate::analyze_standalone)
    /// on another thread: the daemon's `/analyze` builds one here on its command loop,
    /// then runs the analysis on a dedicated thread so the loop — and with it the write
    /// path — never blocks on a slow cold hydrate.
    pub fn analyze_spec(&self) -> AnalyzeSpec {
        self.inner.fanout.pool.analyze_spec()
    }

    /// Convenience: parse a Zero-wire AST from JSON, then [`Cluster::query`].
    pub fn query_json(&self, query_id: QueryId, json: &str) -> Result<usize, ReplicaError> {
        let ast: Ast = serde_json::from_str(json)
            .map_err(|e| ReplicaError::Schema(format!("invalid AST JSON: {e}")))?;
        self.query(query_id, ast)
    }

    /// Gracefully tear down a query the consumer no longer wants (destroyed /
    /// dematerialized / unsubscribed): reclaim its pipeline on the hosting worker.
    /// Unlike a fault, this is **solicited**, so it emits no [`ClusterEvent`] — after
    /// it returns, no further events arrive for `query_id`. Returns `true` if a live
    /// query was found and removed; idempotent (a repeat call returns `false`). The
    /// reclaimed graph slots are recycled by a later [`query`](Self::query).
    pub fn destroy_query(&self, query_id: QueryId) -> bool {
        let worker = (query_id.0 % self.inner.fanout.n_workers as u64) as usize;
        self.inner.fanout.pool.deregister_query_on(worker, query_id) > 0
    }

    /// Re-read the **current assembled view** of `query_id` as a one-shot snapshot — the
    /// SSR read path (`SSR-DESIGN.md` §3). Routes to the hosting worker (same `query_id % n`
    /// shard as [`query`](Self::query)) and returns the live view as hydration `Add`s, with
    /// no subscription and no streaming state. A degraded shard or an unregistered query
    /// yields an empty snapshot (the one-shot reports "no rows" rather than failing); a
    /// raised read boundary on a live worker (§5.3 sum overflow) is `Err` — never a `NULL`
    /// aggregate rendered as data. A snapshot is refused while the open transaction is
    /// **speculative** ([`view_is_speculative`](Self::view_is_speculative)): it has already
    /// streamed uncommitted changes into the worker graph, and only its terminal commit marker
    /// makes those changes publicly visible. Callers that can wait should poll that predicate and
    /// defer the read to the commit rather than surfacing this error — a caller pinned to one
    /// thread with the writer (rindled's engine loop) must NOT block here: the same thread has to
    /// keep running to reach the commit.
    pub fn read_snapshot(&self, query_id: QueryId) -> Result<Vec<CaughtChange>, ReplicaError> {
        if self.inner.fanout.speculative.get() {
            return Err(ReplicaError::ReadRejected(
                "view snapshot unavailable while a speculative write transaction is streaming; \
                 retry after commit"
                    .into(),
            ));
        }
        let worker = (query_id.0 % self.inner.fanout.n_workers as u64) as usize;
        self.inner.fanout.pool.read_snapshot_on(worker, query_id)
    }

    /// Whether the workers' assembled views currently carry **uncommitted** rows — true from the
    /// moment an open transaction streams its first chunk (`PUSH_CHUNK_ROWS`) until that
    /// transaction commits or rolls back. This is the exact window in which
    /// [`read_snapshot`](Self::read_snapshot) refuses, exposed as a predicate so a caller can
    /// QUEUE a one-shot read until the transaction closes instead of failing it. An open
    /// transaction that never crosses the streaming threshold is *not* speculative — it holds
    /// its changes on the writer, so reads stay serviceable.
    pub fn view_is_speculative(&self) -> bool {
        self.inner.fanout.speculative.get()
    }

    /// Open the single-writer transaction. Errors if one is already open.
    pub fn write(&self) -> Result<ClusterWriteTxn, ReplicaError> {
        let core = ApplyTxn::begin(self.inner.store.clone(), self.fanout_rc())?;
        // Nothing streamed yet: reads stay serviceable until this txn crosses `PUSH_CHUNK_ROWS`.
        self.inner.fanout.speculative.set(false);
        Ok(ClusterWriteTxn { core })
    }

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

    /// The observed writer connection, for crate-internal host bookkeeping while the coordinator
    /// thread is idle. Kept crate-private so external callers cannot write around capture.
    pub(crate) fn writer_connection(&self) -> &Connection {
        self.inner.store.writer_connection()
    }

    /// 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.inner.store.schema_cookie()
    }

    /// Walk every declared foreign key and report violating rows — the **opt-in audit**
    /// a replica runs instead of paying for enforcement on every applied row.
    ///
    /// `PRAGMA foreign_key_check` reports what is in the file, not what SQLite would have
    /// refused, so it answers the same question whether this cluster was opened
    /// [`ForeignKeys::Enforced`](crate::ForeignKeys) or
    /// [`ForeignKeys::Unenforced`](crate::ForeignKeys) — which is exactly why a follower
    /// can apply with enforcement off and still prove referential integrity on demand.
    /// It is a full scan of the referencing tables: an operator command, a post-restore
    /// gate, or a soak assertion, never a per-commit step. Runs on the read-only
    /// connection, so it takes no write lock and does not disturb an applier.
    ///
    /// Pass [`FOREIGN_KEY_AUDIT_ROW_CAP`](crate::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<crate::ForeignKeyAudit, ReplicaError> {
        self.inner.store.foreign_key_audit(max_rows)
    }

    /// Block until every worker has drained its command queue — i.e. emitted the
    /// `Hydrated`/`Changed` events for all preceding [`query`](Self::query) /
    /// `commit` calls to the channel-out. A quiesce/test affordance (a steady-state
    /// consumer just drains the channel continuously); relies on per-worker FIFO
    /// command order, so a returned ping implies all prior work is done.
    ///
    /// Doubles as the liveness sweep: a worker that died (or hung) is detected here and
    /// **respawned** (its lost queries delivered a terminal [`ClusterEvent::Faulted`]),
    /// so a steady-state consumer that drains + periodically `sync()`s also self-heals.
    pub fn sync(&self) {
        self.inner.fanout.pool.reap();
    }

    /// Test seam: force a derivation-fault recovery on the worker that hosts
    /// `on_query` — it tears down **every** query on that worker (they share one
    /// graph) and emits a terminal [`ClusterEvent::Faulted`] for each, then rebuilds
    /// a clean engine. Exposed (doc-hidden) because a real derive fault can't be
    /// induced with valid data; not part of the supported API.
    #[doc(hidden)]
    pub fn __test_trigger_fault(&self, on_query: QueryId) {
        let worker = (on_query.0 % self.inner.fanout.n_workers as u64) as usize;
        self.inner.fanout.pool.trigger_fault(worker);
    }

    /// Test seam: **kill the worker thread** hosting `on_query` (simulating a true
    /// thread death). The death is detected + recovered on the next [`sync`](Self::sync)
    /// or `commit`. Doc-hidden; not part of the supported API.
    #[doc(hidden)]
    pub fn __test_kill_worker(&self, on_query: QueryId) {
        let worker = (on_query.0 % self.inner.fanout.n_workers as u64) as usize;
        self.inner.fanout.pool.kill_worker(worker);
    }

    /// Test seam: whether the shard hosting `on_query` is permanently degraded (the
    /// crash-loop breaker tripped). Doc-hidden; not part of the supported API.
    #[doc(hidden)]
    pub fn __test_is_degraded(&self, on_query: QueryId) -> bool {
        let worker = (on_query.0 % self.inner.fanout.n_workers as u64) as usize;
        self.inner.fanout.pool.is_degraded(worker)
    }

    /// Test seam: lower the barrier-ack + liveness-ping timeouts so the death/stall
    /// recovery paths run fast. Doc-hidden; not part of the supported API.
    #[doc(hidden)]
    pub fn __test_set_timeouts(&self, ack: std::time::Duration, ping: std::time::Duration) {
        self.inner.fanout.pool.set_timeouts(ack, ping);
    }

    /// Set the per-push derive deadline (FOLLOWER-LAG-SHED §6.6 — the runaway-push bail),
    /// in milliseconds; `0` disables it. A single chunk's derive exceeding this faults the
    /// worker's queries promptly and cleanly (`FaultCause::PushDeadline`, no detached
    /// thread) instead of wedging until the barrier watchdog detaches the whole worker.
    /// Applies immediately to every worker, respawns included. Default:
    /// 10 s, aligned with the barrier watchdog it pre-empts.
    pub fn set_push_deadline_ms(&self, ms: u64) {
        self.inner.fanout.pool.set_push_deadline_ms(ms);
    }

    /// Set the join membership pre-check bounds
    /// (`designs/311-JOIN-MEMBERSHIP-PRECHECK-DESIGN.md` §8) on every worker engine —
    /// respawns and post-fault rebuilds included. `per_join` is the distinct-key bound one
    /// join may track (`None` = off, the engine default); `per_graph` caps the total across
    /// a worker's joins. Applied at each worker's next commit barrier, before anything
    /// derives; a change resets the workers' sets, which rebuild by observing the next
    /// hydrate-shaped fetch through each join.
    pub fn set_join_precheck_bounds(&self, per_join: Option<usize>, per_graph: usize) {
        self.inner
            .fanout
            .pool
            .set_join_precheck_bounds(per_join, per_graph);
    }

    /// Test seam: the crash-loop restart budget (respawns beyond it degrade the shard).
    /// Doc-hidden; not part of the supported API.
    #[doc(hidden)]
    pub fn __test_restart_budget(&self) -> usize {
        self.inner.fanout.pool.restart_budget()
    }

    /// Test seam: the join membership pre-check report of the worker hosting `on_query` —
    /// the bounds in force on that worker's engine and the state of every join of every
    /// query it hosts (see [`JoinPrecheckReport`](crate::JoinPrecheckReport)). Answered
    /// FIFO behind that worker's earlier registrations and commits, so it reflects exactly
    /// what the next registration on it would find. `None` if the shard is degraded.
    /// Doc-hidden; not part of the supported API.
    #[doc(hidden)]
    pub fn __test_join_precheck_report(
        &self,
        on_query: QueryId,
    ) -> Option<crate::JoinPrecheckReport> {
        let worker = (on_query.0 % self.inner.fanout.n_workers as u64) as usize;
        self.inner.fanout.pool.join_precheck_report_on(worker)
    }

    /// Test seam: force the `FlippedJoin` IN-batch chunk size on every worker engine
    /// (`None` restores the engine default of 256), so a small fixture routes its parent
    /// fetches through the chunked k-way `merge_node_streams` instead of the single-batch
    /// path. The engine's own seam (`rindle::op::set_multi_constraint_chunk_size_for_test`)
    /// is a **thread-local** read at `FlippedJoin::new`, and a cluster builds its pipelines
    /// on worker threads, so setting it on the caller's thread reaches nothing; this
    /// carries it to the workers, applied before each of their commands so registrations,
    /// binds, post-fault rebuilds and respawns all see it. Already-built joins keep the
    /// size they were constructed with. Doc-hidden; not part of the supported API.
    #[doc(hidden)]
    #[cfg(feature = "testkit")]
    pub fn __test_set_flipped_join_chunk_size(&self, n: Option<usize>) {
        self.inner.fanout.pool.set_flipped_join_chunk_size(n);
    }

    /// Test seam: the number of `(key, value)` entries in the operator scratch storage of
    /// `on_query`'s pipeline on its worker — the leak probe of design 310 impl plan D5
    /// (`family_pipeline.rs::unbind_leaves_no_zombie_state`) lifted to the cluster
    /// boundary, so the fuzz lanes can assert an unbind actually released its partition's
    /// slots instead of leaving a zombie the tree parity cannot see. Answered FIFO behind
    /// that worker's earlier binds, unbinds and commits. `None` if the shard is degraded
    /// or nothing is registered under `on_query`. Doc-hidden; not part of the supported API.
    #[doc(hidden)]
    pub fn __test_storage_entries(&self, on_query: QueryId) -> Option<usize> {
        let worker = (on_query.0 % self.inner.fanout.n_workers as u64) as usize;
        self.inner
            .fanout
            .pool
            .storage_report_on(worker, on_query, false)
            .map(|r| r.entries)
    }

    /// Test seam: every `(slot, key, value)` of `on_query`'s operator scratch storage —
    /// the diagnostic twin of [`__test_storage_entries`](Self::__test_storage_entries),
    /// for the failure message when a leak probe finds more than it expected. Doc-hidden;
    /// not part of the supported API.
    #[doc(hidden)]
    pub fn __test_storage_dump(&self, on_query: QueryId) -> Vec<(usize, String, String)> {
        let worker = (on_query.0 % self.inner.fanout.n_workers as u64) as usize;
        self.inner
            .fanout
            .pool
            .storage_report_on(worker, on_query, true)
            .map(|r| r.dump)
            .unwrap_or_default()
    }

    /// Run SQL against a physically read-only connection using SQLite snapshot semantics.
    /// This connection does not see the writer's uncommitted changes. Schema changes
    /// belong in [`Self::exec_ddl`], and row mutations in [`Self::write`].
    pub fn read<T>(
        &self,
        f: impl FnOnce(&Connection) -> rusqlite::Result<T>,
    ) -> Result<T, ReplicaError> {
        self.inner.store.read(f)
    }

    /// The hierarchical view [`Schema`] `ast` materializes to over the registered tables —
    /// the coordinator parity for [`Db::view_schema`](crate::Db::view_schema). Derived from
    /// the schema-only engine (the same source-schema mapping the workers use), so the
    /// shape, sort, `singular` flag, and in-view relationships line up with the change
    /// stream. A layer that ships views to a remote (the normalized wire schema + `hello`)
    /// needs this without registering the query. `BuildError` (unknown table/column)
    /// surfaces as `Err`.
    pub fn view_schema(&self, ast: &Ast) -> Result<Schema, ReplicaError> {
        self.inner.schema_engine.borrow().view_schema(ast)
    }

    /// Run one **online maintenance pass** on the writer connection: refresh planner stats
    /// (`PRAGMA optimize`, bounded by the `analysis_limit` set at open), return up to
    /// `incremental_vacuum_pages` freelist pages to the OS once at least `freelist_threshold_pages`
    /// have built up, and take a non-blocking PASSIVE WAL checkpoint — see [`MaintenanceOptions`].
    ///
    /// Designed to run on a timer in the gap between commits on the coordinator thread: each step
    /// is bounded (no full `VACUUM`/`ANALYZE`, no exclusive lock) and the maintenance writes land
    /// only in `sqlite_stat*` / the freelist, which the CDC capture hook ignores. It is **skipped**
    /// (returning `MaintenanceReport::skipped()`) while a write transaction is open — e.g. a
    /// follower stream holds one open across frames — so it never disturbs an in-flight txn.
    ///
    /// [`MaintenanceOptions`]: crate::MaintenanceOptions
    pub fn maintain(
        &self,
        opts: &crate::MaintenanceOptions,
    ) -> Result<crate::MaintenanceReport, ReplicaError> {
        if self.inner.store.in_write() {
            return Ok(crate::MaintenanceReport::skipped());
        }
        crate::maintenance::run(self.inner.store.writer_connection(), opts)
    }

    /// Checkpoint and truncate the WAL through the coordinator 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<crate::WalCheckpoint, ReplicaError> {
        self.inner.store.checkpoint_truncate()
    }

    /// Run schema DDL (`CREATE`/`ALTER`/`DROP`/`REINDEX`) against the writer connection — the
    /// supported way to define the plain base tables you then [`register_table`](Self::register_table)
    /// 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 the workers until commit). Mirrors [`Db::exec_ddl`](crate::Db::exec_ddl).
    pub fn exec_ddl(&self, sql: &str) -> Result<(), ReplicaError> {
        self.inner.store.exec_ddl(sql)
    }

    /// Apply schema `statements` AND stamp a durable idempotency marker (`key` → `marker_table`)
    /// in ONE ordinary transaction on the writer connection. See
    /// [`ApplyStore::exec_ddl_with_marker`] — this is the same primitive, delegated (design
    /// 309). A marker and its DDL commit together, so retrying an applied migration
    /// can detect the marker without applying the statements again.
    pub fn exec_ddl_with_marker(
        &self,
        marker_table: &str,
        key: &str,
        statements: &[String],
    ) -> Result<schema::DdlApplyReport, ReplicaError> {
        self.inner
            .store
            .exec_ddl_with_marker(marker_table, key, statements)
    }

    /// [`exec_ddl_with_marker`](Self::exec_ddl_with_marker), with one caller-owned bookkeeping
    /// hook invoked after every real statement and its authorizer actions. See
    /// [`ApplyStore::exec_ddl_with_marker_and_step_effects`].
    pub fn exec_ddl_with_marker_and_step_effects<F>(
        &self,
        marker_table: &str,
        key: &str,
        statements: &[String],
        apply_step_effects: F,
    ) -> Result<schema::DdlApplyReport, ReplicaError>
    where
        F: FnMut(&Connection, &schema::DdlStep) -> rusqlite::Result<()>,
    {
        self.inner.store.exec_ddl_with_marker_and_step_effects(
            marker_table,
            key,
            statements,
            apply_step_effects,
        )
    }

    /// Run a **full** `ANALYZE` on the writer to build deep planner statistics (`sqlite_stat4`),
    /// not just the bounded `sqlite_stat1` the maintenance tick refreshes. Call this once after a
    /// bulk load / seed (or periodically) to sharpen the cost model. Restores the bounded
    /// `analysis_limit` afterward, so it does not slow the ongoing maintenance tick. Rejected
    /// while a write transaction is open.
    ///
    /// This used to be described as what makes an `ORDER BY … LIMIT n` displacement re-fetch seek
    /// rather than scan (GitHub #68). It is not, since the query builder began lifting a sargable
    /// leading-column bound — see `maintenance::analyze_full`.
    pub fn analyze_full(&self) -> Result<(), ReplicaError> {
        if self.inner.store.in_write() {
            return Err(ReplicaError::Open(
                "cannot run a full ANALYZE while a write transaction is open".into(),
            ));
        }
        crate::maintenance::analyze_full(self.inner.store.writer_connection())
    }

    /// One-time (idempotent) setup for the client-mutations protocol (the coordinator
    /// parity for [`Db::enable_client_mutations`](crate::Db::enable_client_mutations)):
    /// create [`CLIENT_MUTATIONS_TABLE`] and register it like any base table (CDC
    /// capture + sources on the schema engine and every worker), so `lmid` rides the
    /// change stream co-transactionally with the data and each client's one-row system
    /// query can host it. Rejected mid-write-txn.
    pub fn enable_client_mutations(&self) -> Result<(), ReplicaError> {
        self.inner.store.create_client_mutations_tables()?;
        self.register_table(CLIENT_MUTATIONS_TABLE)
    }

    /// One-time (idempotent) setup for the §4 realtime lifecycle (the coordinator parity
    /// for [`Db::enable_realtime_lifecycle`](crate::Db::enable_realtime_lifecycle) — see
    /// its doc for why every table is REGISTERED, not just created): the scope-session
    /// doorbell (§4.1), the downgrade watermark fence (§4.2), the mutation-outcomes
    /// resolution surface (populated by Slice I-ii), plus
    /// [`ROOM_CLIENT_MUTATIONS_TABLE`] — §7.1's "ordinary footprint data after
    /// downgrade". Requires [`enable_client_mutations`](Self::enable_client_mutations)
    /// first (it owns the room ledger's DDL); rejected mid-write-txn.
    pub fn enable_realtime_lifecycle(&self) -> Result<(), ReplicaError> {
        if self.inner.store.in_write() {
            return Err(ReplicaError::Open(
                "cannot enable the realtime lifecycle while a write transaction is open".into(),
            ));
        }
        if !self.inner.store.capture().has_table(CLIENT_MUTATIONS_TABLE) {
            return Err(ReplicaError::Mutation(
                "realtime lifecycle requires client mutations — call enable_client_mutations() \
                 first (it owns the room ledger's DDL)"
                    .into(),
            ));
        }
        self.inner
            .store
            .writer_connection()
            .execute_batch(&realtime_lifecycle_ddl())
            .map_err(|e| ReplicaError::sqlite("create realtime lifecycle tables", e))?;
        for table in [
            SCOPE_SESSIONS_TABLE,
            ROOM_WATERMARK_TABLE,
            ROOM_MUTATION_OUTCOMES_TABLE,
            ROOM_CLIENT_MUTATIONS_TABLE,
        ] {
            self.register_table(table)?;
        }
        Ok(())
    }

    /// Whether [`enable_realtime_lifecycle`](Self::enable_realtime_lifecycle) has run —
    /// the cheap in-memory gate (`require_client_mutations`' idiom: capture registration
    /// is the enable's observable effect) that `commit_room_flush` consults before its
    /// §4.2 watermark co-commit, so a daemon that never enabled the lifecycle keeps
    /// applying room flushes exactly as before.
    pub(crate) fn realtime_lifecycle_enabled(&self) -> bool {
        self.inner.store.capture().has_table(ROOM_WATERMARK_TABLE)
    }

    /// The high-water mutation id durably recorded for `client_id` (0 if none — a new
    /// client). Coordinator parity for [`Db::client_lmid`](crate::Db::client_lmid);
    /// requires [`enable_client_mutations`](Self::enable_client_mutations).
    pub fn client_lmid(&self, client_id: &str) -> Result<u64, ReplicaError> {
        self.require_client_mutations()?;
        let got = self
            .read(|c| {
                c.query_row(
                    &format!(
                        "SELECT last_mutation_id FROM {CLIENT_MUTATIONS_TABLE} \
                         WHERE client_id = ?1"
                    ),
                    [client_id],
                    |r| r.get::<_, i64>(0),
                )
                .optional()
            })?
            .map(|v| v as u64);
        Ok(got.unwrap_or(0))
    }

    /// Apply a client's mutation push over the cluster (the coordinator parity for
    /// [`Db::apply_mutations`](crate::Db::apply_mutations)): each envelope's mutator runs
    /// in **its own** transaction (with the co-transactional `lmid` upsert), in `mid` order.
    /// Same per-envelope rules as the single-thread path — `mid ≤ lmid` skipped (idempotent
    /// redelivery), `mid == lmid + 1` applied, a gap rejected at that envelope; a mutator
    /// error / panic / unknown name rolls effects back and commits `lmid` only
    /// (processed-as-no-op — no rejection signal).
    ///
    /// Unlike the `Db` path it does **not** return the changed-query set inline — every
    /// commit's batches (including the lmid row on the client's own system query) flow
    /// asynchronously through the drain. Must be called with no write transaction open
    /// (it manages its own).
    pub fn apply_mutations(
        &self,
        registry: &MutatorRegistry,
        envelopes: &[MutationEnvelope],
    ) -> Result<MutationOutcome, ReplicaError> {
        use std::panic::{catch_unwind, AssertUnwindSafe};
        self.require_client_mutations()?;
        let mut out = MutationOutcome::default();
        for env in envelopes {
            let stored = self.client_lmid(&env.client_id)?;
            if env.mid <= stored {
                continue; // duplicate redelivery — already processed
            }
            if env.mid != stored + 1 {
                return Err(ReplicaError::Mutation(format!(
                    "mutation id gap for client {:?}: expected {}, got {}",
                    env.client_id,
                    stored + 1,
                    env.mid
                )));
            }

            let mut txn = self.write()?;
            let invoked: Result<(), MutationReject> = match registry.get(&env.name) {
                None => Err(MutationReject(format!("unknown mutator {:?}", env.name))),
                Some(f) => match catch_unwind(AssertUnwindSafe(|| f(&mut txn, &env.args))) {
                    Ok(r) => r,
                    Err(_) => Err(MutationReject(format!("mutator {:?} panicked", env.name))),
                },
            };

            match invoked {
                Ok(()) => {
                    upsert_lmid(&mut txn, &env.client_id, env.mid)?;
                    out.commits.push(txn.commit_with_info()?);
                }
                Err(MutationReject(reason)) => {
                    txn.rollback();
                    // The lmid-only commit: the durable record that `mid` was processed
                    // (as a no-op). Its capture is the lmid row, which derives through the
                    // client's own system query and snaps the pending prediction back on
                    // release. The reason is server-side observability only.
                    eprintln!(
                        "[rindle-replica] mutation {} for client {:?} failed (lmid still advances): {reason}",
                        env.mid, env.client_id
                    );
                    let mut lmid_txn = self.write()?;
                    upsert_lmid(&mut lmid_txn, &env.client_id, env.mid)?;
                    out.commits.push(lmid_txn.commit_with_info()?);
                }
            }
        }
        Ok(out)
    }

    fn require_client_mutations(&self) -> Result<(), ReplicaError> {
        if !self.inner.store.capture().has_table(CLIENT_MUTATIONS_TABLE) {
            return Err(ReplicaError::Mutation(
                "client mutations not enabled — call enable_client_mutations() first".into(),
            ));
        }
        Ok(())
    }
}

/// An open write transaction on the cluster's single writer connection. Run SQL
/// with [`exec`](Self::exec)/[`exec_batch`](Self::exec_batch); [`commit`](Self::commit)
/// runs the snapshot/commit handshake. Workers can deliver provisional changes while
/// SQL is still running. Dropping rolls back SQL; if chunks were already streamed,
/// affected queries fault and need a new registration. See [`ClusterEvent`].
///
/// A thin wrapper over the apply plane's [`ApplyTxn`] carrying the cluster's fan-out
/// (design 309 §3): the state machine — capture pumping, the barriers, the commit
/// guards — is ONE implementation shared with the headless applier.
pub struct ClusterWriteTxn {
    core: ApplyTxn,
}

impl ClusterWriteTxn {
    /// Run one statement with positional parameters inside the open transaction.
    pub fn exec(&mut self, sql: &str, params: &[OwnedValue]) -> Result<usize, ReplicaError> {
        self.core.exec(sql, params)
    }

    /// Run one mutation statement under the shared writer time/VM budget and, for guarded public
    /// mutations, the reserved-object authorizer. This is the bounded twin of [`exec`](Self::exec)
    /// used by a standalone daemon's write plane; it still pumps capture between statements.
    pub(crate) fn exec_bounded(
        &mut self,
        sql: &str,
        params: &[OwnedValue],
        guarded: bool,
    ) -> Result<usize, StatementRunError> {
        self.core.exec_bounded(sql, params, guarded)
    }

    /// Run one v1 public-SQL statement through the shared classifier/authorizer and writer
    /// budget. Interactive calls request a savepoint so ordinary statement failures preserve the
    /// surrounding transaction; one-shot batches let their coordinator roll the whole unit back.
    pub(crate) fn public_statement(
        &mut self,
        request: &SqlStatementRequest,
        preserve_on_error: bool,
        result_byte_limit: usize,
    ) -> Result<StatementResult, StatementRunError> {
        self.core
            .public_statement(request, preserve_on_error, result_byte_limit)
    }

    pub(crate) fn is_open(&self) -> bool {
        self.core.is_open()
    }

    pub(crate) fn captured_user_event_count(&self) -> usize {
        self.core.captured_user_event_count()
    }

    pub(crate) fn next_tx_id(&self) -> TxId {
        self.core.next_tx_id()
    }

    /// The open transaction's connection for narrow crate-internal bookkeeping helpers. External
    /// callers stay on typed mutation methods so they cannot bypass capture accidentally.
    pub(crate) fn connection(&self) -> &Connection {
        self.core.connection()
    }

    /// The underlying apply-plane transaction, for the delegating apply surface
    /// ([`ClusterConsumer`](crate::ClusterConsumer)).
    pub(crate) fn core_mut(&mut self) -> &mut ApplyTxn {
        &mut self.core
    }

    /// Consume the wrapper, yielding the apply-plane transaction (the commit_* delegates).
    pub(crate) fn into_core(self) -> ApplyTxn {
        self.core
    }

    /// Run a batch of statements (no parameters) inside the open transaction.
    pub fn exec_batch(&mut self, sql: &str) -> Result<(), ReplicaError> {
        self.core.exec_batch(sql)
    }

    /// Run a read **through the open transaction** (sees its own uncommitted writes — the
    /// read-dependent mutator contract, §4.1), each cell mapped from its raw SQLite storage
    /// class. The [`MutationSql`] query flavor for the parallel write path.
    pub fn query(
        &mut self,
        sql: &str,
        params: &[OwnedValue],
    ) -> Result<Vec<Vec<OwnedValue>>, ReplicaError> {
        self.core.query(sql, params)
    }

    /// [`query`](Self::query), additionally reporting the result's column names in order —
    /// what a network front needs to answer a mutator-session read (`{cols, rows}` on the
    /// wire, zipped client-side; DAEMON-INTERACTIVE-TXN-DESIGN.md §4.1). Same open-transaction
    /// read-your-writes semantics and raw-storage-class cell mapping.
    pub fn query_with_cols(
        &mut self,
        sql: &str,
        params: &[OwnedValue],
    ) -> Result<(Vec<String>, Vec<Vec<OwnedValue>>), ReplicaError> {
        self.core.query_with_cols(sql, params)
    }

    /// Finish capture, wait for worker snapshot acknowledgments, then commit SQL and
    /// its watermark. Returns the transaction ID without waiting for every derived
    /// event to reach the receiver. Workers may already have sent provisional changes;
    /// release them only after their `Progressed` markers confirm the commit.
    pub fn commit(self) -> Result<TxId, ReplicaError> {
        self.core.commit()
    }

    /// [`commit`](Self::commit), additionally reporting the transaction's [`CommitInfo`] —
    /// the `cv` to stamp on outgoing batches.
    ///
    /// The whole capture — INCLUDING any `_rindle_client_mutations` rows — fans out to the
    /// workers: the lmid table is engine-hosted like any base table, so a client's lmid
    /// advance derives through its own system query and is released by the same `cv_min`
    /// as the commit's data (no metadata side-channel to race). An empty capture still
    /// crosses the worker barrier: it has no data to derive, but every worker must emit
    /// `Progressed(N)` so an older live query cannot pin a later query's seq-0 snapshot
    /// below its hydrate CV forever.
    pub fn commit_with_info(self) -> Result<CommitInfo, ReplicaError> {
        self.core.commit_with_info()
    }

    /// Commit host-local metadata when the public SQL unit captured no application effects.
    /// This deliberately does not mint a `TxId`: the retained outcome has `cursor = NULL`, and
    /// no worker progress boundary exists for a metadata-only cache write.
    pub(crate) fn commit_metadata_only(self) -> Result<(), ReplicaError> {
        self.core.commit_metadata_only()
    }

    /// Explicitly roll back. Leaves every view untouched; delivers nothing.
    pub fn rollback(self) {
        self.core.rollback()
    }
}

/// The SQL `MutationTx` flavor (design §4.2) for the parallel write path — a server mutator
/// runs against the open cluster transaction, reading through the same connection so it
/// sees its own uncommitted writes (and lower-`mid` mutations' effects). Mirrors the
/// single-thread `impl MutationSql for WriteTxn`.
impl MutationSql for ClusterWriteTxn {
    fn exec(&mut self, sql: &str, params: &[OwnedValue]) -> Result<usize, ReplicaError> {
        ClusterWriteTxn::exec(self, sql, params)
    }

    fn query(
        &mut self,
        sql: &str,
        params: &[OwnedValue],
    ) -> Result<Vec<Vec<OwnedValue>>, ReplicaError> {
        ClusterWriteTxn::query(self, sql, params)
    }
}

#[cfg(test)]
mod stream_pump_tests {
    use super::*;
    use crate::parallel::PUSH_CHUNK_ROWS;

    fn iv(n: i64) -> OwnedValue {
        OwnedValue::Int(n)
    }

    /// The writer forwards captured changes to the workers **between statements**
    /// (`maybe_pump`), so a large transaction's capture buffer stays bounded by ~one
    /// `PUSH_CHUNK_ROWS` chunk instead of growing to the whole txn
    /// (`CLUSTER-INCREMENTAL-FANOUT-DESIGN.md` §7). This is the writer half of the `O(chunk)`
    /// memory bound (the per-worker output half landed in Slice 2). Single-row inserts mirror
    /// the replicator follower's `apply_muts` apply shape, where the hook fires once per row so
    /// the flush lands exactly on the chunk boundary (no mega-statement on the follower).
    #[test]
    fn writer_capture_buffer_stays_chunk_bounded_during_a_large_txn() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("pump.db");
        let (cluster, _events) = Cluster::open(&path, 2).unwrap();
        cluster
            .exec_ddl("CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER)")
            .unwrap();
        cluster.register_table("t").unwrap();

        let mut w = cluster.write().unwrap();
        let rows = (PUSH_CHUNK_ROWS * 3) as i64; // well past one chunk
        let mut high = 0usize;
        for id in 0..rows {
            w.exec("INSERT INTO t VALUES (?,?)", &[iv(id), iv(id * 2)])
                .unwrap();
            high = high.max(w.core.store().capture().buffer_len());
        }
        // Without between-statement pumping this would reach `rows` (3× chunk); with it the
        // writer flushes at the chunk boundary, so the buffer never holds a whole txn.
        assert!(
            high <= PUSH_CHUNK_ROWS,
            "capture-buffer high-water {high} exceeded one chunk ({PUSH_CHUNK_ROWS}) — \
             the writer is not draining as it writes"
        );
        assert!(high > 0, "sanity: the writer captured rows");
        w.commit().unwrap();
        assert_eq!(cluster.committed_tx_id(), TxId(1));
    }

    #[test]
    fn cached_writer_statement_auto_reprepares_after_schema_change() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("cached-writer-schema.db");
        let (cluster, _events) = Cluster::open(&path, 1).unwrap();
        cluster
            .exec_ddl("CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER)")
            .unwrap();
        cluster.register_table("t").unwrap();

        let mut first = cluster.write().unwrap();
        first
            .exec("INSERT INTO t VALUES (?,?)", &[iv(1), iv(10)])
            .unwrap();
        first.commit().unwrap();

        // Changes the schema cookie while leaving the mutation shape intact. The second write
        // checks out the cached VM; sqlite3_step automatically recompiles a v2/v3-prepared
        // statement rather than requiring an application-side cache flush.
        cluster
            .exec_ddl("CREATE INDEX t_v ON t(v)")
            .expect("schema change commits");
        let mut second = cluster.write().unwrap();
        second
            .exec("INSERT INTO t VALUES (?,?)", &[iv(2), iv(20)])
            .expect("cached statement automatically reprepares");
        second.commit().unwrap();

        let count: i64 = cluster
            .read(|connection| connection.query_row("SELECT count(*) FROM t", [], |row| row.get(0)))
            .unwrap();
        assert_eq!(count, 2);
    }

    #[test]
    fn one_shot_snapshot_refuses_speculative_streamed_state() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("snapshot-fence.db");
        let (cluster, _events) = Cluster::open(&path, 1).unwrap();
        cluster
            .exec_ddl("CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER)")
            .unwrap();
        cluster.register_table("t").unwrap();
        let query_id = QueryId(1);
        cluster.query(query_id, rindle::table("t").build()).unwrap();

        let mut write = cluster.write().unwrap();
        for id in 0..(PUSH_CHUNK_ROWS as i64 + 1) {
            write
                .exec("INSERT INTO t VALUES (?,?)", &[iv(id), iv(id * 2)])
                .unwrap();
        }
        assert!(cluster.view_is_speculative());
        let error = cluster.read_snapshot(query_id).unwrap_err();
        assert!(
            matches!(error, ReplicaError::ReadRejected(_)),
            "open write exposed a one-shot view snapshot: {error}"
        );

        write.commit().unwrap();
        assert!(!cluster.view_is_speculative());
        assert_eq!(
            cluster.read_snapshot(query_id).unwrap().len(),
            PUSH_CHUNK_ROWS + 1
        );
    }

    /// The fence is scoped to SPECULATIVE state, not to "a write is open". A txn that never
    /// crosses `PUSH_CHUNK_ROWS` streams nothing, so every worker's view still sits at the last
    /// commit and the one-shot stays serviceable throughout — the read sees the PRE-txn state
    /// (committed), never the open txn's rows. Without this scoping, rindled's follower — which
    /// parks an open aggregate txn between upstream runs (FOLLOWER-LAG-SHED §5) — would reject
    /// SSR one-shots for essentially the whole batching window.
    #[test]
    fn one_shot_snapshot_serves_committed_state_under_a_small_open_write() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("snapshot-small-txn.db");
        let (cluster, _events) = Cluster::open(&path, 1).unwrap();
        cluster
            .exec_ddl("CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER)")
            .unwrap();
        cluster.register_table("t").unwrap();
        let query_id = QueryId(1);
        cluster.query(query_id, rindle::table("t").build()).unwrap();

        let mut seed = cluster.write().unwrap();
        seed.exec("INSERT INTO t VALUES (?,?)", &[iv(0), iv(0)])
            .unwrap();
        seed.commit().unwrap();

        // Well under the streaming threshold: nothing is pushed to the workers.
        let mut write = cluster.write().unwrap();
        for id in 1..10i64 {
            write
                .exec("INSERT INTO t VALUES (?,?)", &[iv(id), iv(id * 2)])
                .unwrap();
        }
        assert!(!cluster.view_is_speculative());
        assert_eq!(
            cluster.read_snapshot(query_id).unwrap().len(),
            1,
            "a non-streaming open txn must serve the last COMMITTED view, not its own rows"
        );

        write.commit().unwrap();
        assert_eq!(cluster.read_snapshot(query_id).unwrap().len(), 10);
    }

    /// Registering a query mid-stream must NOT hydrate from the speculative worker connection.
    /// Verified failure before the guard: the baseline came back with `PUSH_CHUNK_ROWS + 1` rows
    /// (the seed plus a whole uncommitted transaction) tagged with the PRE-commit `tx_id` — a
    /// subscription baseline that claims to be one commit and contains another. Unlike the
    /// one-shot leak this is durable damage: every later delta folds onto that wrong baseline.
    #[test]
    fn registering_a_query_midstream_is_refused_and_correct_after_commit() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("midstream-register.db");
        let (cluster, events) = Cluster::open(&path, 1).unwrap();
        cluster
            .exec_ddl("CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER)")
            .unwrap();
        cluster.register_table("t").unwrap();

        let mut seed = cluster.write().unwrap();
        seed.exec("INSERT INTO t VALUES (?,?)", &[iv(0), iv(0)])
            .unwrap();
        seed.commit().unwrap();
        while events.try_recv().is_ok() {}

        let mut write = cluster.write().unwrap();
        for id in 1..=(PUSH_CHUNK_ROWS as i64 + 1) {
            write
                .exec("INSERT INTO t VALUES (?,?)", &[iv(id), iv(id)])
                .unwrap();
        }
        assert!(cluster.view_is_speculative());

        let qid = QueryId(7);
        let refused = cluster
            .query(qid, rindle::table("t").build())
            .expect_err("registering against a streaming txn must be refused");
        assert!(
            matches!(refused, ReplicaError::Open(_)),
            "unexpected refusal kind: {refused}"
        );

        // After the commit the same registration is serviceable, and its baseline is the WHOLE
        // committed table tagged at the post-commit watermark.
        write.commit().unwrap();
        assert!(!cluster.view_is_speculative());
        let committed = cluster.committed_tx_id();
        cluster.query(qid, rindle::table("t").build()).unwrap();

        let mut baseline = None;
        while let Ok(event) = events.try_recv() {
            if let crate::ClusterEvent::Update {
                query_id,
                update: crate::Update::Hydrated { tx_id, changes },
            } = event
            {
                if query_id == qid {
                    baseline = Some((tx_id, changes.len()));
                }
            }
        }
        assert_eq!(
            baseline,
            Some((committed, PUSH_CHUNK_ROWS + 2)),
            "baseline must be the committed table at the post-commit watermark"
        );
    }

    /// Teardown is the SAFE direction mid-stream, and the worker's FIFO comment now says so:
    /// reclaiming a query while a transaction is streaming must leave that transaction able to
    /// commit and must not disturb its sibling queries. (Registration is the unsafe direction —
    /// see `registering_a_query_midstream_is_refused_and_correct_after_commit`.)
    #[test]
    fn destroy_query_midstream_leaves_the_txn_intact() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("midstream-destroy.db");
        let (cluster, events) = Cluster::open(&path, 1).unwrap();
        cluster
            .exec_ddl("CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER)")
            .unwrap();
        cluster.register_table("t").unwrap();
        let keep = QueryId(1);
        let doomed = QueryId(2);
        cluster.query(keep, rindle::table("t").build()).unwrap();
        cluster.query(doomed, rindle::table("t").build()).unwrap();

        let mut seed = cluster.write().unwrap();
        seed.exec("INSERT INTO t VALUES (?,?)", &[iv(0), iv(0)])
            .unwrap();
        seed.commit().unwrap();
        while events.try_recv().is_ok() {}

        let mut write = cluster.write().unwrap();
        for id in 1..=(PUSH_CHUNK_ROWS as i64 + 1) {
            write
                .exec("INSERT INTO t VALUES (?,?)", &[iv(id), iv(id)])
                .unwrap();
        }
        assert!(cluster.view_is_speculative());
        assert!(cluster.destroy_query(doomed), "teardown mid-stream applies");

        write.commit().expect("the streaming txn still commits");
        assert_eq!(
            cluster.read_snapshot(keep).unwrap().len(),
            PUSH_CHUNK_ROWS + 2,
            "the surviving query missed rows from the txn its sibling was torn down during"
        );
        let faults: Vec<_> = std::iter::from_fn(|| events.try_recv().ok())
            .filter(|e| matches!(e, crate::ClusterEvent::Faulted { .. }))
            .collect();
        assert!(
            faults.is_empty(),
            "solicited teardown faulted a query: {faults:?}"
        );
    }

    /// Rolling back a streamed txn must clear the fence too — otherwise one aborted bulk apply
    /// would wedge every later one-shot read for the process's lifetime.
    #[test]
    fn rollback_of_a_streamed_txn_clears_the_snapshot_fence() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("snapshot-fence-rollback.db");
        let (cluster, _events) = Cluster::open(&path, 1).unwrap();
        cluster
            .exec_ddl("CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER)")
            .unwrap();
        cluster.register_table("t").unwrap();

        let mut write = cluster.write().unwrap();
        for id in 0..(PUSH_CHUNK_ROWS as i64 + 1) {
            write
                .exec("INSERT INTO t VALUES (?,?)", &[iv(id), iv(id * 2)])
                .unwrap();
        }
        assert!(cluster.view_is_speculative());
        drop(write);
        assert!(
            !cluster.view_is_speculative(),
            "an aborted streamed txn left the one-shot fence armed"
        );
    }
}

#[cfg(test)]
mod ddl_marker_tests {
    //! [`Cluster::exec_ddl_with_marker`] is the atom the follower's crash-window DDL dedup rests on:
    //! the DDL and its idempotency marker must land together (so `marker present` ⇔ `DDL applied`) or
    //! roll back together (so a genuine fault leaves no marker and re-applies/faults next time).
    use super::*;

    const MARKER: &str = "_rindle_applied_ddl";

    fn open() -> (Cluster, tempfile::TempDir) {
        let dir = tempfile::tempdir().unwrap();
        let (cluster, _events) = Cluster::open(dir.path().join("m.db"), 1).unwrap();
        cluster
            .exec_ddl(&format!(
                "CREATE TABLE {MARKER} (key TEXT PRIMARY KEY, actions TEXT)"
            ))
            .unwrap();
        (cluster, dir)
    }

    /// The observe-mode authorizer (design 227 §3.6): the returned [`DdlApplyReport`] names
    /// exactly what the entry dropped/altered, PER STATEMENT and in execution order — through
    /// script-valued slots, comments between tokens, and quoting — because SQLite's parser
    /// reports the actions, not a text scan. Additive DDL observes nothing destructive. The
    /// report is also PERSISTED in the marker row (same transaction), so a replay can re-drive
    /// the decisions it feeds (fourth review pass).
    ///
    /// [`DdlApplyReport`]: crate::schema::DdlApplyReport
    #[test]
    fn marker_apply_reports_observed_destructive_actions() {
        let (cluster, _dir) = open();
        let created = cluster
            .exec_ddl_with_marker(
                MARKER,
                "m1",
                &["CREATE TABLE kv (k TEXT PRIMARY KEY NOT NULL, v TEXT); \
                   CREATE INDEX kv_v ON kv (v)"
                    .into()],
            )
            .unwrap();
        assert!(
            created.destructive_tables().next().is_none(),
            "got: {created:?}"
        );
        assert_eq!(
            created.steps.len(),
            2,
            "split at real boundaries: {created:?}"
        );
        // The explicit CREATE INDEX is captured by exact name (fifth review pass); the table's
        // implicit PK auto-index is not.
        assert_eq!(
            created
                .steps
                .iter()
                .flat_map(|s| s.created_indexes.iter())
                .collect::<Vec<_>>(),
            vec!["kv_v"],
            "got: {created:?}"
        );

        // A mid-script, comment-laden drop + same-shape recreate: the authorizer still sees it,
        // attributed to the drop's OWN statement (order feeds the registry bookkeeping).
        let dropped = cluster
            .exec_ddl_with_marker(
                MARKER,
                "m2",
                &["CREATE TABLE decoy (d TEXT PRIMARY KEY NOT NULL); \
                   DROP /* rebuild */ INDEX kv_v; \
                   DROP -- rebuild\n TABLE \"kv\"; \
                   CREATE TABLE kv (k TEXT PRIMARY KEY NOT NULL, v TEXT)"
                    .into()],
            )
            .unwrap();
        assert_eq!(
            dropped.dropped_tables().collect::<Vec<_>>(),
            vec!["kv"],
            "got: {dropped:?}"
        );
        // The qualified/comment-laden `DROP INDEX kv_v` is captured by its exact bare name.
        assert_eq!(
            dropped
                .steps
                .iter()
                .flat_map(|s| s.dropped_indexes.iter())
                .collect::<Vec<_>>(),
            vec!["kv_v"],
            "got: {dropped:?}"
        );
        let drop_step = dropped
            .steps
            .iter()
            .position(|step| !step.dropped_tables.is_empty())
            .expect("a step observed the drop");
        assert_eq!(
            drop_step, 2,
            "attributed to the DROP TABLE statement itself"
        );

        // The report round-trips from the marker row — the durable truth a replay consumes.
        let stored: Option<String> = cluster
            .read(|conn| {
                conn.query_row(
                    &format!("SELECT actions FROM {MARKER} WHERE key = 'm2'"),
                    [],
                    |row| row.get(0),
                )
                .optional()
            })
            .unwrap();
        let parsed: crate::schema::DdlApplyReport =
            serde_json::from_str(&stored.expect("report persisted with the marker")).unwrap();
        assert_eq!(
            parsed.dropped_tables().collect::<Vec<_>>(),
            vec!["kv"],
            "stored report round-trips"
        );

        let altered = cluster
            .exec_ddl_with_marker(MARKER, "m3", &["ALTER TABLE kv ADD COLUMN w TEXT".into()])
            .unwrap();
        assert!(
            altered
                .steps
                .iter()
                .any(|step| step.altered_tables.iter().any(|t| t == "kv")),
            "got: {altered:?}"
        );
    }

    fn has_marker(cluster: &Cluster, key: &str) -> bool {
        cluster
            .read(|c| {
                c.query_row(
                    &format!("SELECT 1 FROM {MARKER} WHERE key = ?1"),
                    [key],
                    |_| Ok(()),
                )
                .optional()
                .map(|r| r.is_some())
            })
            .unwrap()
    }

    fn table_exists(cluster: &Cluster, name: &str) -> bool {
        cluster
            .read(|c| {
                c.query_row(
                    "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1",
                    [name],
                    |_| Ok(()),
                )
                .optional()
                .map(|r| r.is_some())
            })
            .unwrap()
    }

    #[test]
    fn success_applies_ddl_and_stamps_the_marker() {
        let (cluster, _dir) = open();
        cluster
            .exec_ddl_with_marker(
                MARKER,
                "m1",
                &["CREATE TABLE kv (k TEXT PRIMARY KEY NOT NULL, v TEXT)".into()],
            )
            .unwrap();
        assert!(table_exists(&cluster, "kv"), "DDL applied");
        assert!(has_marker(&cluster, "m1"), "marker stamped");
        assert!(
            !has_marker(&cluster, "m2"),
            "only the applied key is journaled"
        );
    }

    #[test]
    fn a_failed_statement_rolls_back_ddl_and_marker_together() {
        let (cluster, _dir) = open();
        cluster
            .exec_ddl_with_marker(
                MARKER,
                "m1",
                &["CREATE TABLE kv (k TEXT PRIMARY KEY NOT NULL)".into()],
            )
            .unwrap();
        // Two statements: the first would succeed, the second faults (no such table). The whole unit
        // must roll back — neither the partial DDL nor the marker may survive.
        let err = cluster
            .exec_ddl_with_marker(
                MARKER,
                "m2",
                &[
                    "ALTER TABLE kv ADD COLUMN v TEXT".into(),
                    "ALTER TABLE ghost ADD COLUMN x TEXT".into(),
                ],
            )
            .unwrap_err();
        assert!(
            format!("{err}")
                .to_ascii_lowercase()
                .contains("no such table"),
            "{err}"
        );
        assert!(
            !has_marker(&cluster, "m2"),
            "a faulted unit leaves no marker"
        );
        assert!(
            !column_present(&cluster, "kv", "v"),
            "the first statement rolled back with the marker (atomic)"
        );
    }

    #[test]
    fn a_failed_step_effect_rolls_back_ddl_effect_and_marker_together() {
        let (cluster, _dir) = open();
        let error = cluster
            .exec_ddl_with_marker_and_step_effects(
                MARKER,
                "m1",
                &["CREATE TABLE kv (k TEXT PRIMARY KEY NOT NULL)".into()],
                |_conn, _step| Err(rusqlite::Error::InvalidQuery),
            )
            .unwrap_err();
        assert!(error.to_string().contains("atomic step effects"), "{error}");
        assert!(!table_exists(&cluster, "kv"), "DDL rolled back");
        assert!(!has_marker(&cluster, "m1"), "marker rolled back");
    }

    #[test]
    fn unsupported_follower_schema_rolls_back_before_stamping_marker() {
        let (cluster, _dir) = open();
        let error = cluster
            .exec_ddl_with_marker(
                MARKER,
                "blob-schema",
                &["CREATE TABLE bad (id INTEGER PRIMARY KEY, payload BLOB)".into()],
            )
            .unwrap_err();
        assert!(error.to_string().contains("unsupported"), "{error}");
        assert!(!table_exists(&cluster, "bad"));
        assert!(!has_marker(&cluster, "blob-schema"));
    }

    #[test]
    fn trusted_marker_path_allows_atomic_create_copy_swap_migrations() {
        let (cluster, _dir) = open();
        cluster
            .exec_ddl_with_marker(
                MARKER,
                "schema",
                &[
                    "CREATE TABLE kv (k TEXT PRIMARY KEY NOT NULL, v TEXT NOT NULL)".into(),
                    "INSERT INTO kv VALUES ('a', 'value')".into(),
                ],
            )
            .unwrap();
        cluster
            .exec_ddl_with_marker(
                MARKER,
                "reshape",
                &[
                    "CREATE TABLE kv_new (k TEXT PRIMARY KEY NOT NULL, v TEXT NOT NULL, note TEXT)"
                        .into(),
                    "INSERT INTO kv_new(k, v) SELECT k, v FROM kv".into(),
                    "DROP TABLE kv".into(),
                    "ALTER TABLE kv_new RENAME TO kv".into(),
                ],
            )
            .unwrap();
        assert!(has_marker(&cluster, "reshape"));
        assert!(column_present(&cluster, "kv", "note"));
        let value: String = cluster
            .read(|conn| conn.query_row("SELECT v FROM kv WHERE k = 'a'", [], |row| row.get(0)))
            .unwrap();
        assert_eq!(value, "value");
    }

    fn column_present(cluster: &Cluster, table: &str, col: &str) -> bool {
        cluster
            .read(|c| {
                c.query_row(
                    "SELECT 1 FROM pragma_table_info(?1) WHERE name = ?2",
                    [table, col],
                    |_| Ok(()),
                )
                .optional()
                .map(|r| r.is_some())
            })
            .unwrap()
    }
}