feat(turso): experimental CDC tail/apply API for external replication - #10
Merged
Conversation
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<TxnBatch>` — 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.
6 tasks
luthermonson
marked this pull request as ready for review
July 15, 2026 04:20
luthermonson
added a commit
to ephpm/ephpm
that referenced
this pull request
Jul 20, 2026
…ngine)
Alternative to the sqld sidecar clustered path for `[db.sqlite] engine =
"turso"`, gated behind a new opt-in knob. **Strictly additive** —
default config unchanged, sqld remains the production clustered default
for `engine = "sqlite"`. Turso engine still marked EXPERIMENTAL/Beta.
`[db.sqlite.replication] cdc_experimental = true` (default `false`).
When combined with `engine = "turso"` and `[cluster] enabled = true`,
selects the Phase 2 CDC-native replication path instead of failing the
old startup error. Without the flag, that combination still errors and
now points at the knob.
`cdc_listen` (default `"0.0.0.0:5015"`) is the TCP address the primary
serves CDC batches on. Ignored when `cdc_experimental = false`.
Both fields follow add-config-knob discipline: read and acted upon in
this same PR, unit-tested in `validate_sqlite_engine` and the ephpm-config
suite, doc-commented on the struct, and enumerated in reference/config.md.
Each node opens two `Turso` factories against the same DB file (verified
safe in one process by
`litewire-turso/tests/multi_factory_same_file.rs`):
- **wire factory** — served to litewire. On the primary, opts every
session into CDC via the new `TursoBuilder::enable_cdc_on_connect(true)`
so writes coming in through the MySQL/PG/Hrana frontends are captured.
- **mgmt factory** — used by the primary tail loop (`CdcTailer` polling
`turso_cdc` → `broadcast::Sender<Arc<TxnBatch>>`) or the replica apply
loop (`apply_batch` keyed to `__litewire_cdc_watermark`).
Primary spawns a TCP subscriber server on `cdc_listen`; every subscriber
gets its own `broadcast::Receiver` and receives length-prefixed
JSON-framed batches (base64 for the SQLite record blobs — v1 chooses
debuggability over a few bytes). Replicas open a TCP connection with an
auto-reconnect loop and apply batches via `apply_batch`.
Failover reuses the existing `SqliteElection`; the `cdc_listen` address
is what gets advertised for lookup by the new-primary auto path.
**Turso 0.7.0 CDC captures DDL.** `CREATE TABLE`/`CREATE INDEX`/
`ALTER TABLE ADD COLUMN`/`DROP TABLE` all appear in the same
`turso_cdc` stream as row DML, encoded as mutations on `sqlite_schema`
where the `after`-image's `sql` column carries the CREATE text.
Verified by `litewire-turso/tests/cdc_ddl_capture.rs` (7 tests). This
finding is the *reason* the replication architecture is a single
ordered stream — no schema-sync side channel needed, WordPress/Laravel
runtime DDL flows through the normal path.
- ephpm-config: 79 tests pass (default + new-fields parse).
- ephpm-server unit tests: 189 pass — includes 3 new `turso_cdc` tests
(wire-batch round-trip, frame codec, oversized-frame rejection) plus
updated `validate_sqlite_engine` tests for the four
(engine × clustered × cdc_experimental) combinations.
- ephpm-server integration test (`turso_cdc_e2e`): 2 tests pass:
1. `two_node_cdc_replicates_ddl_and_dml_end_to_end` — two Turso
engines in one process, a real TCP socket between them; primary
runs CREATE TABLE + INSERT + INSERT + UPDATE + DELETE via the
wire factory; replica converges to 1 row with the updated value
within 5s. Watermark on replica > 0 confirms exactly-once apply.
2. `replica_reconnect_does_not_double_apply` — after the replica
task is aborted and a fresh one started, the watermark check
causes already-applied batches to no-op; the row count stays at 5.
- clippy pedantic -D warnings: clean.
- nightly rustfmt: clean.
- Fresh-replica snapshot bootstrap (v1: operator seeds DB file).
- Persisted subscriber watermark across primary restart.
- CDC traffic encryption (v1: trusted-network or TLS-tunneled).
- `turso_cdc` retention pruning.
- 2-node podman/kind e2e running the ephpm binary against a real MySQL
client (the in-process integration test proves the pipeline; the
compose lift is test orchestration).
- Wire-frontend capture for sessions that skip the factory flag (v1
gotcha: `raw_connection()` callers bypass CDC).
**ephpm/litewire#10** — feat/turso-cdc adds the `CdcTailer`,
`apply_batch`, `enable_cdc_on_connect`, and `TursoConnection`
re-export. `Cargo.toml` temporarily pins litewire to that branch head
(`677795f805ed`). **After litewire#10 merges: re-pin `litewire` to the
main merge commit and `cargo update -p litewire` before merging this
PR.**
Building the actual pipeline forced three corrections to the Phase 1
source-only read of the CDC schema:
1. v2 `turso_cdc` column order is
`(change_id, change_time, change_txn_id, change_type, table_name, id,
before, after, updates)` — `change_txn_id` is column 3, not column 9.
2. `change_type` values: `INSERT=1, UPDATE=0, DELETE=-1, COMMIT=2`.
3. `turso_cdc_version.version` is TEXT `"v2"`, not integer `2`.
We also do NOT adopt `turso_sync_engine` in v1: its coroutine execution
model doesn't compose with tokio, and its use of `turso_core` primitives
would require opening a second Database handle on the same file through
a completely different code path.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Companion to ephpm Phase 2 (roadmap) — adds the CDC tail/apply API ephpm's replication layer needs from litewire. Experimental, additive, no changes to the
Backend/BackendConntraits, no consumer other than ephpm's opt-in[db.sqlite.replication] cdc_experimental = trueknob.Headline empirical finding: Turso 0.7.0 CDC captures DDL.
CREATE TABLE,CREATE INDEX,ALTER TABLE ADD COLUMN,DROP TABLEall appear in the sameturso_cdcstream as row DML, encoded as mutations onsqlite_schema. Theafter-image'ssqlcolumn carries the DDL text — the applier re-executes it. This means the replication design is a single ordered stream (no schema-sync side channel), which is a strictly simpler architecture than Phase 1 anticipated.What lands
Turso::raw_connection()— bareturso::Connectionfor enabling CDC pragma and tailing.cdc::enable_cdc()/cdc::CdcTailer/cdc::apply_batch()/cdc::decode_sqlite_record().__litewire_cdc_watermarkadvances in the same replica transaction as the replayed rows;commit_change_id <= current_watermarkshort-circuits to a no-op.turso_core's private record API.Test plan
cargo test -p litewire-turso— 25/25 pass onturso = "=0.7.0":tests/litewire_basics.rs(2) — unchangedtests/cdc_ddl_capture.rs(7) — CDC schema version, DDL capture for CREATE TABLE/INDEX/DROP TABLE/ALTER TABLE, INSERT+COMMIT delimiter, cross-txn orderingtests/cdc_replication.rs(8) — full tail→apply pipeline between two independent file-backed factories: DDL replicates, DML INSERT/UPDATE/DELETE match primary, watermark monotonic, resume-from-watermark ships only new batches, apply is idempotent across double-replay, uncommitted transactions do not surfacecargo clippy -p litewire-turso --all-targets -- -D warningscleancargo fmt -p litewire-tursocleanCorrections to the Phase 1 design doc
Building the actual implementation surfaced three corrections in ephpm's
docs/turso-phase2-cdc-design.md(updated in the companion ephpm PR):(change_id, change_time, change_txn_id, change_type, table_name, id, before, after, updates)—change_txn_idis column 3, not column 9.change_typevalues areINSERT=1, UPDATE=0, DELETE=-1, COMMIT=2(not0=DELETE, 1=INSERT).turso_cdc_version.versionis TEXT —"v2", not integer2.Scope caveats
Documented in
cdc.rsmodule docs:turso_cdcretention/pruning is the caller's concern — the tailer reads but never deletes.