Struct ApplyConsumer 

pub struct ApplyConsumer { /* private fields */ }

The engine-free CDC apply consumer. !Send — lives on one thread, like the store under it. See the module docs; construction is open (headless) or from_parts (a derivation host composing over a shared store).

Implementations

impl ApplyConsumer

pub fn open(path: &Path) -> Result<ApplyConsumer, ReplicaError>

Open a headless apply consumer over path. The caller owns the file’s lifecycle. A fresh file gets plain wal (design 306 D5) and writes open BEGIN IMMEDIATE — exactly what ClusterConsumer::open’s defaults resolve to, so the headless applier and the default live follower produce byte-identical stores.

Foreign keys are not enforced ([ForeignKeys::Unenforced]), which is the apply plane’s posture and not a relaxation: the rows this consumer writes were validated by the authority that produced the stream, they arrive in the stream’s order rather than a topological one, and the authority’s ON DELETE/ON UPDATE actions are already in the stream as ordinary row changes — re-running them here would apply each cascade twice. [rindle_writeplane::foreign_keys] has the full argument, and foreign_key_audit is how a host proves referential integrity anyway. A host that really is the origin of its rows says so with open_with.

pub fn open_with_journal( path: &Path, journal: JournalMode, ) -> Result<ApplyConsumer, ReplicaError>

open with an explicit fresh-file [JournalMode] — a wal2-fleet host passes [JournalMode::Wal2] (an existing wal/wal2 file keeps its mode).

pub fn open_with( path: &Path, journal: JournalMode, foreign_keys: ForeignKeys, ) -> Result<ApplyConsumer, ReplicaError>

open_with_journal with an explicit [ForeignKeys] posture — the escape hatch for a host that is the ORIGIN of the rows it writes through this consumer rather than a replayer of somebody else’s. Everything else resolves exactly as open.

pub fn open_scratch(path: &Path) -> Result<ApplyConsumer, ReplicaError>

Open a headless consumer over a scratch database — single-connection, journal_mode=memory, synchronous=OFF. See ApplyStore::open_scratch for what that trades and what the caller owes in return; the applied result is identical, only the crash posture differs.

A scratch is by definition DERIVED from an authority that can rebuild it, so it is an apply-plane store and opens [ForeignKeys::Unenforced] like open.

pub fn foreign_key_audit( &self, max_rows: usize, ) -> Result<ForeignKeyAudit, ReplicaError>

Walk every declared foreign key and report violating rows — the opt-in audit that stands in for enforcement on this store. Delegates to ApplyStore::foreign_key_audit; see there for the cost and the cap.

pub fn from_parts( store: Rc<ApplyStore>, fanout: Rc<dyn CommitFanout>, ) -> ApplyConsumer

Compose a consumer over an existing store + fan-out — the seam a derivation host (rindle_replica::ClusterConsumer) uses to share ONE store between its cluster and its apply surface.

pub fn refresh_visibility(&self) -> Result<bool, ReplicaError>

Re-read the visibility sidecars (after a ddl entry, design 406 §9 step 6). true when the hidden set changed — the caller bounces so the advertised schema moves with it.

pub fn hidden(&self) -> Hidden

The current hidden set (406 §10).

pub fn store(&self) -> &ApplyStore

The underlying store (controlled writes, ad-hoc reads, DDL, checkpoints).

pub fn tables(&self) -> MutexGuard<'_, HashMap<String, TableMeta>>

The registered-table metadata registry — one entry per table this consumer can apply into, exactly what apply_muts builds its SQL from. A derivation host reads and extends it (its lifecycle tables carry meta too); most callers never touch it directly.

pub fn begin(&self) -> Result<ApplyTxn, ReplicaError>

Open one write transaction on the store (the follower’s streaming apply drives apply_muts into it per chunk frame, then commit_follower_txn_with_head).

pub fn register_table( &self, table: &str, columns: &[String], pk: &[usize], col_types: &[ColType], ) -> Result<(), ReplicaError>

Define + register a base table (CREATE TABLE + capture registration). Idempotent.

pub fn register_table_via( &self, table: &str, columns: &[String], pk: &[usize], col_types: &[ColType], register: impl FnOnce() -> Result<(), ReplicaError>, ) -> Result<(), ReplicaError>

register_table with the registration step swapped out — the derivation-host seam: ClusterConsumer passes its engine-inclusive Cluster::register_table so the capture half AND the worker sources build in the one place they always did, while the DDL and the meta bookkeeping stay here.

pub fn register_existing_table(&self, table: &str) -> Result<(), ReplicaError>

Register an EXISTING base table (already created by the host’s own DDL): reads its column order + pk from SQLite and registers it for capture (a derivation host adds its sources through the _via seam).

pub fn register_existing_table_via( &self, table: &str, register: impl FnOnce() -> Result<(), ReplicaError>, ) -> Result<(), ReplicaError>

register_existing_table with the registration step swapped out (see register_table_via).

pub fn enable_client_mutations(&self) -> Result<(), ReplicaError>

One-time setup for client mutations on a headless applier: the _rindle_client_mutations (and room-scoped) ledger tables are created and the former registered for capture — its rows are replicated data on the master’s stream, so an applier that skips this stalls on the first lmid row with unknown table. The meta joins the table map like any table’s.

pub fn enable_client_mutations_via( &self, enable: impl FnOnce() -> Result<(), ReplicaError>, ) -> Result<(), ReplicaError>

enable_client_mutations with the DDL + registration step swapped out — ClusterConsumer passes Cluster::enable_client_mutations (the same DDL plus engine hosting) so the meta bookkeeping stays here either way.

pub fn base_table_schemas(&self) -> Vec<BaseTableSchema>

The registered base tables’ schemas (name + ordered columns/types + PK names), sorted by name, for client-schema codegen via /schema (DRIZZLE-MIGRATIONS-DESIGN.md §6.2). Excludes the daemon’s own bookkeeping/internal tables — _rindle_* (e.g. _rindle_client_mutations), __* (e.g. the __replica_meta commit watermark), and sqlite_* — which must stay invisible to the client schema just as they are to CDC + planning (rindle-cdc skips sqlite_* + __*). Reads straight from the in-memory table map — the live introspected schema, no DB round-trip.

pub fn apply_muts( &self, txn: &mut ApplyTxn, muts: &[Mutation], ) -> Result<(), ReplicaError>

Apply positional mutations to an open write transaction (build_sql per row).

pub fn commit_normalized(&self, muts: &[Mutation]) -> Result<u64, ReplicaError>

Apply a batch of positional mutations as one raw foreign write (no lmid, confirms nothing), returning the commit version synchronously.

pub fn commit_normalized_with_offset( &self, muts: &[Mutation], source: &str, offset: &str, chunk_seq: i64, run_id: Option<&str>, ) -> Result<u64, ReplicaError>

Apply a change-source batch AND advance the source’s durable cursor in ONE write txn (CHANGE-SOURCE-DESIGN.md §4). The _rindle_source_offsets upsert rides the same transaction as the effects — exactly the upsert_lmid discipline — so a crash can never commit the data without the cursor (or vice-versa). The caller owns the monotonic-absorb dedup (offset <= stored ⇒ skip) BEFORE calling this; there is no gap rejection (the source owns contiguity, §4).

pub fn commit_follower_txn( &self, txn: ApplyTxn, source: &str, offset: &str, chunk_seq: i64, run_id: Option<&str>, ) -> Result<u64, ReplicaError>

Terminal step of the streaming-follower apply (REPLICATOR-PRECOMMIT-STREAMING-DESIGN.md §7): the caller has opened ONE ApplyTxn via begin and driven apply_muts into it once per chunk frame; this upserts the source cursor in that SAME open txn (co-transactional with the chunk applies — a crash can never commit the data without the cursor) and commits, returning the commit version. It is exactly commit_normalized_with_offset’s cursor discipline, but with the row-changes already applied incrementally as chunks arrived rather than handed over as one batch.

pub fn commit_follower_txn_with_head( &self, txn: ApplyTxn, source: &str, offset: &str, chunk_seq: i64, run_id: Option<&str>, head: SourceHead, ) -> Result<u64, ReplicaError>

commit_follower_txn plus the frame’s durable row-count/commit stamps. Effects, cursor, run fence, and head accounting land in one transaction.

pub fn refresh_follower_run_id( &self, source: &str, offset: &str, previous_run_id: &str, next_run_id: &str, ) -> Result<u64, ReplicaError>

Advance only a CDC transport locator at an unchanged semantic cursor.

The source-offset table is deliberately unregistered, so this empty application commit emits no IVM row delta. The compare-and-swap keeps a stale connection from replacing a newer locator, and the update remains a real SQLite transaction so a portable image observes either locator in full, never torn metadata.

pub fn ensure_source_offsets_table(&self) -> Result<(), ReplicaError>

Create the _rindle_source_offsets bookkeeping table (idempotent). Not registered for capture — daemon metadata, like _rindle_sql_outcomes.

pub fn source_checkpoint( &self, source: &str, ) -> Result<Option<(String, i64, Option<String>)>, ReplicaError>

The durably-stored (offset, chunk_seq, run_id) checkpoint for source (None ⇒ never applied; the caller treats that as the genesis "" and subscribes from the start). chunk_seq is [SOURCE_OFFSET_WHOLE_RUN] for a whole-run checkpoint (the common case) or a real within-run ordinal for a mid-run segment left by the commit-at-DDL-boundary follower (§6.6). The resume/dedup keyset is (offset, chunk_seq); run_id is the checkpointed run’s identity token, echoed on the subscribe as the fencing proof (RELAY-CURSOR-EPOCH-FENCING-DESIGN.md §2) — None for a pre-fence checkpoint.

pub fn source_head( &self, source: &str, ) -> Result<Option<SourceHead>, ReplicaError>

Persisted accounting carried beside the source checkpoint.

pub fn source_checkpoint_hash( &self, source: &str, ) -> Result<Option<String>, ReplicaError>

The stored §8.3 batch identity for source’s checkpoint (None = no row, or a hash-less source). Compared — never recomputed — against a resubmission’s declared hash at the exact stored offset.

pub fn source_offset( &self, source: &str, ) -> Result<Option<String>, ReplicaError>

The durably-stored cursor string for source, discarding the chunk_seq sub-position — for the string-only callers (snapshot-restore resume points, which are always run boundaries). The resume/dedup paths use source_checkpoint for the full keyset.

pub fn ensure_producer_offsets_table(&self) -> Result<(), ReplicaError>

Create the _rindle_producer_offsets foreign-write watermark table (idempotent). The DDL is the shared one, so this cannot drift from the write-master’s or the restore’s copy. Callers register it for capture afterwards — it is replicated data, not host bookkeeping (design 306 §3.3).

pub fn ensure_applied_ddl_table(&self) -> Result<(), ReplicaError>

Create the _rindle_applied_ddl idempotency journal (idempotent). Not registered for capture — daemon metadata, like _rindle_source_offsets. actions holds the entry’s ordered apply report (design 227 fourth review pass), written in the same transaction as the marker.

pub fn ddl_already_applied(&self, key: &str) -> Result<bool, ReplicaError>

Whether a ddl entry keyed by key (migration id / offset) is already journaled — the crash-window-replay dedup, checked BEFORE re-applying.

pub fn apply_ddl_with_marker( &self, key: &str, statements: &[String], ) -> Result<DdlApplyReport, ReplicaError>

Apply a ddl entry’s statements and journal its key atomically — one ordinary transaction, since DDL can’t ride the BEGIN CONCURRENT cursor advance. Delegates to ApplyStore::exec_ddl_with_marker against the APPLIED_DDL_TABLE journal.

pub fn apply_ddl_with_marker_and_step_effects<F>( &self, key: &str, statements: &[String], apply_step_effects: F, ) -> Result<DdlApplyReport, ReplicaError>

where F: FnMut(&Connection, &DdlStep) -> Result<()>,

Apply DDL, caller-owned per-statement bookkeeping effects, and the durable marker in one transaction. See ApplyStore::exec_ddl_with_marker_and_step_effects.

pub fn stored_ddl_report( &self, key: &str, ) -> Result<Option<DdlApplyReport>, ReplicaError>

The ordered apply report persisted with key’s marker (same transaction as the DDL), or None for entries marked before the report column existed (they degrade to the caller’s end-state fallback). A replay consumes this instead of re-observing — the DDL does not re-run, so there is nothing to observe (design 227 fourth review pass).

pub fn apply_public_ddl_operation<F>( &self, statement: &SqlStatementRequest, declared_tables: &[String], outcome_key: &str, request_identity: &str, result_byte_limit: usize, now_ms: i64, apply_step_effects: F, ) -> Result<(PublicOperationCommit, DdlApplyReport), DdlMigrationError>

where F: FnMut(&Connection, &DdlStep) -> Result<()>,

Fresh standalone public DDL: schema + desired-index effects + exact replay outcome + TxId watermark commit as one SQLite atom.

pub fn apply_public_ddl_migration<F>( &self, id: &str, supplied_checksum: Option<&str>, content_checksum: &str, normalized: &[String], identity_json: &str, declared_tables: &[String], now_ms: i64, apply_step_effects: F, ) -> Result<(String, DdlApplyReport), DdlMigrationError>

where F: FnMut(&Connection, &DdlStep) -> Result<()>,

Fresh standalone DDL migration, with its permanent identity row and exact TxId cursor committed in the checked DDL transaction rather than backfilled afterward. Both front doors — the public /v1/sql/migrate route and the private deploy route — apply through this one primitive so they mint identical journal rows and either can absorb the other’s replay; only the opaque checksum is optional (the deploy surface accepts checksum-less DDL files, whose identity is the statement vector alone).