rindle_cdc_apply/

lib.rs

//! # rindle-cdc-apply — the engine-free CDC apply plane (design 309)
//!
//! "Apply a rindle CDC stream to a SQLite database" as a complete, closed capability —
//! **a follower without IVM**. The CDC family has capture (`rindle-cdc`), transport
//! (`rindle-stream`/`rindle-replicator`), and image distribution (`rindle-cdc-image`);
//! this crate is the apply quarter:
//!
//! - **table registration** — declared ([`ApplyConsumer::register_table`]) or
//!   introspected ([`ApplyConsumer::register_existing_table`], with the `_rindle_columns`
//!   PG-mirror sidecar override);
//! - **row apply** — decode a normalized op, build its SQL, execute it on the single
//!   observed writer connection ([`ApplyConsumer::apply_muts`]);
//! - **the cursor discipline** — the `_rindle_source_offsets` upsert riding the SAME
//!   transaction as the effects (a crash can never land data without its cursor or
//!   vice-versa), run-fencing (`run_id`), mid-run `chunk_seq`, and the durable head
//!   stamps ([`ApplyConsumer::commit_follower_txn_with_head`]);
//! - **replicated DDL** — statements + idempotency marker + caller bookkeeping in ONE
//!   ordinary transaction, `marker present ⇔ DDL applied` exactly
//!   ([`ApplyStore::exec_ddl_with_marker`]);
//! - **resume + portability** — [`ApplyConsumer::source_checkpoint`],
//!   [`ApplyStore::checkpoint_truncate`], and (in `rindle-writeplane`) the restore
//!   bootstrap.
//!
//! ## The two hosts
//!
//! A **headless** consumer — `rindle-backup-sqlite`'s portable journal replay, or any
//! external process mirroring a rindle stream into its own SQLite — opens with
//! [`ApplyConsumer::open`] and composes nothing else: no worker threads, no engines, no
//! drain. The **live replica** (`rindle_replica::Cluster`/`ClusterConsumer`) recomposes
//! on top: its worker pool implements [`CommitFanout`], and every apply method there
//! DELEGATES here — one implementation, moved, never forked — so the live follower and
//! a headless replay stay **byte-identical** by construction.
//!
//! The seam is [`CommitFanout`]: the write transaction hands its captured chunks to
//! whatever observes the commit, and *absence* of a derivation pool is a type
//! ([`NoFanout`]), not a zero-worker parameter. Control flow is identical either way —
//! the capture buffer still drains at [`PUSH_CHUNK_ROWS`], the guards still gate every
//! commit, the barriers just have no one to hold.
//!
//! Capture (the `rindle-cdc` preupdate hook) is mandatory in both hosts: its divergence
//! guards (`uncaptured_user_event_count`, capture-error poisoning) protect the headless
//! paths too — a DDL bounce that misses a registration surfaces as uncaptured rows
//! instead of silent drift.
//!
//! The derivation (query) surface deliberately does not exist here — an
//! [`ApplyConsumer`] has no `query`/`read_snapshot`/`view_schema`, so a host that
//! cannot derive cannot express the call (the type system replaces the refusal rule,
//! design 309 §3).

mod connection;
mod consumer;
mod fanout;
mod store;
mod txn;

pub use connection::{
    open_journal, open_journal_read_only, set_wal_autocheckpoint, ANALYSIS_LIMIT,
};
pub use consumer::{
    client_mutations_table_meta, ensure_source_offsets_table_on_conn, source_offsets_table_ddl,
    upsert_source_offset, upsert_source_offset_hashed, ApplyConsumer, SourceHead,
    APPLIED_DDL_TABLE,
};
pub use fanout::{CommitFanout, FanoutGate, FanoutStream, NoFanout, StreamBegin, PUSH_CHUNK_ROWS};
pub use store::{
    ensure_unique_pk_index_for, ApplyStore, DdlMigrationError, WalCheckpoint,
    DEFAULT_WRITER_BEGIN_SQL,
};
pub use txn::{ApplyTxn, CommitInfo};