rindle_replica/

mutations.rs

//! **Client mutations** — the upstream half of the optimistic-writes protocol
//! (`OPTIMISTIC-WRITES-DESIGN.md` §4, §8.1–§8.2).
//!
//! A client ships named-mutator invocations as [`MutationEnvelope`]s
//! (`{ client_id, mid, name, args }` — never code). The server looks `name` up in its
//! own [`MutatorRegistry`] and runs the mutator against an abstract SQL write handle
//! ([`MutationSql`], the SQL flavor of the design's `MutationTx`), **one transaction per
//! mutation**, in `mid` order per client.
//!
//! ## `lmid` is co-transactional authoritative state — and flows AS DATA (§8.2)
//!
//! Each applied mutation's transaction also upserts
//! `_rindle_client_mutations(client_id, last_mutation_id = mid)` — so the client's
//! high-water mutation id (`lmid`) is **data**, durable atomically with the mutation's
//! row effects, captured by the same CDC stream, and **hosted by the IVM engine like
//! any other table**. Each client subscribes to its own one-row system query
//! (`WHERE client_id = me`), so its lmid advance is delivered through the normal
//! derive → batch → `cv_min` release path — in the SAME coherent release as the
//! commit's effects. There is no side-channel: a release can never contain a commit's
//! data without that commit's lmid row, and never an lmid ahead of undelivered data.
//!
//! ## There are no rejections
//!
//! A mutator that fails (returns [`MutationReject`], panics, or is unknown to the
//! registry) is still **processed**: its row effects roll back, and `lmid` advances
//! past its `mid` in a follow-up lmid-only transaction — a normal commit whose lmid
//! row flows to the client like any data. The client drops the mutation from its
//! pending stack via the ordinary `mid ≤ lmid` rule; the optimistic prediction
//! rewinds out on that release (the snap-back). The protocol carries no
//! rejected/error signal — `MutationReject`'s reason is server-side logging only.
//!
//! Server mutators are authoritative and run once, so they need **no** determinism
//! contract — that constraint (§5) binds the *client* registry, which is re-invoked
//! on every rebase.

use std::collections::HashMap;
use std::panic::{catch_unwind, AssertUnwindSafe};

use rindle::value::OwnedValue;

use crate::{Db, ReplicaError};
use crate::{
    CLIENT_MUTATIONS_TABLE, ROOM_CLIENT_MUTATIONS_TABLE, ROOM_MUTATION_OUTCOMES_TABLE,
    ROOM_WATERMARK_TABLE, SCOPE_SESSIONS_TABLE,
};

/// The SQL flavor of the design's `MutationTx` (§4.2) — the abstract write handle a
/// server mutator runs against. Statements execute inside the mutation's own open
/// write transaction, and [`query`](MutationSql::query) reads **through the same
/// connection**, so a mutator sees its own uncommitted writes and the effects of every
/// lower-`mid` mutation — exactly what read-dependent mutators need (§4.1).
pub trait MutationSql {
    /// Run one statement with positional parameters; returns rows changed.
    fn exec(&mut self, sql: &str, params: &[OwnedValue]) -> Result<usize, ReplicaError>;
    /// Run a read returning all rows, each cell mapped from its raw SQLite storage
    /// class (`INTEGER → Int`, `REAL → Float`, `TEXT → Str`, `NULL → Null`; BLOB is
    /// an error). Raw classes, **not** the engine's number-widening coercion — the
    /// mutator is writing SQL, not feeding the pipeline.
    fn query(
        &mut self,
        sql: &str,
        params: &[OwnedValue],
    ) -> Result<Vec<Vec<OwnedValue>>, ReplicaError>;
}

/// The upstream wire envelope (§8.1): one named-mutator invocation. Mutations are
/// totally ordered per client by `mid`; the wire carries `name` + `args`, never code.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
pub struct MutationEnvelope {
    pub client_id: Box<str>,
    pub mid: u64,
    pub name: Box<str>,
    /// JSON-serializable arguments (the determinism contract, §5).
    pub args: serde_json::Value,
}

/// Why a mutator refused a mutation (permission/validation/SQL failure). Converts from
/// the common error shapes so `?` works inside a mutator body.
#[derive(Clone, Debug)]
pub struct MutationReject(pub String);

impl From<ReplicaError> for MutationReject {
    fn from(e: ReplicaError) -> MutationReject {
        MutationReject(e.to_string())
    }
}
impl From<String> for MutationReject {
    fn from(s: String) -> MutationReject {
        MutationReject(s)
    }
}
impl From<&str> for MutationReject {
    fn from(s: &str) -> MutationReject {
        MutationReject(s.to_string())
    }
}

/// A server-side mutator: authoritative, run once, against the open write transaction.
/// May read its own transaction's uncommitted writes via [`MutationSql::query`] (read-
/// dependent mutators see the current base, §4.1). An `Err` rejects the mutation.
pub type ServerMutator =
    Box<dyn Fn(&mut dyn MutationSql, &serde_json::Value) -> Result<(), MutationReject>>;

/// The server's registry of named mutators (§4.2). One of the **two** registries (the
/// client's optimistic twin lives in the client engine); the wire only carries names,
/// so sharing mutator code is a deployment choice, not a protocol mode.
#[derive(Default)]
pub struct MutatorRegistry {
    map: HashMap<Box<str>, ServerMutator>,
}

impl MutatorRegistry {
    pub fn new() -> MutatorRegistry {
        MutatorRegistry::default()
    }

    /// Register `f` under `name` (replacing any previous registration).
    pub fn register(
        &mut self,
        name: &str,
        f: impl Fn(&mut dyn MutationSql, &serde_json::Value) -> Result<(), MutationReject> + 'static,
    ) {
        self.map.insert(name.into(), Box::new(f));
    }

    pub(crate) fn get(&self, name: &str) -> Option<&ServerMutator> {
        self.map.get(name)
    }
}

/// What [`Db::apply_mutations`] did: every transaction it committed (in order — applied
/// mutations and lmid-only failure commits alike, each with its [`crate::CommitInfo`]).
/// Duplicate (already-processed) envelopes commit nothing.
#[derive(Debug, Default)]
pub struct MutationOutcome {
    pub commits: Vec<crate::CommitInfo>,
}

impl Db {
    /// One-time (idempotent) setup for the client-mutations protocol: create
    /// [`CLIENT_MUTATIONS_TABLE`] and register it like any base table (CDC capture +
    /// engine-hosted source), so `lmid` rides the change stream co-transactionally
    /// with the data AND is queryable — each client's one-row system query is an
    /// ordinary registered query over it (§8.2). Rejected while a write transaction
    /// is open (DDL would be invisible to the engine's worker until commit, like
    /// [`register_table`](Db::register_table)).
    pub fn enable_client_mutations(&self) -> Result<(), ReplicaError> {
        if self.inner.in_write.get() {
            return Err(ReplicaError::Open(
                "cannot enable client mutations while a write transaction is open".into(),
            ));
        }
        self.inner
            .writer
            .execute_batch(&format!(
                "CREATE TABLE IF NOT EXISTS {CLIENT_MUTATIONS_TABLE} \
                 (client_id TEXT NOT NULL PRIMARY KEY, last_mutation_id INTEGER NOT NULL);\n\
                 CREATE TABLE IF NOT EXISTS {ROOM_CLIENT_MUTATIONS_TABLE} \
                 (doc TEXT NOT NULL, client_id TEXT NOT NULL, last_mutation_id INTEGER NOT NULL, \
                  PRIMARY KEY (doc, client_id))"
            ))
            .map_err(|e| ReplicaError::sqlite("create client mutations table", e))?;
        self.register_table(CLIENT_MUTATIONS_TABLE)
    }

    /// One-time (idempotent) setup for the §4 realtime lifecycle
    /// (RINDLE-REALTIME-QUERY-ENABLEMENT-DESIGN.md): create the lifecycle tables and
    /// register them — plus [`ROOM_CLIENT_MUTATIONS_TABLE`] — like any base table (CDC
    /// capture + engine-hosted source), because each is load-bearing as SUBSCRIBABLE
    /// data, not just durable state:
    ///
    /// - [`SCOPE_SESSIONS_TABLE`] — the occupancy-row delta arriving through a solo
    ///   client's daemon subscription IS the upgrade doorbell (§4.1);
    /// - [`ROOM_WATERMARK_TABLE`] — the downgrade fence clears only when the client's
    ///   daemon subscription delivers the room's final `flush_seq` (§4.2);
    /// - [`ROOM_MUTATION_OUTCOMES_TABLE`] — outcome resolution with no room socket alive
    ///   (created + registered here; POPULATED by Slice I-ii's flush split);
    /// - [`ROOM_CLIENT_MUTATIONS_TABLE`] — "after downgrade, the doc-scoped ledger row is
    ///   ordinary footprint data" (§7.1, load-bearing for §7.5). Slice C created it
    ///   deliberately direct-SQL-only; the lifecycle is what needs it readable through
    ///   the daemon subscription plane, so its registration lands here, not in
    ///   [`enable_client_mutations`](Db::enable_client_mutations).
    ///
    /// Requires [`enable_client_mutations`](Db::enable_client_mutations) to have run
    /// first: that call owns the room ledger's DDL, and re-issuing a second copy here
    /// would be one edit away from two drifting schemas — enforced with a loud error
    /// (the `require_enabled` idiom) instead. Rejected while a write transaction is
    /// open, like the precedent.
    pub fn enable_realtime_lifecycle(&self) -> Result<(), ReplicaError> {
        if self.inner.in_write.get() {
            return Err(ReplicaError::Open(
                "cannot enable the realtime lifecycle while a write transaction is open".into(),
            ));
        }
        if !self.inner.cdc.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
            .writer
            .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(())
    }

    /// The high-water mutation id durably recorded for `client_id` (0 if none — a new
    /// client). What a server stamps on a (re)connecting client's handshake so it can
    /// drop already-confirmed pending mutations. Requires
    /// [`enable_client_mutations`](Db::enable_client_mutations).
    pub fn client_lmid(&self, client_id: &str) -> Result<u64, ReplicaError> {
        self.require_enabled()?;
        use rusqlite::OptionalExtension;
        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: each envelope's mutator runs in **its own**
    /// transaction (with the co-transactional `lmid` upsert), in the order given.
    /// Per envelope, against its client's stored `lmid`:
    ///
    /// - `mid ≤ lmid` — already processed: **skipped** (idempotent redelivery).
    /// - `mid == lmid + 1` — applied; on mutator failure (error / panic / unknown
    ///   name) the effects roll back and `lmid` still advances in an lmid-only commit
    ///   (processed-as-no-op — there is no rejection signal).
    /// - `mid > lmid + 1` — a gap (the client must send contiguously, so this should
    ///   be impossible): the push fails with [`ReplicaError::Mutation`] *at that
    ///   envelope* (prior envelopes stay applied — they are already durable).
    ///
    /// Every committed transaction fires the registered queries' subscriptions as
    /// usual and is reported (with its [`crate::CommitInfo`]) in the outcome, so the
    /// caller can drive a progress tracker / poke layer. Must be called with no write
    /// transaction open (it manages its own).
    pub fn apply_mutations(
        &self,
        registry: &MutatorRegistry,
        envelopes: &[MutationEnvelope],
    ) -> Result<MutationOutcome, ReplicaError> {
        self.require_enabled()?;
        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()?;
            // Unknown name is a rejection, not a protocol error: the registries may
            // drift across deploys, and the client recovers via rejected + snap-back.
            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 just the lmid row, which
                    // flows to the client's own lmid query like any data — the
                    // pending prediction snaps back on that release. The reason is
                    // server-side observability only; the protocol carries no signal.
                    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_enabled(&self) -> Result<(), ReplicaError> {
        if !self.inner.cdc.has_table(CLIENT_MUTATIONS_TABLE) {
            return Err(ReplicaError::Mutation(
                "client mutations not enabled — call enable_client_mutations() first".into(),
            ));
        }
        Ok(())
    }
}

/// The §4 lifecycle DDL (idempotent), shared verbatim by [`Db::enable_realtime_lifecycle`]
/// and [`Cluster::enable_realtime_lifecycle`](crate::Cluster::enable_realtime_lifecycle) so
/// both paths mint identical schemas (the [`upsert_lmid`] sharing discipline, applied to
/// DDL). [`ROOM_CLIENT_MUTATIONS_TABLE`] is deliberately absent: its DDL is owned by
/// `enable_client_mutations`, which the lifecycle enables require first.
pub(crate) fn realtime_lifecycle_ddl() -> String {
    format!(
        "CREATE TABLE IF NOT EXISTS {SCOPE_SESSIONS_TABLE} \
         (scope TEXT NOT NULL, client_id TEXT NOT NULL, expires_at INTEGER NOT NULL, \
          PRIMARY KEY (scope, client_id));\n\
         CREATE TABLE IF NOT EXISTS {ROOM_WATERMARK_TABLE} \
         (doc TEXT NOT NULL PRIMARY KEY, flush_seq INTEGER NOT NULL);\n\
         CREATE TABLE IF NOT EXISTS {ROOM_MUTATION_OUTCOMES_TABLE} \
         (doc TEXT NOT NULL, client_id TEXT NOT NULL, mid INTEGER NOT NULL, kind TEXT NOT NULL, \
          reason TEXT, name TEXT, args TEXT, PRIMARY KEY (doc, client_id, mid))"
    )
}

/// Upsert `last_mutation_id = mid` for `client_id` in the open transaction. Generic over
/// the [`MutationSql`] write handle so both the single-thread [`WriteTxn`](crate::WriteTxn)
/// and the parallel `ClusterWriteTxn` share it (WS3).
pub(crate) fn upsert_lmid(
    txn: &mut impl MutationSql,
    client_id: &str,
    mid: u64,
) -> Result<(), ReplicaError> {
    txn.exec(
        &format!(
            "INSERT INTO {CLIENT_MUTATIONS_TABLE}(client_id, last_mutation_id) \
             VALUES(?1, ?2) ON CONFLICT(client_id) DO UPDATE \
             SET last_mutation_id = excluded.last_mutation_id"
        ),
        &[OwnedValue::str(client_id), OwnedValue::Int(mid as i64)],
    )?;
    Ok(())
}