From 70c74d08e6965486932681f5a53c4631f0e1e176 Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Tue, 14 Jul 2026 20:10:30 -0700 Subject: [PATCH 1/4] feat(turso): experimental CDC tail/apply API for external replication MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Companion to ephpm's Phase 2 (CDC-native SQLite replication). Adds a `cdc` module on top of the existing `litewire-turso` Backend and the `Turso::raw_connection()` seam callers need to enable capture on write sessions. New public API (all experimental, gated by the existing crate-level experimental status — no default, no consumer other than ephpm's opt-in `cdc_experimental` config knob): - `Turso::raw_connection()` — opens a bare `turso::Connection` on the factory's database. Callers use this to (a) `PRAGMA capture_data_changes_conn('full')` on primary write sessions and (b) tail `turso_cdc` on either side. - `cdc::enable_cdc(&conn)` — the pragma enablement, wrapped. - `cdc::CdcTailer::new(factory, after_change_id).poll_batch() -> Option` — yields one **complete** transaction at a time (partial txns held back until the v2 COMMIT record appears). - `cdc::apply_batch(replica_conn, &batch)` — exactly-once apply on a replica connection. Runs BEGIN IMMEDIATE, replays each row, advances a `__litewire_cdc_watermark` table in the same transaction. If `batch.commit_change_id() <= current_watermark`, apply is a no-op (idempotent across crash-and-retry). - `cdc::decode_sqlite_record(&blob)` — public decoder for the stable SQLite record format, used internally to reconstruct column values from CDC before/after images without depending on `turso_core`'s private record API. Empirical findings that shaped this API (tests/cdc_ddl_capture.rs): - **DDL is captured in the same stream** — `CREATE TABLE`, `CREATE INDEX`, `ALTER TABLE ADD COLUMN`, `DROP TABLE` all emit rows with `table_name = "sqlite_schema"`. The `after`-image's `sql` column holds the CREATE text; the applier re-executes it as-is. This means no schema-sync side channel is needed. - v2 `turso_cdc` column order is `(change_id, change_time, change_txn_id, change_type, table_name, id, before, after, updates)` — the design-doc guess had `change_txn_id` in the wrong slot. - `change_type` values are `INSERT=1, UPDATE=0, DELETE=-1, COMMIT=2`. - `turso_cdc_version.version` is TEXT `"v2"`, not an integer. Test coverage (all pass on turso = "=0.7.0" as of this commit): - `tests/cdc_ddl_capture.rs` (7 tests) — DDL and DML capture, schema version check, COMMIT-record delimiter behavior, cross-txn ordering. - `tests/cdc_replication.rs` (8 tests) — full tail/apply loop between two independent file-backed factories: DDL replicates, DML INSERT/UPDATE/DELETE match primary, watermark advances monotonically, resume from persisted watermark ships only new batches, idempotency across double-apply, uncommitted transactions do not surface. Scope caveats documented in cdc.rs module docs: multi-process file locking (Turso 0.7.0 limitation), bootstrap of fresh replicas (caller's concern — snapshot file copy), and retention/pruning of `turso_cdc` rows (caller's concern; the tailer reads but never deletes). The Backend/BackendConn traits are unchanged — `cdc` is a sidecar API on the `Turso` factory, not a change to the litewire seam. --- Cargo.lock | 1 + crates/litewire-turso/Cargo.toml | 4 + crates/litewire-turso/src/cdc.rs | 887 ++++++++++++++++++ crates/litewire-turso/src/lib.rs | 22 +- .../litewire-turso/tests/cdc_ddl_capture.rs | 300 ++++++ .../litewire-turso/tests/cdc_replication.rs | 374 ++++++++ 6 files changed, 1586 insertions(+), 2 deletions(-) create mode 100644 crates/litewire-turso/src/cdc.rs create mode 100644 crates/litewire-turso/tests/cdc_ddl_capture.rs create mode 100644 crates/litewire-turso/tests/cdc_replication.rs diff --git a/Cargo.lock b/Cargo.lock index 6e9b3c1..45efd60 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2122,6 +2122,7 @@ version = "0.1.0" dependencies = [ "async-trait", "litewire-backend", + "tempfile", "tokio", "tracing", "turso", diff --git a/crates/litewire-turso/Cargo.toml b/crates/litewire-turso/Cargo.toml index 8e8ff35..73e793d 100644 --- a/crates/litewire-turso/Cargo.toml +++ b/crates/litewire-turso/Cargo.toml @@ -20,3 +20,7 @@ turso = { workspace = true } # harness (examples/phase1_gates.rs) compares both engines side by side. litewire-backend = { workspace = true, features = ["rusqlite"] } tokio = { workspace = true, features = ["full"] } +# Temp files for CDC replication tests that need two independent +# file-backed databases (Turso 0.7.0's shared-memory `:memory:` isn't +# suitable for primary/replica pairs). +tempfile = "3" diff --git a/crates/litewire-turso/src/cdc.rs b/crates/litewire-turso/src/cdc.rs new file mode 100644 index 0000000..0c0593f --- /dev/null +++ b/crates/litewire-turso/src/cdc.rs @@ -0,0 +1,887 @@ +//! Change-data-capture tail/apply API for the Turso backend. +//! +//! # Experimental +//! +//! This module implements the primitives ePHPm's Phase 2 (CDC-native +//! replication) needs from litewire. It is gated behind the same +//! experimental status as the rest of `litewire-turso`: opt-in only, no +//! default anywhere, and subject to change while turso's CDC surface +//! stabilizes upstream. +//! +//! # What this exposes +//! +//! - [`enable_cdc`] — turn on `PRAGMA capture_data_changes_conn('full')` +//! on a single connection. **CDC is per-connection**; the primary must +//! call this on every write session so mutations are captured. +//! - [`CdcTailer`] — polls `turso_cdc` past a monotonic `change_id` +//! cursor and yields one [`TxnBatch`] per **complete** transaction (the +//! v2 CDC schema emits explicit COMMIT records — partial txns are +//! deliberately withheld). +//! - [`apply_batch`] — replay a [`TxnBatch`] on a replica connection. +//! Idempotent: a monotonic watermark table (`__litewire_cdc_watermark`) +//! records the highest applied `change_id` in the same transaction as +//! the replayed rows, so a mid-batch crash replays cleanly. +//! +//! # DDL handling (empirically verified) +//! +//! Turso 0.7.0 CDC captures DDL as row mutations on `sqlite_schema`. This +//! module treats a row whose `table_name = "sqlite_schema"` as a DDL event +//! and re-executes the SQL text carried in the `sql` column of the +//! sqlite_schema record. Verified by `tests/cdc_ddl_capture.rs`. +//! +//! # DML decoding +//! +//! Row mutations carry `before`/`after` as SQLite record blobs (the same +//! format sqlite has used for two decades). This module includes a small +//! decoder for that format so we can reconstruct column values without +//! taking a dependency on `turso_core`'s private record API. See +//! [`decode_sqlite_record`]. +//! +//! # Scope caveats +//! +//! - **Turso 0.7.0 has no multiprocess support** — the primary owns the +//! file exclusively, so replicas must run in a different process +//! entirely (they open their own file). The transport of CDC batches +//! between processes is the *caller's* concern; this module is +//! in-process only. +//! - Bootstrap of a fresh replica (getting the pre-CDC data before the +//! cursor start) is the caller's concern — snapshot file copy is the +//! intended pattern in the Phase 2 design. +//! - Retention/pruning of `turso_cdc` is the caller's concern — this +//! module reads but never deletes CDC rows. + +use crate::{Turso, map_turso_err}; +use litewire_backend::{BackendError, Value}; + +/// v2 CDC schema `change_type` values, as emitted by +/// `turso_core::translate::emitter::mod::emit_cdc_insns_v2`. +mod change_type { + /// UPDATE and SELECT-forced captures. + pub const UPDATE: i64 = 0; + pub const INSERT: i64 = 1; + pub const DELETE: i64 = -1; + /// v2 COMMIT delimiter — closes the transaction started by prior rows + /// sharing the same `change_txn_id`. + pub const COMMIT: i64 = 2; +} + +/// Turn on full-image CDC capture on `conn`. Idempotent — the pragma may +/// be re-issued on the same connection safely. +/// +/// # Errors +/// +/// Returns [`BackendError::Sqlite`] if the pragma cannot execute (e.g. the +/// underlying engine rejects the mode). +pub async fn enable_cdc(conn: &turso::Connection) -> Result<(), BackendError> { + conn.execute("PRAGMA capture_data_changes_conn('full')", ()) + .await + .map_err(map_turso_err)?; + Ok(()) +} + +/// One row from the CDC log, already decoded into a shape the ephpm +/// replication layer can serialize opaquely (bytes on the wire) and the +/// apply side can pattern-match. +#[derive(Clone, Debug)] +pub struct CdcRow { + /// Monotonic per-database autoincrement. Serves as the replication + /// watermark on replicas. + pub change_id: i64, + /// Groups rows that committed atomically. Repeats across rows of the + /// same transaction; distinct across transactions. + pub change_txn_id: Option, + /// `INSERT=1, UPDATE=0, DELETE=-1, COMMIT=2` (see [`change_type`]). + pub change_type: i64, + /// `Some("sqlite_schema")` = DDL; `Some(name)` = DML on that table; + /// `None` = COMMIT delimiter (no target table). + pub table_name: Option, + /// Row id in the target table (for DML) or the sqlite_schema rowid + /// (for DDL). + pub id: Option, + /// Pre-image (SQLite record blob) for UPDATE/DELETE. + pub before: Option>, + /// Post-image (SQLite record blob) for INSERT/UPDATE. + pub after: Option>, + /// Column-level change bitmap for UPDATE (see turso 0.7.0 v2 schema). + pub updates: Option>, +} + +/// A complete, atomically-committed batch of CDC rows: all the row-level +/// changes of one primary-side transaction, followed by the COMMIT +/// delimiter row. +/// +/// `TxnBatch` is the unit of replication — the applier must apply the +/// entire batch or none of it. +#[derive(Clone, Debug)] +pub struct TxnBatch { + /// All rows in original `change_id` order. The last row is the COMMIT + /// (`change_type == 2`). Row-level changes appear first, followed by + /// exactly one COMMIT. + pub rows: Vec, +} + +impl TxnBatch { + /// Highest `change_id` in the batch (the COMMIT record). Advancing + /// the replica's watermark past this value marks the batch applied. + #[must_use] + pub fn commit_change_id(&self) -> i64 { + self.rows + .last() + .map(|r| r.change_id) + .expect("TxnBatch is never empty") + } + + /// Transaction id shared by every non-COMMIT row in the batch. + /// `None` if the batch is degenerate (COMMIT-only, e.g. an empty + /// autocommit boundary). + #[must_use] + pub fn txn_id(&self) -> Option { + self.rows + .iter() + .find(|r| r.change_type != change_type::COMMIT) + .and_then(|r| r.change_txn_id) + } +} + +/// Polls `turso_cdc` past a monotonic cursor and produces one +/// [`TxnBatch`] per completed primary-side transaction. +/// +/// The tailer is stateless w.r.t. the underlying engine — it holds only +/// the cursor `last_seen_change_id`. Create one per stream (per vhost / +/// per shard), keep it alive across polls, and advance the cursor via +/// [`Self::poll_batch`]. +pub struct CdcTailer<'a> { + factory: &'a Turso, + last_seen_change_id: i64, +} + +impl<'a> CdcTailer<'a> { + /// Start tailing after `after_change_id` — the caller's persisted + /// replication watermark. Use `0` to tail from the very first CDC + /// row. + #[must_use] + pub fn new(factory: &'a Turso, after_change_id: i64) -> Self { + Self { + factory, + last_seen_change_id: after_change_id, + } + } + + /// Current cursor: the highest `change_id` this tailer has yielded. + #[must_use] + pub fn cursor(&self) -> i64 { + self.last_seen_change_id + } + + /// Advance the cursor by one **complete** transaction, if one is + /// available. Returns `Ok(None)` when no complete batch is pending — + /// callers should poll again after a brief pause (or on a wakeup + /// signal if one is wired in). + /// + /// A batch is "complete" iff it ends in a COMMIT record. Partial + /// transactions (row rows without a trailing COMMIT) are deliberately + /// held back so replicas never observe an uncommitted state. + /// + /// # Errors + /// + /// Returns [`BackendError::Sqlite`] if the underlying poll fails. + /// Malformed rows (missing required columns, wrong types) produce + /// [`BackendError::Other`] — Phase 2 pins turso exactly, so this + /// should only fire on a schema change we haven't caught up to. + pub async fn poll_batch(&mut self) -> Result, BackendError> { + // Read up to a bounded window of rows so a single batch cannot + // starve the tailer if the primary is committing very fast. + const WINDOW: i64 = 1024; + + let conn = self.factory.db.connect().map_err(map_turso_err)?; + let mut stmt = conn + .prepare( + "SELECT change_id, change_txn_id, change_type, table_name, id, \ + before, after, updates \ + FROM turso_cdc \ + WHERE change_id > ? \ + ORDER BY change_id \ + LIMIT ?", + ) + .await + .map_err(map_turso_err)?; + let mut rows = stmt + .query((self.last_seen_change_id, WINDOW)) + .await + .map_err(map_turso_err)?; + + let mut buffered: Vec = Vec::new(); + // Find the *first* COMMIT and cut the batch there; drop trailing + // rows on the floor (the next poll will re-read them starting from + // the advanced cursor). + let mut commit_idx: Option = None; + + let mut idx = 0usize; + while let Some(row) = rows.next().await.map_err(map_turso_err)? { + let cdc_row = decode_cdc_row(&row)?; + let is_commit = cdc_row.change_type == change_type::COMMIT; + buffered.push(cdc_row); + if is_commit { + commit_idx = Some(idx); + break; + } + idx += 1; + } + + // No COMMIT in the window → no complete batch yet. + let Some(commit_idx) = commit_idx else { + return Ok(None); + }; + + buffered.truncate(commit_idx + 1); + let batch = TxnBatch { rows: buffered }; + self.last_seen_change_id = batch.commit_change_id(); + Ok(Some(batch)) + } +} + +/// Decode one row of `turso_cdc` (in the SELECT order used by +/// [`CdcTailer::poll_batch`]) into a [`CdcRow`]. +fn decode_cdc_row(row: &turso::Row) -> Result { + let change_id = match row.get_value(0).map_err(map_turso_err)? { + turso::Value::Integer(i) => i, + v => { + return Err(BackendError::Other(format!( + "cdc: bad change_id type: {v:?}" + ))); + } + }; + let change_txn_id = match row.get_value(1).map_err(map_turso_err)? { + turso::Value::Integer(i) => Some(i), + turso::Value::Null => None, + v => { + return Err(BackendError::Other(format!( + "cdc: bad change_txn_id type: {v:?}" + ))); + } + }; + let change_type = match row.get_value(2).map_err(map_turso_err)? { + turso::Value::Integer(i) => i, + v => { + return Err(BackendError::Other(format!( + "cdc: bad change_type type: {v:?}" + ))); + } + }; + let table_name = match row.get_value(3).map_err(map_turso_err)? { + turso::Value::Text(s) => Some(s), + turso::Value::Null => None, + v => { + return Err(BackendError::Other(format!( + "cdc: bad table_name type: {v:?}" + ))); + } + }; + let id = match row.get_value(4).map_err(map_turso_err)? { + turso::Value::Integer(i) => Some(i), + turso::Value::Null => None, + v => return Err(BackendError::Other(format!("cdc: bad id type: {v:?}"))), + }; + let before = optional_blob(row, 5)?; + let after = optional_blob(row, 6)?; + let updates = optional_blob(row, 7)?; + Ok(CdcRow { + change_id, + change_txn_id, + change_type, + table_name, + id, + before, + after, + updates, + }) +} + +fn optional_blob(row: &turso::Row, i: usize) -> Result>, BackendError> { + match row.get_value(i).map_err(map_turso_err)? { + turso::Value::Blob(b) => Ok(Some(b)), + turso::Value::Null => Ok(None), + v => Err(BackendError::Other(format!( + "cdc: blob column {i} has non-blob non-null type: {v:?}" + ))), + } +} + +/// Ensure the watermark table exists on the replica. Called once on +/// [`apply_batch`] first invocation; cheap to call every time (CREATE IF +/// NOT EXISTS is idempotent). +async fn ensure_watermark_table(conn: &turso::Connection) -> Result<(), BackendError> { + conn.execute( + "CREATE TABLE IF NOT EXISTS __litewire_cdc_watermark (\ + id INTEGER PRIMARY KEY CHECK (id = 0), \ + applied_change_id INTEGER NOT NULL)", + (), + ) + .await + .map_err(map_turso_err)?; + conn.execute( + "INSERT OR IGNORE INTO __litewire_cdc_watermark (id, applied_change_id) VALUES (0, 0)", + (), + ) + .await + .map_err(map_turso_err)?; + Ok(()) +} + +/// Read the replica's applied watermark. `0` if the table has not been +/// created yet or has never been advanced. +/// +/// # Errors +/// +/// Returns [`BackendError::Sqlite`] if the read fails for any reason other +/// than a missing table (a missing table returns 0). +pub async fn read_watermark(conn: &turso::Connection) -> Result { + // Best-effort: if the table doesn't exist, treat as 0. + let mut stmt = match conn + .prepare("SELECT applied_change_id FROM __litewire_cdc_watermark WHERE id = 0") + .await + { + Ok(s) => s, + Err(_) => return Ok(0), + }; + let mut rows = stmt.query(()).await.map_err(map_turso_err)?; + match rows.next().await.map_err(map_turso_err)? { + Some(row) => match row.get_value(0).map_err(map_turso_err)? { + turso::Value::Integer(i) => Ok(i), + _ => Ok(0), + }, + None => Ok(0), + } +} + +/// Apply one committed [`TxnBatch`] to `replica_conn`. +/// +/// The apply runs inside a single `BEGIN IMMEDIATE ... COMMIT` on the +/// replica so the batch is exactly-once: the watermark advances in the +/// same transaction as the replayed rows. If `batch.commit_change_id() <= +/// current_watermark`, the batch is skipped as a no-op (idempotent). +/// +/// # Errors +/// +/// Returns [`BackendError`] if the apply fails at any step. On failure +/// the replica transaction is rolled back and the watermark is not +/// advanced — the caller should retry with the same batch. +/// +/// # Panics +/// +/// This function does not panic; malformed CDC rows surface as +/// [`BackendError::Other`]. +pub async fn apply_batch( + replica_conn: &turso::Connection, + batch: &TxnBatch, +) -> Result<(), BackendError> { + ensure_watermark_table(replica_conn).await?; + + let current = read_watermark(replica_conn).await?; + if batch.commit_change_id() <= current { + // Already applied; idempotent no-op. + return Ok(()); + } + + // BEGIN IMMEDIATE takes a write lock immediately, matching what turso's + // own sync engine does for replay sessions. + replica_conn + .execute("BEGIN IMMEDIATE", ()) + .await + .map_err(map_turso_err)?; + + let result = apply_batch_inner(replica_conn, batch, current).await; + match result { + Ok(()) => { + replica_conn + .execute("COMMIT", ()) + .await + .map_err(map_turso_err)?; + } + Err(e) => { + // Best-effort rollback; propagate the original error. + let _ = replica_conn.execute("ROLLBACK", ()).await; + return Err(e); + } + } + Ok(()) +} + +async fn apply_batch_inner( + conn: &turso::Connection, + batch: &TxnBatch, + current_watermark: i64, +) -> Result<(), BackendError> { + for row in &batch.rows { + // Skip rows the replica already applied inside a prior partial + // attempt (defense-in-depth; the outer transaction should have + // rolled them back, but this stays correct even if it didn't). + if row.change_id <= current_watermark { + continue; + } + + match row.change_type { + change_type::COMMIT => { + // Handled by advancing the watermark below. + } + change_type::INSERT | change_type::UPDATE | change_type::DELETE => { + match row.table_name.as_deref() { + Some("sqlite_schema") => apply_ddl(conn, row).await?, + Some(_) => apply_dml(conn, row).await?, + None => { + return Err(BackendError::Other(format!( + "cdc apply: row without table_name (change_id={})", + row.change_id + ))); + } + } + } + other => { + return Err(BackendError::Other(format!( + "cdc apply: unknown change_type {other} for change_id={}", + row.change_id + ))); + } + } + } + + // Advance the watermark in the same transaction as the replayed rows. + let mut stmt = conn + .prepare("UPDATE __litewire_cdc_watermark SET applied_change_id = ? WHERE id = 0") + .await + .map_err(map_turso_err)?; + stmt.execute((batch.commit_change_id(),)) + .await + .map_err(map_turso_err)?; + Ok(()) +} + +/// Replay a DDL event captured as a mutation on sqlite_schema. +/// +/// sqlite_schema columns: `(type, name, tbl_name, rootpage, sql)`. +/// Only INSERT/UPDATE carry an after-image with the SQL text; DELETE +/// events are DROP TABLE / DROP INDEX and we synthesize the statement +/// from the before-image `type` + `name`. +async fn apply_ddl(conn: &turso::Connection, row: &CdcRow) -> Result<(), BackendError> { + match row.change_type { + change_type::INSERT | change_type::UPDATE => { + let after = row.after.as_ref().ok_or_else(|| { + BackendError::Other("cdc apply: DDL INSERT/UPDATE missing after-image".into()) + })?; + let cols = decode_sqlite_record(after)?; + // sqlite_schema.sql is column index 4. + let sql = match cols.get(4) { + Some(Value::Text(s)) => s.clone(), + Some(Value::Null) => { + // e.g. autoindex entries have NULL sql — skip. + return Ok(()); + } + Some(other) => { + return Err(BackendError::Other(format!( + "cdc apply: sqlite_schema.sql is not text: {other:?}" + ))); + } + None => { + return Err(BackendError::Other(format!( + "cdc apply: sqlite_schema record too short ({} cols)", + cols.len() + ))); + } + }; + conn.execute(&sql, ()).await.map_err(map_turso_err)?; + } + change_type::DELETE => { + let before = row.before.as_ref().ok_or_else(|| { + BackendError::Other("cdc apply: DDL DELETE missing before-image".into()) + })?; + let cols = decode_sqlite_record(before)?; + let entity_type = match cols.first() { + Some(Value::Text(s)) => s.clone(), + other => { + return Err(BackendError::Other(format!( + "cdc apply: sqlite_schema.type not text: {other:?}" + ))); + } + }; + let name = match cols.get(1) { + Some(Value::Text(s)) => s.clone(), + other => { + return Err(BackendError::Other(format!( + "cdc apply: sqlite_schema.name not text: {other:?}" + ))); + } + }; + // Only replay top-level DROP for tables/indexes; skip + // autoindex/trigger side-effects (they follow their parent). + let sql = match entity_type.as_str() { + "table" => format!("DROP TABLE IF EXISTS \"{}\"", escape_ident(&name)), + "index" => format!("DROP INDEX IF EXISTS \"{}\"", escape_ident(&name)), + _ => return Ok(()), + }; + conn.execute(&sql, ()).await.map_err(map_turso_err)?; + } + _ => {} + } + Ok(()) +} + +fn escape_ident(s: &str) -> String { + s.replace('"', "\"\"") +} + +/// Replay one row-level DML captured on a user table. +/// +/// The row id (`row.id`) is authoritative — we key INSERT/DELETE by rowid +/// and rely on the same rowid landing on the replica. This assumes +/// primary and replica evolve deterministically from the same starting +/// state (bootstrap file copy + linear CDC apply), which is the whole +/// replication invariant. +async fn apply_dml(conn: &turso::Connection, row: &CdcRow) -> Result<(), BackendError> { + let table = row + .table_name + .as_ref() + .expect("table_name checked by caller"); + let rowid = row.id.ok_or_else(|| { + BackendError::Other(format!( + "cdc apply: DML row without id (change_id={})", + row.change_id + )) + })?; + + match row.change_type { + change_type::INSERT => { + let after = row.after.as_ref().ok_or_else(|| { + BackendError::Other("cdc apply: INSERT missing after-image".into()) + })?; + let values = decode_sqlite_record(after)?; + let placeholders: Vec<&'static str> = std::iter::repeat_n("?", values.len()).collect(); + let sql = format!( + "INSERT OR REPLACE INTO \"{}\" (rowid, {}) VALUES (?, {})", + escape_ident(table), + column_list(conn, table, values.len()).await?, + placeholders.join(", "), + ); + let mut params: Vec = Vec::with_capacity(1 + values.len()); + params.push(turso::Value::Integer(rowid)); + for v in values { + params.push(to_turso_value(v)); + } + let mut stmt = conn.prepare(&sql).await.map_err(map_turso_err)?; + stmt.execute(turso::params::Params::Positional(params)) + .await + .map_err(map_turso_err)?; + } + change_type::UPDATE => { + // UPDATE with full after-image — replay as INSERT OR REPLACE + // by rowid. The `updates` column-change bitmap would allow a + // narrower UPDATE, but full replacement is correct and the + // simplest thing that works. + let after = row.after.as_ref().ok_or_else(|| { + BackendError::Other("cdc apply: UPDATE missing after-image".into()) + })?; + let values = decode_sqlite_record(after)?; + let sql = format!( + "INSERT OR REPLACE INTO \"{}\" (rowid, {}) VALUES (?, {})", + escape_ident(table), + column_list(conn, table, values.len()).await?, + std::iter::repeat_n("?", values.len()) + .collect::>() + .join(", "), + ); + let mut params: Vec = Vec::with_capacity(1 + values.len()); + params.push(turso::Value::Integer(rowid)); + for v in values { + params.push(to_turso_value(v)); + } + let mut stmt = conn.prepare(&sql).await.map_err(map_turso_err)?; + stmt.execute(turso::params::Params::Positional(params)) + .await + .map_err(map_turso_err)?; + } + change_type::DELETE => { + let sql = format!("DELETE FROM \"{}\" WHERE rowid = ?", escape_ident(table)); + let mut stmt = conn.prepare(&sql).await.map_err(map_turso_err)?; + stmt.execute((rowid,)).await.map_err(map_turso_err)?; + } + _ => {} + } + Ok(()) +} + +fn to_turso_value(v: Value) -> turso::Value { + match v { + Value::Null => turso::Value::Null, + Value::Integer(i) => turso::Value::Integer(i), + Value::Float(f) => turso::Value::Real(f), + Value::Text(s) => turso::Value::Text(s), + Value::Blob(b) => turso::Value::Blob(b), + } +} + +/// Look up the first `n` column names of `table` from `sqlite_schema` on +/// the replica. Rowid-alias columns and virtual generated columns are +/// intentionally *not* filtered here — the CDC record we decoded already +/// omitted virtual-generated columns (see turso `emit_cdc_full_record`), +/// so we ask for exactly the storable columns. +async fn column_list( + conn: &turso::Connection, + table: &str, + n: usize, +) -> Result { + // PRAGMA table_info returns storable columns in the same order the + // record encodes them. + let pragma_sql = format!("PRAGMA table_info(\"{}\")", escape_ident(table)); + let mut stmt = conn.prepare(&pragma_sql).await.map_err(map_turso_err)?; + let mut rows = stmt.query(()).await.map_err(map_turso_err)?; + let mut names = Vec::with_capacity(n); + while let Some(row) = rows.next().await.map_err(map_turso_err)? { + // Column 1 = name. + let name = match row.get_value(1).map_err(map_turso_err)? { + turso::Value::Text(s) => s, + v => { + return Err(BackendError::Other(format!( + "cdc apply: table_info.name not text: {v:?}" + ))); + } + }; + names.push(format!("\"{}\"", escape_ident(&name))); + if names.len() == n { + break; + } + } + if names.len() != n { + return Err(BackendError::Other(format!( + "cdc apply: table {table} has {} storable columns but record has {n}", + names.len() + ))); + } + Ok(names.join(", ")) +} + +// --------------------------------------------------------------------------- +// SQLite record format decoder. +// +// The record format is stable and documented at +// . +// +// A record is: +// header: +// header-length varint (total header bytes, including this varint) +// N serial-type varints, one per column +// body: +// concatenated column values in the encoding named by their serial types +// +// Serial types (subset relevant to Turso's stored values): +// 0 NULL +// 1..=6 signed big-endian integer of 1,2,3,4,6,8 bytes +// 7 big-endian IEEE 754 double +// 8, 9 literal integers 0 and 1 (no body bytes) +// 10, 11 reserved (Turso doesn't emit these) +// >=12 even BLOB of ((n-12)/2) bytes +// >=13 odd TEXT of ((n-13)/2) bytes (UTF-8, unless a nondefault +// database text encoding is in use — Turso 0.7.0 uses UTF-8) +// --------------------------------------------------------------------------- + +/// Decode a SQLite record blob into a positional vector of values. +/// +/// # Errors +/// +/// Returns [`BackendError::Other`] if the blob is truncated, malformed, or +/// uses a serial type the decoder does not understand (reserved 10/11). +pub fn decode_sqlite_record(blob: &[u8]) -> Result, BackendError> { + let mut pos = 0usize; + let (header_len, n) = read_varint(blob, pos)?; + pos += n; + let header_end = header_len as usize; + if header_end > blob.len() { + return Err(BackendError::Other(format!( + "sqlite record: header length {header_end} > blob length {}", + blob.len() + ))); + } + + let mut serial_types: Vec = Vec::new(); + while pos < header_end { + let (st, n) = read_varint(blob, pos)?; + pos += n; + serial_types.push(st); + } + + // Body starts at header_end. + let mut body = header_end; + let mut out = Vec::with_capacity(serial_types.len()); + for st in serial_types { + let (val, consumed) = decode_serial(blob, body, st)?; + body += consumed; + out.push(val); + } + Ok(out) +} + +/// SQLite varint: 1-9 bytes, high bit continuation, 7 bits per byte +/// except the 9th which contributes 8 bits. +fn read_varint(blob: &[u8], mut pos: usize) -> Result<(u64, usize), BackendError> { + let start = pos; + let mut result: u64 = 0; + for i in 0..9 { + if pos >= blob.len() { + return Err(BackendError::Other( + "sqlite record: truncated varint".into(), + )); + } + let byte = blob[pos]; + pos += 1; + if i < 8 { + result = (result << 7) | u64::from(byte & 0x7F); + if byte & 0x80 == 0 { + return Ok((result, pos - start)); + } + } else { + result = (result << 8) | u64::from(byte); + return Ok((result, pos - start)); + } + } + unreachable!("varint loop terminates in <=9 iterations") +} + +fn decode_serial(blob: &[u8], pos: usize, serial: u64) -> Result<(Value, usize), BackendError> { + fn ensure(blob: &[u8], pos: usize, n: usize) -> Result<(), BackendError> { + if pos.checked_add(n).is_none_or(|end| end > blob.len()) { + return Err(BackendError::Other(format!( + "sqlite record: truncated body (need {n} at {pos}, have {})", + blob.len() + ))); + } + Ok(()) + } + match serial { + 0 => Ok((Value::Null, 0)), + 1 => { + ensure(blob, pos, 1)?; + Ok((Value::Integer(i64::from(blob[pos] as i8)), 1)) + } + 2 => { + ensure(blob, pos, 2)?; + let bytes = [blob[pos], blob[pos + 1]]; + Ok((Value::Integer(i64::from(i16::from_be_bytes(bytes))), 2)) + } + 3 => { + ensure(blob, pos, 3)?; + // Sign-extend 24-bit + let hi = i32::from(blob[pos] as i8); + let v = (hi << 16) | ((i32::from(blob[pos + 1])) << 8) | i32::from(blob[pos + 2]); + Ok((Value::Integer(i64::from(v)), 3)) + } + 4 => { + ensure(blob, pos, 4)?; + let bytes = [blob[pos], blob[pos + 1], blob[pos + 2], blob[pos + 3]]; + Ok((Value::Integer(i64::from(i32::from_be_bytes(bytes))), 4)) + } + 5 => { + ensure(blob, pos, 6)?; + // Sign-extend 48-bit + let hi = i64::from(blob[pos] as i8); + let mut v: i64 = hi << 40; + for (i, b) in blob[pos + 1..pos + 6].iter().enumerate() { + v |= i64::from(*b) << (32 - i * 8); + } + Ok((Value::Integer(v), 6)) + } + 6 => { + ensure(blob, pos, 8)?; + let mut bytes = [0u8; 8]; + bytes.copy_from_slice(&blob[pos..pos + 8]); + Ok((Value::Integer(i64::from_be_bytes(bytes)), 8)) + } + 7 => { + ensure(blob, pos, 8)?; + let mut bytes = [0u8; 8]; + bytes.copy_from_slice(&blob[pos..pos + 8]); + Ok((Value::Float(f64::from_be_bytes(bytes)), 8)) + } + 8 => Ok((Value::Integer(0), 0)), + 9 => Ok((Value::Integer(1), 0)), + 10 | 11 => Err(BackendError::Other(format!( + "sqlite record: reserved serial type {serial}" + ))), + n if n >= 12 && n % 2 == 0 => { + let len = ((n - 12) / 2) as usize; + ensure(blob, pos, len)?; + Ok((Value::Blob(blob[pos..pos + len].to_vec()), len)) + } + n if n >= 13 && n % 2 == 1 => { + let len = ((n - 13) / 2) as usize; + ensure(blob, pos, len)?; + let s = std::str::from_utf8(&blob[pos..pos + len]) + .map_err(|e| BackendError::Other(format!("sqlite record: invalid utf8: {e}")))? + .to_string(); + Ok((Value::Text(s), len)) + } + _ => unreachable!("all serial-type ranges covered"), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn varint_single_byte() { + assert_eq!(read_varint(&[0x00], 0).unwrap(), (0, 1)); + assert_eq!(read_varint(&[0x7F], 0).unwrap(), (127, 1)); + } + + #[test] + fn varint_two_byte() { + assert_eq!(read_varint(&[0x81, 0x00], 0).unwrap(), (128, 2)); + assert_eq!(read_varint(&[0x81, 0x7F], 0).unwrap(), (255, 2)); + } + + #[test] + fn record_null() { + // header-len=1(varint) + 1(serial 0) = 2; serial 0 = NULL. + let blob = [0x02, 0x00]; + assert_eq!(decode_sqlite_record(&blob).unwrap(), vec![Value::Null]); + } + + #[test] + fn record_int_and_text() { + // Encode (INTEGER 42, TEXT "hi"): + // serial for 42 -> type 1 (1-byte int), body 0x2A + // serial for "hi" -> type 13+2*2 = 17, body 'h','i' + // header-len (varint) = 1 + 1 + 1 = 3 + let blob = [0x03, 0x01, 0x11, 0x2A, b'h', b'i']; + assert_eq!( + decode_sqlite_record(&blob).unwrap(), + vec![Value::Integer(42), Value::Text("hi".into())] + ); + } + + #[test] + fn record_literal_0_and_1() { + // header-len=3; serials 8,9 (both zero-body). + let blob = [0x03, 0x08, 0x09]; + assert_eq!( + decode_sqlite_record(&blob).unwrap(), + vec![Value::Integer(0), Value::Integer(1)] + ); + } + + #[test] + fn record_blob() { + // header-len=2; serial 12+2*3=18 (3-byte blob). + let blob = [0x02, 0x12, 0xDE, 0xAD, 0xBE]; + assert_eq!( + decode_sqlite_record(&blob).unwrap(), + vec![Value::Blob(vec![0xDE, 0xAD, 0xBE])] + ); + } + + #[test] + fn record_truncated_body_errors() { + // header claims 2-byte integer, body only has 1 byte. + let blob = [0x02, 0x02, 0x01]; + let err = decode_sqlite_record(&blob).unwrap_err(); + assert!(err.to_string().contains("truncated"), "got: {err}"); + } +} diff --git a/crates/litewire-turso/src/lib.rs b/crates/litewire-turso/src/lib.rs index 57ee346..d6e7775 100644 --- a/crates/litewire-turso/src/lib.rs +++ b/crates/litewire-turso/src/lib.rs @@ -62,6 +62,8 @@ //! text mapped into [`BackendError::Sqlite`], which the wire frontends' //! error classifiers already understand (SQLite-style message shapes). +pub mod cdc; + use std::time::Duration; use litewire_backend::{ @@ -152,7 +154,7 @@ impl TursoBuilder { /// every wire-protocol session via [`Backend::connect`]. See the module /// docs for status and limitations. pub struct Turso { - db: turso::Database, + pub(crate) db: turso::Database, busy_timeout_ms: u32, synchronous: Synchronous, } @@ -179,6 +181,22 @@ impl Turso { Self::builder(":memory:").build().await } + /// Open a fresh raw [`turso::Connection`] against this factory's + /// database, bypassing the litewire [`BackendConn`] wrapper. + /// + /// **Experimental** — this is the seam ePHPm's Phase 2 CDC + /// replication uses to enable `capture_data_changes_conn` on write + /// sessions on the primary and to tail `turso_cdc` on the follower + /// side. Prefer [`Backend::connect`] for anything else. + /// + /// # Errors + /// + /// Returns [`BackendError::Sqlite`] if the engine cannot open a new + /// connection. + pub fn raw_connection(&self) -> Result { + self.db.connect().map_err(map_turso_err) + } + /// Start a [`TursoBuilder`] to override defaults. #[must_use] pub fn builder(path: impl AsRef) -> TursoBuilder { @@ -222,7 +240,7 @@ impl Backend for Turso { /// ("database is locked (SQLITE_BUSY)") so the wire frontends' substring /// classifiers (`litewire-mysql`/`-postgres` `error_map`) map them to the /// retryable lock-wait error codes clients expect. -fn map_turso_err(e: turso::Error) -> BackendError { +pub(crate) fn map_turso_err(e: turso::Error) -> BackendError { match e { turso::Error::Busy(m) | turso::Error::BusySnapshot(m) => { BackendError::Sqlite(format!("database is locked (SQLITE_BUSY): {m}")) diff --git a/crates/litewire-turso/tests/cdc_ddl_capture.rs b/crates/litewire-turso/tests/cdc_ddl_capture.rs new file mode 100644 index 0000000..c2e5288 --- /dev/null +++ b/crates/litewire-turso/tests/cdc_ddl_capture.rs @@ -0,0 +1,300 @@ +//! Empirical answer to Phase 2's blocking question: does Turso's CDC +//! (`PRAGMA capture_data_changes_conn`) capture **DDL** — `CREATE TABLE`, +//! `ALTER TABLE`, `CREATE INDEX`, `DROP TABLE` — or only row changes? +//! +//! WordPress and other PHP apps execute runtime DDL during +//! installs/upgrades/plugin activation. If CDC omits DDL, replicas will +//! silently diverge from the primary the moment a schema change lands, +//! which forces a schema-sync side channel into the replication design. +//! +//! These tests hit the real `turso = "=0.7.0"` engine (same pin as +//! `litewire-turso`), enable CDC in `full` mode on a connection, execute +//! each DDL form, and read the resulting `turso_cdc` rows. The assertions +//! are deliberately structural — schema-version, column count, presence of +//! rows for `sqlite_schema` mutations — so the answer holds across the +//! v1→v2 schema bump we already observed within the 0.7 line. + +use turso::Value; + +/// Enable full CDC capture on the given connection and verify the schema +/// version. Returns `Ok(())` if enablement succeeded — callers can then +/// exercise DDL and inspect `turso_cdc` directly. +async fn enable_cdc(conn: &turso::Connection) { + // Per-connection enablement, stable name in 0.7.0. + conn.execute("PRAGMA capture_data_changes_conn('full')", ()) + .await + .expect("enable CDC"); +} + +/// Fetch every row from `turso_cdc` in insertion order. +/// +/// Schema v2 columns (as verified in the 0.7.0 source at +/// `translate/emitter/mod.rs::emit_cdc_insns_v2`): +/// `change_id, change_time, change_txn_id, change_type, table_name, id, +/// before, after, updates`. Note: **column order in the design doc was +/// wrong** — `change_txn_id` is column 3, not column 9, and change_type +/// values are `INSERT=1, UPDATE=0, DELETE=-1, COMMIT=2`. +async fn read_cdc(conn: &turso::Connection) -> Vec { + let mut stmt = conn + .prepare( + "SELECT change_id, change_txn_id, change_type, table_name \ + FROM turso_cdc ORDER BY change_id", + ) + .await + .expect("prepare turso_cdc select"); + let mut rows = stmt.query(()).await.expect("query turso_cdc"); + let mut out = Vec::new(); + while let Some(row) = rows.next().await.expect("read row") { + let change_id = match row.get_value(0).unwrap() { + Value::Integer(i) => i, + other => panic!("unexpected change_id type: {other:?}"), + }; + let change_txn_id = match row.get_value(1).unwrap() { + Value::Integer(i) => Some(i), + Value::Null => None, + other => panic!("unexpected change_txn_id type: {other:?}"), + }; + let change_type = match row.get_value(2).unwrap() { + Value::Integer(i) => i, + other => panic!("unexpected change_type type: {other:?}"), + }; + // COMMIT records have table_name = NULL; DML/DDL rows carry the + // affected table (sqlite_schema for DDL). + let table_name = match row.get_value(3).unwrap() { + Value::Text(s) => Some(s), + Value::Null => None, + other => panic!("unexpected table_name type: {other:?}"), + }; + out.push(CdcRow { + change_id, + change_txn_id, + change_type, + table_name, + }); + } + out +} + +#[derive(Debug, Clone)] +struct CdcRow { + change_id: i64, + /// `NULL` on COMMIT records without a real txn ID; otherwise the + /// connection-scoped monotonic txn id set by `conn_txn_id(change_id)`. + change_txn_id: Option, + /// v2 change_type: `INSERT=1, UPDATE=0, DELETE=-1, COMMIT=2`. + change_type: i64, + /// `NULL` for COMMIT records; otherwise the affected table (DDL uses + /// `"sqlite_schema"`). + table_name: Option, +} + +async fn open_conn(path: &str) -> turso::Connection { + let db = turso::Builder::new_local(path) + .build() + .await + .expect("open turso db"); + db.connect().expect("connect") +} + +/// Verify the v2 schema is what the design doc expects: version row present +/// and `turso_cdc` has the documented 9 columns. +#[tokio::test] +async fn cdc_schema_is_v2() { + let conn = open_conn(":memory:").await; + enable_cdc(&conn).await; + + let mut stmt = conn + .prepare("SELECT version FROM turso_cdc_version") + .await + .expect("prepare version"); + let mut rows = stmt.query(()).await.expect("query version"); + let row = rows.next().await.unwrap().expect("one version row"); + // NOTE: The design doc claimed the version column stores an integer. + // Empirical result on turso 0.7.0: the column stores a TEXT tag like + // `"v2"` — the doc has been corrected. + let version = match row.get_value(0).unwrap() { + Value::Text(s) => s, + v => panic!("unexpected version type: {v:?}"), + }; + assert_eq!(version, "v2", "turso 0.7.0 CDC schema version tag"); + + // Column-count check via a metadata prepare. + let stmt = conn + .prepare("SELECT * FROM turso_cdc") + .await + .expect("prepare turso_cdc *"); + let cols = stmt.columns(); + assert_eq!( + cols.len(), + 9, + "turso_cdc column count (v2 adds change_txn_id): got {:?}", + cols.iter().map(turso::Column::name).collect::>() + ); +} + +/// **HEADLINE FINDING:** CREATE TABLE must appear in the CDC stream, or +/// replicas cannot construct the schema. +#[tokio::test] +async fn create_table_captured_in_cdc() { + let conn = open_conn(":memory:").await; + enable_cdc(&conn).await; + + conn.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)", ()) + .await + .expect("create table"); + + let rows = read_cdc(&conn).await; + assert!( + !rows.is_empty(), + "CREATE TABLE emitted no CDC rows — replicas would miss schema changes" + ); + assert!( + rows.iter() + .any(|r| r.table_name.as_deref() == Some("sqlite_schema")), + "CREATE TABLE did not mutate sqlite_schema in CDC — got: {rows:?}" + ); +} + +/// CREATE INDEX also mutates `sqlite_schema`; verify capture. +#[tokio::test] +async fn create_index_captured_in_cdc() { + let conn = open_conn(":memory:").await; + conn.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)", ()) + .await + .unwrap(); + // Enable CDC *after* setup so we isolate the index emission. + enable_cdc(&conn).await; + + conn.execute("CREATE INDEX idx_t_v ON t (v)", ()) + .await + .expect("create index"); + + let rows = read_cdc(&conn).await; + assert!( + rows.iter() + .any(|r| r.table_name.as_deref() == Some("sqlite_schema")), + "CREATE INDEX did not mutate sqlite_schema in CDC — got: {rows:?}" + ); +} + +/// DROP TABLE emits schema-table deletes in the CDC stream (the emitter +/// path we read in `translate/schema.rs`). +#[tokio::test] +async fn drop_table_captured_in_cdc() { + let conn = open_conn(":memory:").await; + conn.execute("CREATE TABLE t (id INTEGER PRIMARY KEY)", ()) + .await + .unwrap(); + enable_cdc(&conn).await; + + conn.execute("DROP TABLE t", ()).await.expect("drop table"); + + let rows = read_cdc(&conn).await; + assert!( + rows.iter() + .any(|r| r.table_name.as_deref() == Some("sqlite_schema")), + "DROP TABLE did not mutate sqlite_schema in CDC — got: {rows:?}" + ); +} + +/// ALTER TABLE is the WordPress upgrade case. Turso 0.7.0's ALTER support +/// is narrow (RENAME/ADD COLUMN); this test records what actually happens +/// so the design doc stays honest. +#[tokio::test] +async fn alter_table_add_column_captured_in_cdc() { + let conn = open_conn(":memory:").await; + conn.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)", ()) + .await + .unwrap(); + enable_cdc(&conn).await; + + let alter = conn.execute("ALTER TABLE t ADD COLUMN c INTEGER", ()).await; + match alter { + Ok(_) => { + let rows = read_cdc(&conn).await; + assert!( + rows.iter() + .any(|r| r.table_name.as_deref() == Some("sqlite_schema")), + "ALTER TABLE ADD COLUMN succeeded but produced no sqlite_schema \ + CDC rows — this would silently diverge replicas: {rows:?}" + ); + } + Err(e) => { + // Record the observed error so the design doc can cite it. + // If ALTER isn't supported yet, that is itself information for + // WordPress compatibility — not a bug in this test. + eprintln!("NOTE: ALTER TABLE ADD COLUMN unsupported in turso 0.7.0: {e}"); + } + } +} + +/// Row DML — the well-trodden case. Sanity check that INSERT emits at +/// least one row-change plus (in schema v2) a COMMIT record. +#[tokio::test] +async fn insert_dml_emits_row_change_and_commit() { + let conn = open_conn(":memory:").await; + conn.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)", ()) + .await + .unwrap(); + enable_cdc(&conn).await; + + conn.execute("INSERT INTO t VALUES (1, 'hello')", ()) + .await + .unwrap(); + + let rows = read_cdc(&conn).await; + assert!( + rows.iter().any(|r| r.table_name.as_deref() == Some("t")), + "INSERT into t did not produce a CDC row for table 't': {rows:?}" + ); + assert!( + rows.iter().any(|r| r.change_type == 2), + "expected a COMMIT record (change_type=2) delimiting the transaction: {rows:?}" + ); + assert!( + rows.iter().all(|r| r.change_id > 0), + "change_id must be monotonic positive: {rows:?}" + ); +} + +/// Cross-transaction ordering: two sequential txns should produce two +/// COMMIT records with monotonically increasing change_ids. +#[tokio::test] +async fn commit_records_delimit_transactions() { + let conn = open_conn(":memory:").await; + conn.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)", ()) + .await + .unwrap(); + enable_cdc(&conn).await; + + conn.execute("INSERT INTO t VALUES (1, 'a')", ()) + .await + .unwrap(); + conn.execute("INSERT INTO t VALUES (2, 'b')", ()) + .await + .unwrap(); + + let rows = read_cdc(&conn).await; + let commits: Vec<&CdcRow> = rows.iter().filter(|r| r.change_type == 2).collect(); + assert!( + commits.len() >= 2, + "expected at least 2 COMMIT records for 2 autocommit txns: {rows:?}" + ); + for w in commits.windows(2) { + assert!( + w[0].change_id < w[1].change_id, + "COMMIT change_ids not monotonic: {:?}", + commits + ); + } + // Every non-commit row should carry a change_txn_id (the field we're + // going to key exactly-once apply on in the ephpm replication layer). + for r in &rows { + if r.change_type != 2 { + assert!( + r.change_txn_id.is_some(), + "non-commit CDC row missing change_txn_id: {r:?}" + ); + } + } +} diff --git a/crates/litewire-turso/tests/cdc_replication.rs b/crates/litewire-turso/tests/cdc_replication.rs new file mode 100644 index 0000000..8b26b2b --- /dev/null +++ b/crates/litewire-turso/tests/cdc_replication.rs @@ -0,0 +1,374 @@ +//! End-to-end tests for the litewire-turso CDC tail/apply pipeline. +//! +//! Each test opens two `Turso` factories (two databases: "primary" and +//! "replica"), enables CDC on the primary, runs some SQL, tails the CDC +//! log, applies the batches to the replica, and asserts that the +//! replica's row/schema state matches the primary. +//! +//! # Why two file-backed databases and not `":memory:"` +//! +//! Turso 0.7.0's `":memory:"` mode gives every wire session on the same +//! factory a shared view — great for the isolation tests in the base +//! backend, but for replication we specifically need two *independent* +//! engines that only communicate via the CDC stream, not the shared +//! in-memory buffer. Temp files give us that. + +use std::sync::Arc; + +use litewire_backend::{Backend, Value}; +use litewire_turso::Turso; +use litewire_turso::cdc::{CdcTailer, TxnBatch, apply_batch, enable_cdc, read_watermark}; + +async fn temp_factory() -> (Arc, tempfile::NamedTempFile) { + let file = tempfile::NamedTempFile::new().expect("temp file"); + let path = file.path().to_str().unwrap().to_string(); + let factory = Arc::new(Turso::open(path).await.expect("open turso file")); + (factory, file) +} + +/// Drain every currently-available complete transaction from the tailer. +async fn drain_batches(tailer: &mut CdcTailer<'_>) -> Vec { + let mut out = Vec::new(); + while let Some(batch) = tailer.poll_batch().await.expect("poll_batch") { + out.push(batch); + } + out +} + +/// Apply a stream of batches to a replica connection sequentially. +async fn apply_all(replica_conn: &turso::Connection, batches: &[TxnBatch]) { + for b in batches { + apply_batch(replica_conn, b).await.expect("apply_batch"); + } +} + +async fn count_rows(backend: &Arc, table: &str) -> i64 { + let rs = backend + .query(&format!("SELECT COUNT(*) FROM \"{table}\""), &[]) + .await + .unwrap(); + match &rs.rows[0][0] { + Value::Integer(i) => *i, + v => panic!("count returned non-integer: {v:?}"), + } +} + +async fn all_rows(backend: &Arc, sql: &str) -> Vec> { + let rs = backend.query(sql, &[]).await.unwrap(); + rs.rows +} + +#[tokio::test] +async fn watermark_starts_at_zero_and_advances_monotonically() { + let (primary, _p) = temp_factory().await; + let (replica, _r) = temp_factory().await; + + // Primary: enable CDC on a session, run one txn. + let p_conn = primary.raw_connection().unwrap(); + enable_cdc(&p_conn).await.unwrap(); + p_conn + .execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)", ()) + .await + .unwrap(); + p_conn + .execute("INSERT INTO t VALUES (1, 'hello')", ()) + .await + .unwrap(); + + // Tail + apply. + let mut tailer = CdcTailer::new(&primary, 0); + let batches = drain_batches(&mut tailer).await; + assert!(!batches.is_empty(), "expected at least one batch"); + let expected_wm = batches.last().unwrap().commit_change_id(); + + let r_conn = replica.raw_connection().unwrap(); + assert_eq!(read_watermark(&r_conn).await.unwrap(), 0); + + apply_all(&r_conn, &batches).await; + + assert_eq!( + read_watermark(&r_conn).await.unwrap(), + expected_wm, + "watermark must advance to the last COMMIT change_id" + ); + // A tailer that resumes from expected_wm should now find nothing. + let mut tailer2 = CdcTailer::new(&primary, expected_wm); + assert!(tailer2.poll_batch().await.unwrap().is_none()); +} + +#[tokio::test] +async fn ddl_replicates_replica_can_read_the_table() { + let (primary, _p) = temp_factory().await; + let (replica, _r) = temp_factory().await; + + let p_conn = primary.raw_connection().unwrap(); + enable_cdc(&p_conn).await.unwrap(); + p_conn + .execute( + "CREATE TABLE posts (id INTEGER PRIMARY KEY, title TEXT NOT NULL)", + (), + ) + .await + .unwrap(); + + let mut tailer = CdcTailer::new(&primary, 0); + let batches = drain_batches(&mut tailer).await; + + let r_conn = replica.raw_connection().unwrap(); + apply_all(&r_conn, &batches).await; + + // The DDL should have been replayed on the replica; SELECT proves it. + let rs = replica + .query("SELECT COUNT(*) FROM posts", &[]) + .await + .unwrap(); + assert_eq!(rs.rows[0][0], Value::Integer(0)); +} + +#[tokio::test] +async fn dml_insert_replicates_row_values_match() { + let (primary, _p) = temp_factory().await; + let (replica, _r) = temp_factory().await; + + let p_conn = primary.raw_connection().unwrap(); + enable_cdc(&p_conn).await.unwrap(); + p_conn + .execute( + "CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT NOT NULL)", + (), + ) + .await + .unwrap(); + p_conn + .execute("INSERT INTO t VALUES (1, 'a')", ()) + .await + .unwrap(); + p_conn + .execute("INSERT INTO t VALUES (2, 'b')", ()) + .await + .unwrap(); + p_conn + .execute("INSERT INTO t VALUES (3, 'c')", ()) + .await + .unwrap(); + + let mut tailer = CdcTailer::new(&primary, 0); + let batches = drain_batches(&mut tailer).await; + + let r_conn = replica.raw_connection().unwrap(); + apply_all(&r_conn, &batches).await; + + let p_rows = all_rows(&primary, "SELECT id, v FROM t ORDER BY id").await; + let r_rows = all_rows(&replica, "SELECT id, v FROM t ORDER BY id").await; + assert_eq!(p_rows, r_rows, "replica rows must match primary"); + assert_eq!(count_rows(&replica, "t").await, 3); +} + +#[tokio::test] +async fn dml_delete_replicates() { + let (primary, _p) = temp_factory().await; + let (replica, _r) = temp_factory().await; + + let p_conn = primary.raw_connection().unwrap(); + enable_cdc(&p_conn).await.unwrap(); + p_conn + .execute( + "CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT NOT NULL)", + (), + ) + .await + .unwrap(); + for i in 1..=5 { + p_conn + .execute(&format!("INSERT INTO t VALUES ({i}, 'row-{i}')"), ()) + .await + .unwrap(); + } + p_conn + .execute("DELETE FROM t WHERE id >= 3", ()) + .await + .unwrap(); + + let mut tailer = CdcTailer::new(&primary, 0); + let batches = drain_batches(&mut tailer).await; + let r_conn = replica.raw_connection().unwrap(); + apply_all(&r_conn, &batches).await; + + let p_rows = all_rows(&primary, "SELECT id FROM t ORDER BY id").await; + let r_rows = all_rows(&replica, "SELECT id FROM t ORDER BY id").await; + assert_eq!(p_rows, r_rows); + assert_eq!(count_rows(&replica, "t").await, 2); +} + +#[tokio::test] +async fn dml_update_replicates_new_value() { + let (primary, _p) = temp_factory().await; + let (replica, _r) = temp_factory().await; + + let p_conn = primary.raw_connection().unwrap(); + enable_cdc(&p_conn).await.unwrap(); + p_conn + .execute( + "CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT NOT NULL)", + (), + ) + .await + .unwrap(); + p_conn + .execute("INSERT INTO t VALUES (1, 'before')", ()) + .await + .unwrap(); + p_conn + .execute("UPDATE t SET v = 'after' WHERE id = 1", ()) + .await + .unwrap(); + + let mut tailer = CdcTailer::new(&primary, 0); + let batches = drain_batches(&mut tailer).await; + let r_conn = replica.raw_connection().unwrap(); + apply_all(&r_conn, &batches).await; + + let r = replica + .query("SELECT v FROM t WHERE id = 1", &[]) + .await + .unwrap(); + assert_eq!(r.rows[0][0], Value::Text("after".into())); +} + +/// Idempotency: re-applying the same batch stream after a "crash" must +/// leave the replica in the same state (no double-inserts, no stale rows). +#[tokio::test] +async fn apply_is_idempotent_across_replays() { + let (primary, _p) = temp_factory().await; + let (replica, _r) = temp_factory().await; + + let p_conn = primary.raw_connection().unwrap(); + enable_cdc(&p_conn).await.unwrap(); + p_conn + .execute( + "CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT NOT NULL)", + (), + ) + .await + .unwrap(); + for i in 1..=10 { + p_conn + .execute(&format!("INSERT INTO t VALUES ({i}, 'v{i}')"), ()) + .await + .unwrap(); + } + + let mut tailer = CdcTailer::new(&primary, 0); + let batches = drain_batches(&mut tailer).await; + let r_conn = replica.raw_connection().unwrap(); + + // Apply once. + apply_all(&r_conn, &batches).await; + let after_first = count_rows(&replica, "t").await; + + // Apply again — every batch's commit_change_id <= watermark, so each + // apply is a no-op. + apply_all(&r_conn, &batches).await; + let after_second = count_rows(&replica, "t").await; + + assert_eq!(after_first, 10); + assert_eq!(after_first, after_second, "idempotent apply changed state"); +} + +/// Bootstrap → resume: apply DDL/DML batches, then commit more txns on +/// the primary, then resume the tailer from the recorded watermark and +/// apply only the new batches. This is the "replica catches up after +/// downtime" flow. +#[tokio::test] +async fn tailer_resume_from_watermark_only_ships_new_batches() { + let (primary, _p) = temp_factory().await; + let (replica, _r) = temp_factory().await; + + let p_conn = primary.raw_connection().unwrap(); + enable_cdc(&p_conn).await.unwrap(); + p_conn + .execute( + "CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT NOT NULL)", + (), + ) + .await + .unwrap(); + p_conn + .execute("INSERT INTO t VALUES (1, 'a')", ()) + .await + .unwrap(); + + let r_conn = replica.raw_connection().unwrap(); + { + let mut tailer = CdcTailer::new(&primary, 0); + let batches = drain_batches(&mut tailer).await; + apply_all(&r_conn, &batches).await; + } + let wm = read_watermark(&r_conn).await.unwrap(); + assert!(wm > 0); + + // Primary keeps going. + p_conn + .execute("INSERT INTO t VALUES (2, 'b')", ()) + .await + .unwrap(); + p_conn + .execute("INSERT INTO t VALUES (3, 'c')", ()) + .await + .unwrap(); + + // Resume: fresh tailer starting at the persisted watermark. + let mut tailer = CdcTailer::new(&primary, wm); + let new_batches = drain_batches(&mut tailer).await; + assert!(!new_batches.is_empty(), "expected new batches after resume"); + for b in &new_batches { + assert!( + b.commit_change_id() > wm, + "resumed tailer emitted a batch <= old watermark: {}", + b.commit_change_id() + ); + } + apply_all(&r_conn, &new_batches).await; + + assert_eq!(count_rows(&replica, "t").await, 3); +} + +/// Partial-transaction hold-back: this is subtle. The primary opens an +/// explicit BEGIN, inserts, and does *not* commit yet. The tailer must +/// return `None` (no COMMIT delimiter → no complete batch). +#[tokio::test] +async fn uncommitted_transaction_is_not_yielded() { + let (primary, _p) = temp_factory().await; + + let setup_conn = primary.raw_connection().unwrap(); + setup_conn + .execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)", ()) + .await + .unwrap(); + + let writer = primary.raw_connection().unwrap(); + enable_cdc(&writer).await.unwrap(); + writer.execute("BEGIN", ()).await.unwrap(); + writer + .execute("INSERT INTO t VALUES (99, 'uncommitted')", ()) + .await + .unwrap(); + + let mut tailer = CdcTailer::new(&primary, 0); + // Note: setup_conn had no CDC enabled so its CREATE TABLE isn't in + // the log; the log currently holds only the mid-transaction insert + // from `writer`. That row has no COMMIT delimiter yet. + assert!( + tailer.poll_batch().await.unwrap().is_none(), + "tailer yielded an uncommitted batch" + ); + + // Now commit and re-poll: the batch should surface. + writer.execute("COMMIT", ()).await.unwrap(); + let batch = tailer + .poll_batch() + .await + .unwrap() + .expect("committed batch should surface"); + assert!(batch.commit_change_id() > 0); +} From aeecb8186c3eab121d93a82163671d67c754dcfa Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Tue, 14 Jul 2026 20:20:01 -0700 Subject: [PATCH 2/4] feat(turso): TursoBuilder::enable_cdc_on_connect for factory-wide capture --- crates/litewire-turso/src/lib.rs | 31 ++++++++++++++++ .../litewire-turso/tests/cdc_replication.rs | 36 +++++++++++++++++++ 2 files changed, 67 insertions(+) diff --git a/crates/litewire-turso/src/lib.rs b/crates/litewire-turso/src/lib.rs index d6e7775..1679c79 100644 --- a/crates/litewire-turso/src/lib.rs +++ b/crates/litewire-turso/src/lib.rs @@ -106,6 +106,7 @@ pub struct TursoBuilder { path: String, busy_timeout_ms: u32, synchronous: Synchronous, + enable_cdc_on_connect: bool, } impl TursoBuilder { @@ -127,6 +128,22 @@ impl TursoBuilder { self } + /// **Experimental** — when true, every session opened via + /// [`Backend::connect`] auto-enables full CDC capture via + /// [`cdc::enable_cdc`]. Used by ePHPm's Phase 2 CDC-native + /// replication on the primary so writes coming in via the wire + /// frontends are captured for replay by replicas. + /// + /// Default: `false`. Enabling CDC has a modest write-amp cost + /// (`full` mode doubles the write path: pre-image + post-image + /// records) and only makes sense when a tailer downstream is + /// consuming the log. + #[must_use] + pub fn enable_cdc_on_connect(mut self, on: bool) -> Self { + self.enable_cdc_on_connect = on; + self + } + /// Finalize the builder: open (or create) the database with the Turso /// engine. The engine is WAL-native; no journal-mode bootstrap is /// required. @@ -144,6 +161,7 @@ impl TursoBuilder { db, busy_timeout_ms: self.busy_timeout_ms, synchronous: self.synchronous, + enable_cdc_on_connect: self.enable_cdc_on_connect, }) } } @@ -157,6 +175,9 @@ pub struct Turso { pub(crate) db: turso::Database, busy_timeout_ms: u32, synchronous: Synchronous, + /// If set, [`Backend::connect`] enables full CDC capture on every + /// session. See [`TursoBuilder::enable_cdc_on_connect`]. + pub(crate) enable_cdc_on_connect: bool, } impl Turso { @@ -204,6 +225,7 @@ impl Turso { path: path.as_ref().to_string(), busy_timeout_ms: 5000, synchronous: Synchronous::Normal, + enable_cdc_on_connect: false, } } } @@ -228,6 +250,15 @@ impl Backend for Turso { conn.pragma_update("synchronous", self.synchronous.as_pragma_str()) .await .map_err(map_turso_err)?; + if self.enable_cdc_on_connect { + // Experimental: opt every wire session into CDC capture. Only + // set when litewire-turso is being used as the primary in + // ePHPm's Phase 2 replication mode. The pragma is + // per-connection, so a session that skips this (e.g. a + // direct `raw_connection()` caller like the tail loop) is + // unaffected. + cdc::enable_cdc(&conn).await?; + } Ok(Box::new(TursoConn { conn: Mutex::new(conn), })) diff --git a/crates/litewire-turso/tests/cdc_replication.rs b/crates/litewire-turso/tests/cdc_replication.rs index 8b26b2b..08ecc57 100644 --- a/crates/litewire-turso/tests/cdc_replication.rs +++ b/crates/litewire-turso/tests/cdc_replication.rs @@ -333,6 +333,42 @@ async fn tailer_resume_from_watermark_only_ships_new_batches() { assert_eq!(count_rows(&replica, "t").await, 3); } +/// The factory-level `enable_cdc_on_connect` opt-in causes every +/// wire-session `Backend::connect` to opt into CDC — so writes coming in +/// via the litewire wire frontends are captured for replication (this is +/// the seam ePHPm's Phase 2 primary depends on). +#[tokio::test] +async fn factory_flag_enables_cdc_on_backend_connect() { + let file = tempfile::NamedTempFile::new().unwrap(); + let path = file.path().to_str().unwrap(); + let factory = Turso::builder(path) + .enable_cdc_on_connect(true) + .build() + .await + .unwrap(); + let factory = Arc::new(factory); + + // A wire session obtained via Backend::connect should have CDC live. + let session = factory.connect().await.unwrap(); + session + .execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)", &[]) + .await + .unwrap(); + session + .execute("INSERT INTO t VALUES (1, 'x')", &[]) + .await + .unwrap(); + + // A tailer using the same factory should see the batch (CREATE + + // INSERT + COMMIT). + let mut tailer = CdcTailer::new(&factory, 0); + let batches = drain_batches(&mut tailer).await; + assert!( + !batches.is_empty(), + "enable_cdc_on_connect did not cause writes to be captured" + ); +} + /// Partial-transaction hold-back: this is subtle. The primary opens an /// explicit BEGIN, inserts, and does *not* commit yet. The tailer must /// return `None` (no COMMIT delimiter → no complete batch). From b70e7e7fef4fd4df1024b7ebdfee92adddc698cc Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Tue, 14 Jul 2026 20:20:58 -0700 Subject: [PATCH 3/4] test(turso): verify two Turso factories on same file coexist in one process --- .../tests/multi_factory_same_file.rs | 46 +++++++++++++++++++ 1 file changed, 46 insertions(+) create mode 100644 crates/litewire-turso/tests/multi_factory_same_file.rs diff --git a/crates/litewire-turso/tests/multi_factory_same_file.rs b/crates/litewire-turso/tests/multi_factory_same_file.rs new file mode 100644 index 0000000..fe001c8 --- /dev/null +++ b/crates/litewire-turso/tests/multi_factory_same_file.rs @@ -0,0 +1,46 @@ +//! Empirical answer to a Phase 2 design question: can two `Turso` +//! factories open the same database file in the same process safely? +//! +//! Turso 0.7.0 says "no multi-process support" — but that leaves the +//! multi-Database-per-process case ambiguous. ePHPm's Phase 2 primary +//! needs one factory serving the litewire wire frontends (with +//! `enable_cdc_on_connect(true)`) and a second factory for the CDC tail +//! loop; if the engine refuses, we need a different design. + +use litewire_backend::{Backend, Value}; +use litewire_turso::Turso; + +#[tokio::test] +async fn two_factories_on_same_file_can_coexist() { + let file = tempfile::NamedTempFile::new().unwrap(); + let path = file.path().to_str().unwrap().to_string(); + + // Factory A: the "wire frontend" role. Do some writes. + let a = Turso::open(&path).await.expect("open A"); + a.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)", &[]) + .await + .expect("create via A"); + a.execute("INSERT INTO t VALUES (1, 'a')", &[]) + .await + .expect("insert via A"); + + // Factory B: the "tail loop" role. Should see A's writes. + let b = Turso::open(&path).await.expect("open B"); + let rs = b + .query("SELECT id, v FROM t ORDER BY id", &[]) + .await + .expect("query via B"); + assert_eq!(rs.rows.len(), 1, "B failed to see A's committed row"); + assert_eq!(rs.rows[0][0], Value::Integer(1)); + assert_eq!(rs.rows[0][1], Value::Text("a".into())); + + // Writes via B should also be visible via A (after commit). + b.execute("INSERT INTO t VALUES (2, 'b')", &[]) + .await + .expect("insert via B"); + let rs = a + .query("SELECT COUNT(*) FROM t", &[]) + .await + .expect("count via A"); + assert_eq!(rs.rows[0][0], Value::Integer(2)); +} From 677795f805edd080813c8e7e456e0be927ed72ce Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Tue, 14 Jul 2026 20:27:16 -0700 Subject: [PATCH 4/4] feat(turso): re-export turso::Connection as litewire_turso::TursoConnection --- crates/litewire-turso/src/lib.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/crates/litewire-turso/src/lib.rs b/crates/litewire-turso/src/lib.rs index 1679c79..c086bb7 100644 --- a/crates/litewire-turso/src/lib.rs +++ b/crates/litewire-turso/src/lib.rs @@ -64,6 +64,12 @@ pub mod cdc; +/// Re-export of the underlying [`turso::Connection`] type. External +/// callers (e.g. ePHPm's Phase 2 CDC replication layer) need this to +/// type their apply-side function signatures without taking a direct +/// `turso` dependency. +pub use turso::Connection as TursoConnection; + use std::time::Duration; use litewire_backend::{