rindle_cdc_apply/

fanout.rs

//! The **commit fan-out seam** (design 309 §3): how an open apply transaction hands its
//! captured row deltas to whatever wants to observe the commit — the live replica's IVM
//! worker pool, or nothing at all.
//!
//! The apply plane's write-transaction state machine ([`ApplyTxn`](crate::ApplyTxn))
//! interacts with its observer through exactly ONE call shape: `tx_begin(tx_id)` →
//! `Ready`/`Failed`, then `push(chunk)`* / `finish()` → a gate released with `commit()` or
//! `abort()`. That shape is lifted here as a trait so *absence* of a derivation pool is a
//! **type** ([`NoFanout`]), not a zero-worker parameter value — the dropped first
//! extraction attempt proved the zero-worker shape wrong at runtime (two
//! `assert!(n_workers > 0)`s and six `% n_workers` sites; design 309 §2).
//!
//! Control flow through the transaction is IDENTICAL under both implementations: the
//! capture buffer still drains at [`PUSH_CHUNK_ROWS`] (so the memory bound for a large
//! restore transaction survives headless), the begin/finish barriers still run — they
//! just have no one to hold.

use std::sync::Arc;

use rindle_cdc::Captured;

/// Upper bound on one streamed fan-out chunk, in captured rows
/// (`CLUSTER-INCREMENTAL-FANOUT-DESIGN.md` §4.2). Bounds the writer's transient capture
/// buffer — and, on the live replica, the per-worker transient output and the
/// cross-thread message — to `O(chunk)` rather than `O(txn)`; a txn ≤ this size streams
/// as a single chunk. Owned by the apply plane because the between-statement drain
/// (`ApplyTxn::maybe_pump`) is what enforces the writer-side bound, with or without a
/// derivation pool on the other end.
pub const PUSH_CHUNK_ROWS: usize = 1024;

/// The observer of an apply store's commits. Implemented by the live replica's worker
/// pool (fanning captured chunks out to the IVM workers behind the snapshot/commit
/// handshake) and by [`NoFanout`] (a headless applier — backup replay, an external
/// stream consumer — with no derivation to feed).
pub trait CommitFanout {
    /// Open the fan-out for transaction `tx_id` (the post-(N-1) → N cursor). A **send**,
    /// not a wait (`follow-ups/cluster-barrier-wakeup.md` §6.1): the observer pins its
    /// pre-commit state concurrently with the caller's remaining statements, and the
    /// caller collects the confirmation at the commit edge
    /// ([`FanoutStream::await_acks`] — the fence). `Ready` ⇒ stream chunks and finish;
    /// `Failed` ⇒ the observer was unreachable at send — the caller must NOT commit: it
    /// finishes the returned stream with an abort and fails the write (see
    /// [`ApplyTxn::commit_with_info`](crate::ApplyTxn)).
    fn tx_begin(&self, tx_id: u64) -> StreamBegin;

    /// Repair after a `Failed` begin at commit time (the live pool reaps + respawns the
    /// offending worker so the caller's retry can succeed). Headless: nothing to repair.
    fn recover_failed_begin(&self) {}
}

/// The result of [`CommitFanout::tx_begin`]. Either way a stream handle is returned so
/// whatever the observer pinned at the begin-barrier is always released.
#[must_use = "a StreamBegin carries a stream that must be finished to release the fan-out"]
pub enum StreamBegin {
    Ready(Box<dyn FanoutStream>),
    Failed(Box<dyn FanoutStream>),
}

/// One in-flight transaction's fan-out: bounded [`push`](Self::push) chunks, then a
/// terminal [`finish`](Self::finish) handing back the commit-verdict gate. Dropping it
/// without `finish` is an abort — the implementation must release anything it pinned.
pub trait FanoutStream {
    /// The transaction id this stream was opened for — carried on the stream so a
    /// lazily-opened mid-transaction fan-out keeps its cursor without recomputation.
    fn tx_id(&self) -> u64;

    /// Forward one bounded chunk (≤ [`PUSH_CHUNK_ROWS`] rows, one shared `Arc` — a
    /// refcount bump, never a row copy). May block for backpressure; must not fail
    /// (a lost observer is the implementation's business to repair at `finish`).
    fn push(&mut self, changes: Arc<[Captured]>);

    /// Collect the begin-barrier acks sent back since [`CommitFanout::tx_begin`] — the
    /// fence that must pass **before** the caller's durable COMMIT makes the transaction
    /// visible (an observer pinning late would see a base that already contains it).
    /// Called at the commit edge, after the last [`push`](Self::push); in the common case
    /// the acks arrived while the caller ran its own statements and nothing parks.
    /// `false` ⇒ an observer died or stayed stuck past the fence's watchdog: the caller
    /// must NOT commit — finish with an abort,
    /// [`recover_failed_begin`](CommitFanout::recover_failed_begin), and fail the write.
    /// Headless observers have no acks to collect; the default always passes.
    fn await_acks(&mut self) -> bool {
        true
    }

    /// Close the push stream and return the gate the caller releases after (or instead
    /// of) the durable COMMIT.
    fn finish(self: Box<Self>) -> Box<dyn FanoutGate>;
}

/// The commit verdict for an in-flight fan-out, released after the writer's durable
/// COMMIT (or its failure).
pub trait FanoutGate {
    /// The COMMIT succeeded: the observer publishes what it derived.
    fn commit(self: Box<Self>);

    /// The COMMIT failed (or never ran): the observer discards / recovers.
    fn abort(self: Box<Self>);
}

/// The headless fan-out: accepts pushes by dropping them, commits trivially. What a
/// consumer that applies a foreign CDC stream with **no derivation** composes with —
/// `rindle-backup-sqlite`'s portable replay, an external SQLite-mirroring consumer.
/// `tx_begin` never fails, so the `Failed` arms of the transaction machinery are
/// structurally unreachable headless.
pub struct NoFanout;

impl CommitFanout for NoFanout {
    fn tx_begin(&self, tx_id: u64) -> StreamBegin {
        StreamBegin::Ready(Box::new(NoStream { tx_id }))
    }
}

struct NoStream {
    tx_id: u64,
}

impl FanoutStream for NoStream {
    fn tx_id(&self) -> u64 {
        self.tx_id
    }

    fn push(&mut self, _changes: Arc<[Captured]>) {}

    fn finish(self: Box<Self>) -> Box<dyn FanoutGate> {
        Box::new(NoGate)
    }
}

struct NoGate;

impl FanoutGate for NoGate {
    fn commit(self: Box<Self>) {}

    fn abort(self: Box<Self>) {}
}