rindle_replica/

writer.rs

//! The single-writer transaction: capture, derive against a read-only overlay, commit, deliver.
//!
//! The crate OWNS the transaction boundaries (we wrap `begin`/`commit`, we do not use
//! SQLite's commit/rollback hooks): this is what lets us run the IVM derivation at
//! exactly the right moment — after the writes are captured, against a pre-commit
//! snapshot, before the durable COMMIT.

use std::rc::Rc;

use rindle::graph::NodeId;
use rindle::value::OwnedValue;
use rindle::{CaughtChange, RindleError};
use rusqlite::params_from_iter;

use crate::engine::Engine;
use crate::mutations::MutationSql;
use crate::query::{Callbacks, RegId};
use crate::sql::{from_sql, to_value};
use crate::{Inner, ReplicaError, TxId, Update};

/// 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).
/// Moved to the apply plane (the shared write-transaction state machine mints it —
/// design 309); re-exported here so `rindle_replica::CommitInfo` is unchanged.
pub use crate::apply::CommitInfo;

/// An open write transaction on the single writer connection. Run ordinary SQL with
/// [`exec`](WriteTxn::exec)/[`exec_batch`](WriteTxn::exec_batch); the preupdate hook
/// captures row deltas. [`commit`](WriteTxn::commit) derives against a read-only
/// pre-commit snapshot and batch overlay, commits SQL, then calls subscribers.
/// Dropping without committing rolls back and delivers no events.
pub struct WriteTxn {
    inner: Rc<Inner>,
    /// Set once committed or rolled back, so `Drop` doesn't double-finish.
    done: bool,
}

/// Open the writer transaction. Batch-overlay derivation never writes from the worker,
/// so an ordinary `BEGIN IMMEDIATE` is exact (the writer is already the sole committer
/// behind `in_write`).
pub(crate) fn begin(inner: Rc<Inner>) -> Result<WriteTxn, ReplicaError> {
    if inner.in_write.get() {
        return Err(ReplicaError::Open(
            "a write transaction is already open (single writer)".into(),
        ));
    }
    inner.cdc.reset(); // drop any stale captures
    inner
        .writer
        .execute_batch("BEGIN IMMEDIATE")
        .map_err(|e| ReplicaError::sqlite("writer BEGIN", e))?;
    inner.in_write.set(true);
    Ok(WriteTxn { inner, done: false })
}

impl WriteTxn {
    /// Run one statement with positional parameters inside the open transaction.
    /// Returns the number of rows changed. The preupdate hook captures the row deltas.
    pub fn exec(&mut self, sql: &str, params: &[OwnedValue]) -> Result<usize, ReplicaError> {
        self.inner
            .writer
            .execute(sql, params_from_iter(params.iter().map(to_value)))
            .map_err(|e| ReplicaError::sqlite("exec", e))
    }

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

    /// Commit: derive each query's incremental change against the pre-commit snapshot,
    /// persist the tx watermark + COMMIT the durable data, then deliver
    /// `Update::Changed` to affected subscribers. Returns the new tx id.
    ///
    /// Subscriber callbacks run synchronously after commit. A callback panic does not
    /// roll back the SQL transaction.
    ///
    /// A transaction that exceeds the batch-delta memory budget still commits — it is
    /// **shed** (design 306 D4): the engine rebuilds and every query re-hydrates,
    /// subscribers receiving a replacing [`Update::Hydrated`] instead of `Changed`
    /// events (see [`CommitInfo::shed`]). Any other derive error aborts the
    /// transaction as before.
    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 `_rindle_client_mutations` rows) feeds the engine; lmid flows to
    /// clients as ordinary query data (see `crate::mutations`).
    pub fn commit_with_info(mut self) -> Result<CommitInfo, ReplicaError> {
        self.done = true;
        let inner = self.inner.clone();

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

        let captured = inner.cdc.drain();
        let tx_id = inner.committed_tx.get() + 1;

        // Derive per-query deltas while the writer txn is still OPEN (snapshot = T-1).
        let mut shed = false;
        let per_query: Vec<(NodeId, Vec<CaughtChange>)> = if captured.is_empty() {
            Vec::new()
        } else {
            // Bind the result BEFORE matching: the shed arm re-borrows the engine
            // mutably, so the scrutinee's shared borrow must already be released.
            let derived = inner.engine.borrow().apply_batch(captured);
            match derived {
                Ok(pq) => pq,
                // The D4 delta-overflow SHED (design 306): the transaction outgrew the
                // batch delta, so it cannot be derived incrementally — but the DATA is
                // fine, and failing a write the storage layer happily holds would be
                // wrong. Treat it like load shedding: swap in a fresh engine now (while
                // we can still abort if even that fails), COMMIT the data, and
                // re-hydrate every query from the committed state below.
                Err(ReplicaError::Rindle(RindleError::DeltaOverflow { .. })) => {
                    if let Err(e) = rebuild_engine(&inner) {
                        self.abort(&inner);
                        return Err(e);
                    }
                    shed = true;
                    Vec::new()
                }
                Err(e) => {
                    self.abort(&inner);
                    return Err(e);
                }
            }
        };

        // Persist the watermark IN the writer txn, then COMMIT (data + tx_id atomic).
        if let Err(e) = persist_and_commit(&inner, tx_id) {
            self.abort(&inner);
            return Err(e);
        }
        inner.committed_tx.set(tx_id);
        inner.in_write.set(false);

        // Deliver AFTER the durable commit, so subscribers never see a tx that failed.
        // A shed re-hydrates instead: registration on the fresh engine reads the
        // just-committed state, and each subscriber gets a replacing `Hydrated`.
        if shed {
            rehydrate_all(&inner, tx_id);
        } else {
            deliver(&inner, tx_id, per_query);
        }
        Ok(CommitInfo {
            tx_id: TxId(tx_id),
            shed,
        })
    }

    /// Explicitly roll back. Leaves every view untouched; delivers nothing.
    pub fn rollback(mut self) {
        self.done = true;
        let inner = self.inner.clone();
        self.abort(&inner);
    }

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

impl Drop for WriteTxn {
    fn drop(&mut self) {
        if !self.done {
            let _ = self.inner.writer.execute_batch("ROLLBACK");
            self.inner.cdc.reset();
            self.inner.in_write.set(false);
        }
    }
}

/// The SQL `MutationTx` flavor (design §4.2): a server mutator runs against the open
/// single-writer transaction, reading through the same connection so it sees its own
/// uncommitted writes.
impl MutationSql for WriteTxn {
    fn exec(&mut self, sql: &str, params: &[OwnedValue]) -> Result<usize, ReplicaError> {
        WriteTxn::exec(self, sql, params)
    }

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

/// Upsert the committed-tx watermark into `__replica_meta` (inside the writer txn, so
/// it is durable atomically with the data), then COMMIT.
fn persist_and_commit(inner: &Inner, tx_id: u64) -> Result<(), ReplicaError> {
    inner
        .writer
        .execute(
            "INSERT INTO __replica_meta(id, tx_id) VALUES(0, ?1) \
             ON CONFLICT(id) DO UPDATE SET tx_id = excluded.tx_id",
            [tx_id as i64],
        )
        .map_err(|e| ReplicaError::sqlite("persist watermark", e))?;
    inner
        .writer
        .execute_batch("COMMIT")
        .map_err(|e| ReplicaError::sqlite("COMMIT", e))?;
    Ok(())
}

/// Deliver `Update::Changed` to each affected query's subscribers, without holding the
/// `subs` borrow across user callbacks (they may register/inspect).
fn deliver(inner: &Inner, tx_id: u64, per_query: Vec<(NodeId, Vec<CaughtChange>)>) {
    for (sink, changes) in per_query {
        // The engine reports by sink; subscriptions are keyed by the stable `RegId`.
        let Some(reg) = reg_of(inner, sink) else {
            continue;
        };
        let update = Update::Changed {
            tx_id: TxId(tx_id),
            changes,
        };
        fire(inner, reg, &update);
    }
}

/// The stable registration id currently wired to the engine sink `sink`, if any.
fn reg_of(inner: &Inner, sink: NodeId) -> Option<RegId> {
    inner
        .subs
        .borrow()
        .iter()
        .find(|(_, entry)| entry.sink == sink)
        .map(|(&reg, _)| reg)
}

/// Fire `update` at one query's subscribers. Takes the callbacks out of the entry so a
/// callback can't deadlock on the `subs` borrow, then puts them back (preserving any
/// added during delivery, which sit after).
fn fire(inner: &Inner, reg: RegId, update: &Update) {
    let mut cbs: Callbacks = {
        let mut subs = inner.subs.borrow_mut();
        match subs.get_mut(&reg) {
            Some(entry) => std::mem::take(&mut entry.callbacks),
            None => return,
        }
    };
    for cb in cbs.iter_mut() {
        cb(update);
    }
    let mut subs = inner.subs.borrow_mut();
    if let Some(entry) = subs.get_mut(&reg) {
        cbs.append(&mut entry.callbacks);
        entry.callbacks = cbs;
    }
}

/// The pre-commit half of the D4 shed: replace the (torn) engine with a fresh one,
/// sources re-registered from the cached schemas — the single-thread twin of the
/// worker threads' `fault_recover`, minus the `Faulted` events (queries re-hydrate in
/// place below instead). Runs while the writer txn is still open so a failure here can
/// still abort the transaction (the only fallible pieces are the operator-scratch
/// backend and source re-registration).
fn rebuild_engine(inner: &Inner) -> Result<(), ReplicaError> {
    let mut engine = inner.engine.borrow_mut();
    let mut fresh = Engine::new(
        engine.worker_conn(),
        engine.plan_queries(),
        engine.operator_storage(),
    )?;
    fresh.set_max_delta_bytes(engine.max_delta_bytes());
    for (table, ts) in inner.tables.borrow().iter() {
        fresh.register_table(table, ts.clone())?;
    }
    *engine = fresh; // drops the torn graph + all its pipelines
    Ok(())
}

/// The post-commit half of the D4 shed: re-register every subscribed query on the
/// fresh engine (hydration reads the just-committed state) and hand each subscriber a
/// replacing [`Update::Hydrated`]. A query whose re-registration fails (e.g. a §5.3
/// sum overflow over the new state — a fresh `SELECT` errors too) is dropped from the
/// registry: it can deliver no honest events, and its handle's `subscribe`/`destroy`
/// degrade to no-ops.
fn rehydrate_all(inner: &Inner, tx_id: u64) {
    let mut regs: Vec<RegId> = inner.subs.borrow().keys().copied().collect();
    regs.sort_unstable(); // registration order — deterministic delivery
    for reg in regs {
        let (query_id, ast) = {
            let subs = inner.subs.borrow();
            let entry = &subs[&reg];
            (entry.query_id, entry.ast.clone())
        };
        let registered = inner.engine.borrow_mut().register_query(query_id, &ast);
        match registered {
            Ok((sink, initial)) => {
                {
                    let mut subs = inner.subs.borrow_mut();
                    let entry = subs.get_mut(&reg).expect("entry collected above");
                    entry.sink = sink;
                    entry.hydrated = initial.clone();
                    entry.hydrated_tx = tx_id;
                }
                let update = Update::Hydrated {
                    tx_id: TxId(tx_id),
                    changes: initial,
                };
                fire(inner, reg, &update);
            }
            Err(_) => {
                inner.subs.borrow_mut().remove(&reg);
            }
        }
    }
}