Struct ClusterConsumer 

pub struct ClusterConsumer { /* private fields */ }

The cluster-backed consumer (coordinator side). !Send — lives on one thread; the drain pushes batches/progress/faults into the sink from its own thread.

Implementations

impl ClusterConsumer

pub fn open( path: &Path, n_workers: usize, sink: impl DrainSink + 'static, ) -> Result<ClusterConsumer, ReplicaError>

Open a cluster-backed consumer over path with n_workers IVM worker threads, wiring drain output into sink. The caller owns the file’s lifecycle. A fresh file gets plain wal (design 306 D5); the daemon’s roles pass their wal2 opt-in via open_with_journal.

pub fn open_with_journal( path: &Path, n_workers: usize, journal: JournalMode, sink: impl DrainSink + 'static, ) -> Result<ClusterConsumer, ReplicaError>

open with an explicit fresh-file crate::JournalMode — the daemon passes crate::JournalMode::Wal2 (checkpoint headroom + the follower/backup fleet contract; an existing wal/wal2 file keeps its mode).

pub fn open_with( path: &Path, n_workers: usize, opts: OpenOptions, sink: impl DrainSink + 'static, ) -> Result<ClusterConsumer, ReplicaError>

The full-combination opener: every crate::OpenOptions field is honored — the daemon’s escape hatch for combining wal2 with a non-default derivation mode or operator storage (e.g. the design-306 soak fallback) without a new rung per combination.

pub fn cluster(&self) -> &Cluster

The underlying coordinator (controlled writes, lmid reads, raw event taps).

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 a follower runs instead of enforcing per row. Delegates to Cluster::foreign_key_audit; see there for the cost, the cap, and why enforcement and verification are separable.

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

Define + register a base table (CREATE TABLE + cluster source). Idempotent.

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 + derivation.

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

One-time setup for client mutations: the _rindle_client_mutations table is created, captured, AND engine-hosted (lmid-as-data) — its meta joins the table map so the per-client system query’s hello resolves like any table’s.

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

One-time setup for the §4 realtime lifecycle (Cluster::enable_realtime_lifecycle — see Db::enable_realtime_lifecycle for the table roster and why each is registered): the four lifecycle tables join the table map so doorbell / fence / outcome / ledger subscriptions resolve their hello like any table’s — ROOM_CLIENT_MUTATIONS_TABLE had NO meta before this (Slice C kept it direct-SQL-only). All four are _rindle_*-prefixed and so excluded from base_table_schemas. Requires enable_client_mutations first. Enabling also switches on commit_room_flush’s §4.2 watermark co-commit.

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 refresh_visibility(&self) -> Result<bool, ReplicaError>

Re-read the visibility sidecars after a ddl entry (design 406 §10); true when the advertised schema changed and a bounce is due.

pub fn client_lmid(&self, client_id: &str) -> Result<u64, ReplicaError>

The durably stored high-water mutation id for client_id (0 if new).

pub fn producer_seq(&self, producer: &str) -> Result<u64, BookkeepingError>

The durable last sequence a foreign producer wrote (0 if new). This deliberately runs the shared connection-only helper on the observed writer while no mutation is open, keeping the read serialized with every write-plane decision.

pub fn seed_producer_census( &self, census: &ProducerCensus, ) -> Result<(), BookkeepingError>

Read the distinct-producer count into census off the observed writer — the one scan the incrementally-maintained gauge takes, at open (design 306 §4). Same narrow-access rule as producer_seq: the shared helper runs here rather than exposing the writer connection across crates.

pub fn migration_record( &self, id: &str, ) -> Result<Option<MigrationRecord>, BookkeepingError>

Read one durable deploy/public migration identity while the standalone engine owns the idle writer. Kept narrow instead of exposing the raw writer connection across crates.

pub fn data_migration_checksum( &self, id: &str, ) -> Result<Option<String>, BookkeepingError>

Read the captured data-migration marker for id on the serialized writer.

pub fn journal_local_migration( &self, id: &str, checksum: Option<&str>, content_checksum: &str, run_id: &str, statements: &str, applied_at: i64, ) -> Result<(), BookkeepingError>

Persist the standalone producer’s DDL journal row after the marker transaction. A crash between the two is healed by replay: the durable marker skips DDL and this insert retries.

pub fn adopt_migration_checksums( &self, id: &str, checksum: Option<&str>, content_checksum: &str, ) -> Result<(), BookkeepingError>

Fill legacy migration identity columns without minting a captured data commit.

pub fn stored_public_outcome( &self, outcome_key: &str, ) -> Result<Option<StoredPublicOutcome>, BookkeepingError>

Read one exact public-SQL outcome while the serialized standalone writer is idle.

pub fn public_operation_floor_ms( &self, now_ms: i64, ) -> Result<i64, BookkeepingError>

The durable one-shot outcome retention floor at now_ms.

pub fn public_commit_outcome( &self, transaction_id: &str, now_ms: i64, ) -> Result<Option<Option<String>>, SessionError>

Resolve a retried public transaction commit from standalone’s co-transactional outcome.

pub fn sweep_sql_outcomes(&self, now_ms: i64) -> Result<(), BookkeepingError>

Sweep the bounded outcome cache in one host-local metadata transaction.

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).

pub fn connect(&self, conn: ConnId)

Register a connection (drain-side progress bookkeeping).

pub fn disconnect(&self, conn: ConnId)

Drop a connection’s drain-side progress bookkeeping. The caller destroys the connection’s queries separately (via destroy_query).

pub fn query_normalized( &self, conn: ConnId, server_qid: u64, ast: Ast, epoch: u64, ) -> Result<NormalizedHello, ReplicaError>

Register a NORMALIZED live query for conn under server_qid. Returns the slim hello synchronously (schema-derived); the seq-0 snapshot and every later batch arrive asynchronously through the sink.

The 1:1 convenience over the shared-query primitives: it registers an engine query and attaches a single subscriber whose sub id is the same server_qid. A server that dedups identical ASTs across subscribers uses register_shared_query + attach_subscriber directly instead.

pub fn register_shared_query( &self, eqid: u64, ast: Ast, ) -> Result<SharedQueryInfo, ReplicaError>

Register a shared engine query under eqid (one IVM pipeline + one footprint fold), with NO subscriber yet. Identical ASTs registered under one eqid are computed once; subscribers attach via attach_subscriber, each with its own epoch / seq cursor / connection route (RINDLE-SERVER-DESIGN.md §6/§11). Returns the deterministic table set + fingerprint the caller turns into each subscriber’s hello.

pub fn register_shared_family( &self, eqid: u64, template: &FamilyTemplate, bindings: &[Binding], ) -> Result<Vec<(Binding, SharedQueryInfo)>, ReplicaError>

Register a parameterized query family under eqid (design 310 §5.1/§5.2): one IVM pipeline over the template, plus one footprint fold per binding, each built from the concrete member AST (FamilyTemplate::instantiate, impl plan D8) so every frame a partition emits equals the standalone query’s byte for byte. Returns each binding’s SharedQueryInfo (its tables + fingerprint — the same for every member, modulo the literal, but computed per member so nothing is assumed).

pub fn bind_shared( &self, eqid: u64, template: &FamilyTemplate, binding: &Binding, ) -> Result<SharedQueryInfo, ReplicaError>

Bind one more partition of the family registered under eqid (design 310 §4.4): the engine hydrates only that partition; the drain gets its own footprint.

pub fn unbind_shared( &self, eqid: u64, binding: &Binding, ) -> Result<bool, ReplicaError>

Unbind one partition of the family under eqid: the engine drains it silently, the drain drops its footprint (faulting any subscriber still on it).

pub fn attach_partition_subscriber( &self, eqid: u64, binding: &Binding, sub: u64, conn: ConnId, epoch: u64, )

Attach subscriber sub to one partition of a shared family (the family form of attach_subscriber).

pub fn partition_snapshot( &self, eqid: u64, template: &FamilyTemplate, binding: &Binding, ) -> Result<Value, ReplicaError>

The one-shot snapshot of ONE partition of a family (the family form of query_snapshot): the family’s assembled view filtered to the rows whose partition key is binding, rendered under the concrete member’s view schema.

pub fn query_snapshot( &self, eqid: u64, ast: &Ast, ) -> Result<Value, ReplicaError>

Read the assembled view of a registered shared query as a one-shot snapshot — the SSR REST path (SSR-DESIGN.md §3). No subscriber, no streaming, no lease: it re-reads the live view of eqid and renders it to the flat/nested wire JSON a stateless API server can hydrate directly (cells keyed by name, relationships nested inline). The query must already be registered (via register_shared_query); an absent query (or degraded shard) yields an empty array. ast is the same AST the query was registered under — it supplies the view schema (relationship names / projection / .one() shape).

pub fn attach_subscriber(&self, eqid: u64, sub: u64, conn: ConnId, epoch: u64)

Attach subscriber sub (its own conn route + epoch) to a shared engine query. Its seq-0 snapshot (from the cached footprint if the query is already hydrated, else when it hydrates) and every later batch arrive asynchronously through the sink. Build the subscriber’s hello from the SharedQueryInfo register_shared_query returned (see hello_for).

pub fn detach_subscriber(&self, sub: u64)

Detach one subscriber; the shared engine query and its peers keep running.

pub fn fault_subscribers(&self, eqid: u64, reason: &str)

Fault every subscriber on a shared engine query (each gets a terminal faulted) and detach them, leaving the query for a following destroy_query — gives active subscribers a re-subscribe signal before an explicit dematerialize.

pub fn destroy_query(&self, server_qid: u64)

Tear down a registered query (cluster pipeline + drain bookkeeping, dropping every subscriber still on it). For the 1:1 query_normalized path this removes the lone subscriber too.

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: ClusterWriteTxn, 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 ClusterWriteTxn via cluster.write() 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: ClusterWriteTxn, 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 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_room_placement_table(&self) -> Result<(), ReplicaError>

Create the _rindle_room_placement fencing table (idempotent). Unregistered bookkeeping like the offsets table.

pub fn claim_room_epoch(&self, doc: &str) -> Result<i64, ReplicaError>

Claim the next placement epoch for doc (§2.5): one write transaction, one monotone bump. The claim is what fences every prior epoch’s flushes.

pub fn room_epoch(&self, doc: &str) -> Result<Option<i64>, ReplicaError>

The current placement epoch for doc (None = never claimed — a fence-bearing request against it is stale by definition).

pub fn room_lmids( &self, doc: &str, clients: &[String], ) -> Result<Vec<(String, i64)>, ReplicaError>

The domain-scoped ledger’s last_mutation_id per client under doc (absent = 0 to the caller) — the room’s boot probe (§3.3): what lets a rebooted room absorb already-durable mutations as replay dedup instead of double-applying them. Reads ROOM_CLIENT_MUTATIONS_TABLE keyed by (doc, client_id) (§7.1) — NOT the slow-path CLIENT_MUTATIONS_TABLE, so a client’s room stream and its daemon stream never alias.

pub fn commit_room_flush( &self, muts: &[Mutation], source: &str, offset: &str, doc: Option<&str>, batch_hash: Option<&str>, cas: bool, ) -> Result<RoomFlushOutcome, ReplicaError>

Apply a room flush batch (§5.3 step 5): optional CAS preconditions — every change’s old image (Edit.old / Remove’s old / Add’s asserted absence) must match the current row under the identity comparator (null == null, one number domain; NOT join semantics) — then effects + cursor + batch identity in ONE transaction. Any CAS miss rolls the whole batch back and returns the authoritative current images. The caller owns the fence and the dedup (both race-free on the single-threaded engine).

When doc is Some this is a room flush: its ledger co-edits (§5.3 step 3 — the changes targeting CLIENT_MUTATIONS_TABLE) are domain-scoped and retarget to ROOM_CLIENT_MUTATIONS_TABLE keyed by (doc, client_id) (§7.1). They are pulled out of the CAS + effects pass and upserted monotonically in the same transaction, so the daemon’s slow-path CLIENT_MUTATIONS_TABLE row is never perturbed by a room flush (§8.5’s ledger-isolation invariant — the Rev 1 data-loss bug). When doc is None (a plain change source / replicator follower) the batch applies verbatim, so the slow-path lmid stream still rides its own CLIENT_MUTATIONS_TABLE rows unchanged.

When the §4 lifecycle is enabled (enable_realtime_lifecycle) a room flush also co-commits the §4.2 downgrade-fence watermark — ROOM_WATERMARK_TABLE(doc, flush_seq = offset), monotone — in the SAME transaction, so the fence rides the flush’s echo through every authority shape (see the inline comment for the dedup/regression reasoning).

A room flush’s outcome co-edits (Slice I-ii — changes targeting ROOM_MUTATION_OUTCOMES_TABLE, doc-less 6-wide rows) split off exactly like the ledger’s: they NEVER enter the CAS pass or apply_muts — the split is unconditional for a doc flush (a lifecycle-disabled daemon would otherwise fail the whole batch on the unregistered table), while the upsert itself is gated on realtime_lifecycle_enabled, so a lifecycle-disabled daemon accepts a batch carrying outcome rows and silently discards them (it has no downgrade path to resolve through — graceful, not an error). With the lifecycle on, each row upserts keyed (doc, client_id, mid) (idempotent on replay by PK) in the SAME transaction as the lmid rows covering those mids — the I-ii soundness contract: an absent outcome row under a covering daemon-carried lmid means applied. Retention prunes in the same transaction by lmid DISTANCE (OUTCOME_RETENTION_LMIDS), for exactly the (doc, client) rows this flush advanced.

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_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, so retries can detect applied DDL from the durable marker. Delegates to Cluster::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 Cluster::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 begin_mutation(&self) -> Result<ClusterMutationWrite, ReplicaError>

Open one mutation’s write transaction; run SQL against the handle, then ClusterMutationWrite::commit_with_lmid lands effects + lmid atomically (or commit for a foreign write).

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

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

pub fn normalized_table_schemas( &self, ast: &Ast, ) -> Result<Vec<TableWireSchema>, ReplicaError>

The flat schema of every base table the query’s tree can surface (root + each related/EXISTS child table), for the publisher’s hello + PK map. A relationship aggregate surfaces a SYNTHETIC table (AGGREGATE-SYNC-DESIGN.md §3.2) whose schema is derived from the AST (agg_table_schemas), not the DB registry — the child rows it replaces are never synced.