rindle_cdc_apply/

txn.rs

//! [`ApplyTxn`] — the single-writer transaction state machine of the apply plane,
//! lifted from `rindle-replica`'s `ClusterWriteTxn` (design 309 §3). Run SQL with
//! [`exec`](ApplyTxn::exec)/[`exec_batch`](ApplyTxn::exec_batch); the preupdate hook
//! captures the row deltas; [`commit_with_info`](ApplyTxn::commit_with_info) drives the
//! begin-barrier / push / finish / gate handshake against whatever [`CommitFanout`] the
//! transaction was opened with. Dropping without committing rolls back.
//!
//! Control flow is IDENTICAL whether the fan-out is the live replica's worker pool or
//! [`NoFanout`](super::NoFanout): the capture buffer still drains between statements at
//! [`PUSH_CHUNK_ROWS`] (the writer-side memory bound for a large restore transaction),
//! the capture-integrity guards (error poisoning, the uncaptured-row silent-divergence
//! check) still gate every commit — the barriers just have no one to hold headless.

use std::rc::Rc;

use rindle_value::value::OwnedValue;
use rusqlite::{params_from_iter, Connection};

use rindle_cdc::Captured;
use rindle_writeplane::{
    classify_statement, install_public_authorizer, run_statement, sql, statement_is_insert,
    writeplane, ReplicaError, SqlStatementRequest, StatementClass, StatementResult,
    StatementRunError, TxId, SQL_RESULT_BYTE_LIMIT,
};

use super::fanout::{CommitFanout, FanoutStream, StreamBegin, PUSH_CHUNK_ROWS};
use super::store::ApplyStore;

/// What one committed transaction did, beyond its data effects: the new global tx id
/// (the commit version `cv` the optimistic protocol stamps on outgoing batches).
/// Client `lmid` advances are NOT reported here — `_rindle_client_mutations` rows ride
/// the capture like any data and reach each client through its own system query (§8.2).
#[derive(Clone, Debug)]
pub struct CommitInfo {
    pub tx_id: TxId,
    /// True when this transaction overflowed the batch-delta row cap (design 306 D4)
    /// and was **shed**: the data committed, but no incremental `Changed` events were
    /// derived — the engine was rebuilt and every registered query re-hydrated, its
    /// subscribers receiving a fresh `Update::Hydrated` instead. The parallel
    /// `Cluster` reports its sheds per query via `ClusterEvent::Faulted`, never here;
    /// a headless applier has nothing to shed.
    pub shed: bool,
}

/// An open write transaction on the store's single writer connection. See the module
/// docs; the derivation host's `ClusterWriteTxn` is a thin public wrapper over this.
pub struct ApplyTxn {
    store: Rc<ApplyStore>,
    fanout: Rc<dyn CommitFanout>,
    /// Set once committed or rolled back, so `Drop` doesn't double-finish.
    done: bool,
    /// The streaming fan-out for this transaction, opened **eagerly** at the first captured
    /// change (`maybe_pump` — the begin-barrier *send*, `follow-ups/cluster-barrier-wakeup.md`
    /// §6.1, so a live pool pins its snapshots concurrently with the writer's remaining
    /// statements); the capture then forwards as bounded chunks between statements once it
    /// crosses `PUSH_CHUNK_ROWS`, instead of buffering the whole txn on the writer
    /// (`CLUSTER-INCREMENTAL-FANOUT-DESIGN.md` §7). `None` until the first captured change —
    /// an empty txn never opens it and commits via the one-shot begin-barrier in
    /// [`commit_with_info`](Self::commit_with_info). Dropping it (an abort/rollback) releases
    /// whatever the fan-out pinned; a worker pool that already received pushes tears down +
    /// re-hydrates (rehydrate-all, §5) — so rollback of a *streamed* txn is no longer free.
    stream: Option<Box<dyn FanoutStream>>,
    /// Set if the begin-barrier send failed mid-transaction (a worker was dead before the
    /// first push): stop streaming and let the changes accumulate, so `commit_with_info` falls
    /// back to the one-shot barrier (which repairs + fails cleanly + asks the caller to retry).
    /// `exec` therefore never gains a mid-statement "worker died" failure. Nothing was pushed
    /// before a first-open failure, so the fallback is a clean abort, not a rehydrate.
    stream_failed: bool,
}

impl ApplyTxn {
    /// Open the single-writer transaction (the store's `writer_begin_sql` flavor).
    /// Errors if one is already open.
    pub fn begin(
        store: Rc<ApplyStore>,
        fanout: Rc<dyn CommitFanout>,
    ) -> Result<ApplyTxn, ReplicaError> {
        if store.in_write.get() {
            return Err(ReplicaError::Open(
                "a write transaction is already open (single writer)".into(),
            ));
        }
        store.cdc.reset();
        store
            .writer
            .execute_batch(store.writer_begin_sql)
            .map_err(|e| ReplicaError::sqlite("writer BEGIN", e))?;
        store.in_write.set(true);
        Ok(ApplyTxn {
            store,
            fanout,
            done: false,
            stream: None,
            stream_failed: false,
        })
    }

    /// The store this transaction is open on.
    pub fn store(&self) -> &ApplyStore {
        &self.store
    }

    /// Run one statement with positional parameters inside the open transaction.
    pub fn exec(&mut self, sql: &str, params: &[OwnedValue]) -> Result<usize, ReplicaError> {
        let n = {
            let mut statement = self
                .store
                .writer
                .prepare_cached(sql)
                .map_err(|e| ReplicaError::sqlite("prepare exec", e))?;
            statement
                .execute(params_from_iter(params.iter().map(sql::to_value)))
                .map_err(|e| ReplicaError::sqlite("exec", e))?
        };
        self.maybe_pump();
        Ok(n)
    }

    /// 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 fn exec_bounded(
        &mut self,
        sql: &str,
        params: &[OwnedValue],
        guarded: bool,
    ) -> Result<usize, StatementRunError> {
        let conn = &self.store.writer;
        let result = writeplane::with_writer_statement_budget(conn, false, || {
            let _authorizer =
                guarded.then(|| install_public_authorizer(conn, StatementClass::Write));
            conn.execute(
                sql,
                params_from_iter(params.iter().map(rindle_writeplane::sql::to_value)),
            )
        })?;
        let changed = result?;
        self.maybe_pump();
        Ok(changed)
    }

    /// 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 fn public_statement(
        &mut self,
        request: &SqlStatementRequest,
        preserve_on_error: bool,
        result_byte_limit: usize,
    ) -> Result<StatementResult, StatementRunError> {
        let class = classify_statement(&request.sql);
        if class == StatementClass::Ddl {
            return Err(StatementRunError::Unsupported {
                code: "DDL_IN_TRANSACTION",
                message: "DDL is not supported inside an interactive transaction".into(),
            });
        }

        let checkpoint = self.store.cdc.checkpoint();
        let uncaptured_before = self.store.cdc.uncaptured_user_event_count();
        if preserve_on_error {
            if let Err(error) = self
                .store
                .writer
                .execute_batch("SAVEPOINT _rindle_sql_statement")
            {
                self.abort_in_place();
                return Err(StatementRunError::Sqlite(error));
            }
        }

        let report_rowid = class == StatementClass::Write && statement_is_insert(&request.sql);
        let attempted = (|| {
            let result =
                writeplane::with_writer_statement_budget(&self.store.writer, false, || {
                    run_statement(&self.store.writer, request, class, report_rowid)
                })??;
            if let Some(error) = self.store.cdc.take_error() {
                return Err(StatementRunError::ValueUnsupported(error.to_string()));
            }
            let uncaptured = self
                .store
                .cdc
                .uncaptured_user_event_count()
                .saturating_sub(uncaptured_before);
            if uncaptured != 0 {
                return Err(StatementRunError::Unsupported {
                    code: "STATEMENT_FAILED",
                    message: format!(
                        "silent-divergence guard: SQL modified {uncaptured} row(s) in an \
                         unregistered application table"
                    ),
                });
            }
            let encoded_len = serde_json::to_vec(&result.to_wire_json()?)
                .map_err(|error| StatementRunError::Malformed(error.to_string()))?
                .len();
            if encoded_len > result_byte_limit {
                return Err(StatementRunError::ResultLimit(format!(
                    "aggregate SQL result exceeds the {SQL_RESULT_BYTE_LIMIT}-byte limit"
                )));
            }
            Ok(result)
        })();

        match attempted {
            Ok(result) => {
                if preserve_on_error {
                    if let Err(error) = self
                        .store
                        .writer
                        .execute_batch("RELEASE _rindle_sql_statement")
                    {
                        self.abort_in_place();
                        return Err(StatementRunError::Sqlite(error));
                    }
                }
                self.maybe_pump();
                Ok(result)
            }
            Err(error) => {
                if preserve_on_error {
                    if self
                        .store
                        .writer
                        .execute_batch(
                            "ROLLBACK TO _rindle_sql_statement; RELEASE _rindle_sql_statement",
                        )
                        .is_err()
                    {
                        self.abort_in_place();
                    } else {
                        self.store.cdc.rewind(checkpoint);
                    }
                }
                Err(error)
            }
        }
    }

    /// Whether the underlying writer transaction is still open (neither finished by this
    /// handle nor auto-aborted by SQLite).
    pub fn is_open(&self) -> bool {
        !self.done && !self.store.writer.is_autocommit()
    }

    /// How many user-table row events the capture hook has recorded this transaction.
    pub fn captured_user_event_count(&self) -> usize {
        self.store.cdc.event_count()
    }

    /// The tx id this transaction will commit as.
    pub fn next_tx_id(&self) -> TxId {
        TxId(self.store.committed_tx.get() + 1)
    }

    /// The open transaction's connection for narrow host bookkeeping helpers. Anything
    /// written here still passes through the capture hook (the hook is on the
    /// connection), so application rows written raw are counted — but hosts should stay
    /// on the typed mutation methods and keep this for unregistered bookkeeping.
    pub fn connection(&self) -> &Connection {
        &self.store.writer
    }

    /// Run a batch of statements (no parameters) inside the open transaction.
    pub fn exec_batch(&mut self, sql: &str) -> Result<(), ReplicaError> {
        self.store
            .writer
            .execute_batch(sql)
            .map_err(|e| ReplicaError::sqlite("exec_batch", e))?;
        self.maybe_pump();
        Ok(())
    }

    /// Open the begin-barrier at the transaction's **first captured change** (a send, not a
    /// wait — `follow-ups/cluster-barrier-wakeup.md` §6.1: a live pool's workers wake, pin,
    /// and ack while the writer keeps executing statements; the acks are collected at the
    /// commit edge in [`finish_stream`](Self::finish_stream)), then forward the capture to
    /// the fan-out as bounded chunks once it crosses `PUSH_CHUNK_ROWS`. Called between
    /// statements so the writer's capture buffer stays ~`PUSH_CHUNK_ROWS` for a large
    /// transaction instead of growing to the whole txn
    /// (`CLUSTER-INCREMENTAL-FANOUT-DESIGN.md` §7); a small txn never crosses the push
    /// threshold and streams its whole capture at commit — through the barrier opened here,
    /// already acked by then in the common case. The hook fires once per *row*, so a
    /// transaction built from single-row statements (the replicator follower's apply path,
    /// `apply_muts`) pumps at exactly the chunk boundary — the §7 floor of "the largest single
    /// statement" is one row there, so the bound is truly `O(chunk)`.
    ///
    /// A begin-barrier send failure here is non-fatal: stop streaming (`stream_failed`) and
    /// leave the changes buffered, so the txn falls back to the commit-time barrier. Nothing
    /// was pushed before a first-open failure, so that fallback is a clean abort, not a
    /// rehydrate.
    fn maybe_pump(&mut self) {
        if self.stream_failed {
            return;
        }
        if self.stream.is_none() {
            if self.store.cdc.buffer_len() == 0 {
                return;
            }
            let tx_id = self.store.committed_tx.get() + 1;
            match self.fanout.tx_begin(tx_id) {
                StreamBegin::Ready(stream) => self.stream = Some(stream),
                StreamBegin::Failed(stream) => {
                    // Nothing pushed yet → a clean snapshot rollback of whatever pinned.
                    stream.finish().abort();
                    self.stream_failed = true;
                    return;
                }
            }
        }
        if self.store.cdc.buffer_len() < PUSH_CHUNK_ROWS {
            return;
        }
        let drained = self.store.cdc.drain();
        let stream = self.stream.as_mut().expect("opened above");
        for chunk in drained.chunks(PUSH_CHUNK_ROWS) {
            stream.push(chunk.to_vec().into());
        }
    }

    /// 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.
    pub fn query(
        &mut self,
        sql_text: &str,
        params: &[OwnedValue],
    ) -> Result<Vec<Vec<OwnedValue>>, ReplicaError> {
        let map_err = |e| ReplicaError::sqlite("mutator query", e);
        let mut stmt = self.store.writer.prepare(sql_text).map_err(map_err)?;
        let n = stmt.column_count();
        let rows = stmt
            .query_map(params_from_iter(params.iter().map(sql::to_value)), |r| {
                (0..n).map(|i| sql::from_sql(r.get_ref(i)?, i)).collect()
            })
            .map_err(map_err)?
            .collect::<rusqlite::Result<Vec<Vec<OwnedValue>>>>()
            .map_err(map_err)?;
        Ok(rows)
    }

    /// [`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_text: &str,
        params: &[OwnedValue],
    ) -> Result<(Vec<String>, Vec<Vec<OwnedValue>>), ReplicaError> {
        let map_err = |e| ReplicaError::sqlite("mutator query", e);
        let mut stmt = self.store.writer.prepare(sql_text).map_err(map_err)?;
        // The session-read contract (`rindle_writeplane::session::SessionBackend::query`): a
        // write smuggled through the read path would bypass the between-statement capture pump
        // and the exec-side accounting, so reject it up front rather than trust the caller's
        // compiler to only emit SELECTs.
        if !stmt.readonly() {
            return Err(ReplicaError::ReadRejected(
                "mutator session query must be read-only".into(),
            ));
        }
        let cols: Vec<String> = stmt.column_names().iter().map(|c| c.to_string()).collect();
        let n = cols.len();
        let rows = stmt
            .query_map(params_from_iter(params.iter().map(sql::to_value)), |r| {
                (0..n).map(|i| sql::from_sql(r.get_ref(i)?, i)).collect()
            })
            .map_err(map_err)?
            .collect::<rusqlite::Result<Vec<Vec<OwnedValue>>>>()
            .map_err(map_err)?;
        Ok((cols, rows))
    }

    /// Commit: run the fan-out's begin-barrier, persist the watermark + COMMIT the durable
    /// data, then release the gate. Returns the new tx id.
    pub fn commit(self) -> Result<TxId, ReplicaError> {
        self.commit_with_info().map(|info| info.tx_id)
    }

    /// [`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 — flows to the
    /// fan-out: on the live replica 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 barrier: it has no data to derive, but on the live replica 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(mut self) -> Result<CommitInfo, ReplicaError> {
        self.done = true;
        let store = self.store.clone();

        // A capture error (e.g. invalid UTF-8) poisons the whole transaction.
        if let Some(e) = store.cdc.take_error() {
            self.abort(&store);
            return Err(e.into());
        }
        let uncaptured = store.cdc.uncaptured_user_event_count();
        if uncaptured != 0 {
            self.abort(&store);
            return Err(ReplicaError::Capture(format!(
                "write transaction modified {uncaptured} row(s) in an unregistered application table"
            )));
        }

        // Whatever was captured since the last between-statement flush (`maybe_pump`): on a
        // streamed txn just the tail (often only the lmid row, §4); on a small txn the whole
        // transaction.
        let remainder = store.cdc.drain();

        // Already streaming this txn (its begin-barrier sent at the first captured change):
        // flush the tail and finish the open begin-barrier — no second barrier, no
        // empty-commit fast path.
        if let Some(stream) = self.stream.take() {
            return self.finish_stream(&store, stream, remainder);
        }

        let tx_id = store.committed_tx.get() + 1;

        // The fallback path for a txn that never opened its stream (an empty capture, or a
        // begin-barrier send failure mid-txn): send the begin-barrier now, then the
        // (lmid-inclusive) capture streams as bounded `Arc` chunks and `finish_stream`
        // collects the acks — for this path the fence is at its old commit-time position,
        // with only the push work to overlap. A txn ≤ `PUSH_CHUNK_ROWS` is one chunk
        // (wire-identical to the old batch). The empty case deliberately takes this path
        // too: zero chunks are pushed, but the committed finish makes a live pool publish
        // `Progressed(tx_id)`.
        match self.fanout.tx_begin(tx_id) {
            StreamBegin::Ready(stream) => self.finish_stream(&store, stream, remainder),
            StreamBegin::Failed(stream) => {
                // A worker was dead at the begin-barrier send — *before* any push, so nothing
                // derived. The writer has NOT committed: roll the reached observers back
                // cleanly (no fault), roll back the writer, repair the fan-out (reap +
                // respawn the offender), and fail the write so the caller retries. The
                // single-writer invariant holds — no observer can have missed a committed tx.
                // Structurally unreachable under `NoFanout` (its begin never fails).
                stream.finish().abort();
                self.abort(&store);
                self.fanout.recover_failed_begin();
                Err(ReplicaError::Open(
                    "a worker failed the commit barrier; it was respawned — retry the write".into(),
                ))
            }
        }
    }

    /// 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 progress boundary exists for a metadata-only cache write.
    pub fn commit_metadata_only(mut self) -> Result<(), ReplicaError> {
        self.done = true;
        let store = self.store.clone();
        if let Some(error) = store.cdc.take_error() {
            self.abort(&store);
            return Err(error.into());
        }
        let event_count = store.cdc.event_count();
        let uncaptured = store.cdc.uncaptured_user_event_count();
        if event_count != 0 || uncaptured != 0 || self.stream.is_some() {
            self.abort(&store);
            return Err(ReplicaError::Capture(format!(
                "metadata-only commit observed {event_count} user event(s), including \
                 {uncaptured} unregistered event(s)"
            )));
        }
        if let Err(error) = store.writer.execute_batch("COMMIT") {
            self.abort(&store);
            return Err(ReplicaError::sqlite("metadata-only COMMIT", error));
        }
        store.cdc.reset();
        store.in_write.set(false);
        Ok(())
    }

    /// Drive an open streaming transaction to completion: push the final `remainder` chunks
    /// (the tail since the last `maybe_pump`), collect the begin-barrier **acks** (the fence
    /// sent back at `tx_begin`, overlapped with the writer's own statements — usually already
    /// buffered, so no park), `persist_and_commit`, then release the gate — a live pool sends
    /// each worker its terminal marker with the verdict inline — or, on a COMMIT failure
    /// after a partial fan-out, `abort` (a pool whose workers advanced operator state the
    /// snapshot rollback can't undo tears down + re-hydrates; abort≙epoch-rehydrate, §4.1).
    /// Shared by the streamed-mid-flight path and the small-txn one-shot path; the cursor
    /// rides on the stream (`tx_id`), already minted at begin.
    fn finish_stream(
        &self,
        store: &ApplyStore,
        mut stream: Box<dyn FanoutStream>,
        remainder: Vec<Captured>,
    ) -> Result<CommitInfo, ReplicaError> {
        let tx_id = stream.tx_id();
        for chunk in remainder.chunks(PUSH_CHUNK_ROWS) {
            stream.push(chunk.to_vec().into());
        }
        // The fence: every observer still in the stream must have pinned its post-(N-1)
        // snapshot before COMMIT makes tx N visible (a late pin would derive against a base
        // that already contains N). A failure here is the begin-barrier failure of old,
        // discovered at the commit edge: nothing may commit — release the stream with an
        // abort (workers that were pushed to rehydrate; un-pushed ones roll back cleanly),
        // roll back the writer, repair the fan-out (reap + respawn the offender), and fail
        // the write so the caller retries. Structurally unreachable under `NoFanout` (no
        // acks to collect).
        if !stream.await_acks() {
            stream.finish().abort();
            self.abort(store);
            self.fanout.recover_failed_begin();
            return Err(ReplicaError::Open(
                "a worker failed the commit barrier; it was respawned — retry the write".into(),
            ));
        }
        let gate = stream.finish();
        match store.persist_and_commit(tx_id) {
            Ok(()) => {
                store.committed_tx.set(tx_id);
                store.in_write.set(false);
                // Release the gate: a live pool emits the `commit N` markers (`Progressed`).
                gate.commit();
                // A worker that overflowed its batch delta sheds itself (tear down +
                // `Faulted` per query, consumers re-hydrate) — the commit never reports
                // `shed` here.
                Ok(CommitInfo {
                    tx_id: TxId(tx_id),
                    shed: false,
                })
            }
            Err(e) => {
                gate.abort();
                self.abort(store);
                Err(e)
            }
        }
    }

    /// Explicitly roll back. Delivers nothing; a live pool's views are untouched.
    pub fn rollback(mut self) {
        self.done = true;
        let store = self.store.clone();
        self.abort(&store);
    }

    /// Roll back the writer txn and clear capture + the write guard.
    fn abort(&self, store: &ApplyStore) {
        let _ = store.writer.execute_batch("ROLLBACK");
        store.cdc.reset();
        store.in_write.set(false);
    }

    fn abort_in_place(&mut self) {
        if self.done {
            return;
        }
        self.done = true;
        let store = self.store.clone();
        self.abort(&store);
    }
}

impl Drop for ApplyTxn {
    fn drop(&mut self) {
        if !self.done {
            let _ = self.store.writer.execute_batch("ROLLBACK");
            self.store.cdc.reset();
            self.store.in_write.set(false);
        }
        // The `stream` field (if any) drops after this body: the fan-out releases whatever
        // it pinned — a live pool's workers that received pushes tear down + re-hydrate.
    }
}