rindle_replica/

drain.rs

//! The **drain thread** — the keystone of the cluster fold-in (`CLUSTER-FOLD-IN-DESIGN.md`
//! §5/§6/§8). It moves the consumer-facing fold (per-query `NormalizedPublisher`) and the
//! progress/`cv_min` layer off the single-thread `Db`'s synchronous `commit` callback and
//! onto an asynchronous consumer of the [`Cluster`](crate::Cluster) channel-out.
//!
//! ## What it owns
//! - the `Receiver<ClusterEvent>` returned by [`Cluster::open`](crate::Cluster::open)
//!   (events cross from the worker threads as `Send` data);
//! - a control channel from the coordinator (the Node main thread) carrying query
//!   installs/removals and connection lifecycle;
//! - the `query_id → NormalizedPublisher` map (the stateful `CaughtChange → NormalizedBatch`
//!   fold that on the single-thread path lived in the sync subscribe callback);
//! - the **per-worker position table** + the event-driven progress core (§8): a query's
//!   known-current `cv` is the position of its hosting worker, so a slow worker holds back
//!   only the clients with a query on it.
//!
//! There is NO per-commit metadata channel: mutation confirmation (`lmid`) is a row in
//! [`crate::CLIENT_MUTATIONS_TABLE`], delivered through the client's own system query
//! like any data — so it shares the worker event stream's ordering instead of racing it
//! from the coordinator thread (the root cause of the stale-lmid release bug this
//! replaced; see `CLUSTER-FOLD-IN-FINDINGS.md`).
//!
//! ## How the two input streams are merged
//! `std::sync::mpsc` has no select, so both inputs are funnelled into one ordered
//! [`DrainMsg`] channel: a tiny forwarder thread wraps each [`ClusterEvent`] as
//! `DrainMsg::Event`, and the coordinator sends `DrainMsg::Control` on a clone of the same
//! sender. This preserves the **install-before-`Hydrated`** ordering for free: the
//! coordinator sends `InstallQuery` and *then* calls `cluster.query` (which only emits
//! `Hydrated` after the worker replies), so the install is enqueued before the worker
//! event it pairs with. (A not-yet-installed query's events are still buffered defensively.)
//!
//! ## Ordering invariants preserved
//! - **`Changed(N)` before `Progressed(N)`** per worker (the worker emits in that order):
//!   the data frames for tx N are folded + delivered before the progress frame that
//!   releases them.
//! - **data frames before the progress frame** per connection: a connection's progress
//!   frame for tx N is emitted only once its queries' workers have reported `Progressed(N)`,
//!   so its `cv_min` advance never out-runs the data it releases.

use std::collections::{BTreeSet, HashMap};
use std::sync::mpsc::{self, Receiver, SyncSender};
use std::thread::{self, JoinHandle};

use crate::normalize::NormalizeFold;
use crate::normalize_protocol::{project_ops, NormalizedBatch, ProgressFrame, ProjMap};
use crate::{ChangeEvent, ClusterEvent, QueryId, Update};
use rindle::canon::canonical_key;
use rindle::value::ColId;
use rindle_wire::family_key::Binding;

/// A consumer-assigned connection identifier (one per subscriber/WebSocket). The drain
/// groups queries by connection for the `cv_min`/poke computation; it never interprets it.
pub type ConnId = u64;

/// The sink the drain delivers finished output to. Implemented by the napi layer over a
/// `ThreadsafeFunction` (→ JS `onEvent`) and by tests over a collector. Called **on the
/// drain thread**, so keep each call cheap (marshal + hand off).
pub trait DrainSink: Send {
    /// A query's normalized batch (the seq-0 hydrate snapshot or an incremental tick) is
    /// ready — route it to the connection that owns the query. Delivered **eagerly**
    /// (before the progress frame that releases it).
    fn batch(&mut self, conn: ConnId, query_id: QueryId, batch: NormalizedBatch);
    /// A connection's progress frame (`{cv_min}`) — emitted after the data frames it
    /// releases, per the poke rule.
    fn progress(&mut self, conn: ConnId, frame: ProgressFrame);
    /// A query faulted on its worker (terminal): the connection should re-subscribe
    /// (re-hydrate). No further events arrive for this `query_id`.
    fn faulted(&mut self, conn: ConnId, query_id: QueryId, reason: String);
    /// An **engine query** faulted — the cluster discarded its pipeline (`parallel.rs`
    /// `fault_recover`), so a consumer that owns the query's lifecycle (e.g. a
    /// materialization manager) must arrange to **re-register** it; its subscribers are
    /// separately told to re-subscribe via [`faulted`](Self::faulted). Called once per
    /// faulted engine query, BEFORE its subscribers are notified, with the **engine** query
    /// id (not a subscription id) and the fault's classification — a push-deadline bail
    /// (FOLLOWER-LAG-SHED §6.6) must reach the owner's mode machine before the re-register
    /// decision. Default: no-op (the 1:1/napi path re-registers on the next subscribe).
    fn query_faulted(&mut self, _query_id: QueryId, _cause: crate::FaultCause) {}
}

/// A subscription id — one per attached subscriber (the routing identity the [`DrainSink`]
/// delivers on, surfaced as a [`QueryId`] in the sink callbacks). Distinct from the
/// **engine** query id: many subscribers can share one engine query (dedup), each with its
/// own epoch / seq cursor / connection route.
pub type SubId = u64;

/// The control plane: coordinator → drain. Kept private; the coordinator drives it through
/// [`DrainHandle`]'s methods.
enum DrainControl {
    /// Register a (possibly shared) engine query: its hosting worker, the shared footprint
    /// fold + table set/fingerprint, and the hydrate watermark floor. Subscribers attach
    /// separately so an identical AST registers ONE engine pipeline (`§11`).
    RegisterQuery {
        query_id: QueryId,
        worker: usize,
        fold: Box<NormalizeFold>,
        normalized_fp: u64,
        /// Project-at-emit map: every batch emitted for this query is projected through it
        /// before fan-out (empty ⇒ pass-through). Shared by all subscribers (one AST).
        proj: ProjMap,
        hydrated_cv: u64,
    },
    /// Register a parameterized query family (design 310 §5.2): routing/progress state plus
    /// the partition key the drain demuxes `Changed` events by; partitions arrive
    /// separately via [`BindPartition`](DrainControl::BindPartition).
    RegisterFamily {
        query_id: QueryId,
        worker: usize,
        param_cols: Vec<ColId>,
        hydrated_cv: u64,
    },
    /// Add one partition to a family: its own footprint fold / fingerprint / projection,
    /// built from the **concrete** member AST (impl plan D8) so every frame equals the
    /// standalone query's byte for byte. A `PartitionHydrated` (or a `Changed` bucket)
    /// that raced ahead of this control is replayed onto it.
    BindPartition {
        query_id: QueryId,
        binding: Binding,
        fold: Box<NormalizeFold>,
        normalized_fp: u64,
        proj: ProjMap,
        hydrated_cv: u64,
    },
    /// Drop one partition's footprint (the "view slice" of design §4.4); any subscriber
    /// still on it is faulted with a distinct reason.
    UnbindPartition {
        query_id: QueryId,
        binding: Binding,
    },
    /// Attach a subscriber to an engine query: its own conn route + epoch. If the query is
    /// already hydrated the seq-0 snapshot is emitted from the cached footprint immediately;
    /// otherwise it lands when the query hydrates. `binding` names the partition when the
    /// query is a family (an attach to an unbound partition is ignored).
    AttachSubscriber {
        query_id: QueryId,
        sub: SubId,
        conn: ConnId,
        epoch: u64,
        binding: Option<Binding>,
    },
    /// Detach one subscriber (its lease/connection went away); the shared engine query and
    /// its other subscribers are untouched.
    DetachSubscriber {
        sub: SubId,
    },
    /// Notify every subscriber on an engine query that it faulted/was revoked (each gets a
    /// terminal `faulted`) and detach them, WITHOUT removing the query. Used to give active
    /// subscribers a re-subscribe signal before an explicit dematerialize tears the query
    /// down (a silent [`RemoveQuery`](DrainControl::RemoveQuery) would just drop them).
    FaultSubscribers {
        query_id: QueryId,
        reason: String,
    },
    /// Tear down an engine query and every subscriber still on it (the materialization was
    /// dematerialized / swept).
    RemoveQuery {
        query_id: QueryId,
    },
    Connect {
        conn: ConnId,
    },
    Disconnect {
        conn: ConnId,
    },
    Shutdown,
}

/// Bound on the drain loop's unified input queue, in messages (FOLLOWER-LAG-SHED §4,
/// rung 0b) — the second hop of the delivery chain. A full queue blocks the forwarder (which
/// stops draining the worker channel-out) and, transiently, a control sender; the loop's sink
/// never blocks, so the queue always drains.
const LOOP_CHANNEL_BOUND: usize = 256;

/// The unified, ordered input to the drain loop (worker events + coordinator control).
enum DrainMsg {
    Event(ClusterEvent),
    Control(DrainControl),
}

/// The coordinator-side handle: a cheap clone of the control sender. Lives on the Node main
/// thread; its methods are called from the napi `Db` surface as queries are
/// registered/destroyed, connections come and go, and commits land.
#[derive(Clone)]
pub struct DrainHandle {
    tx: SyncSender<DrainMsg>,
}

impl DrainHandle {
    /// Register a (possibly shared) engine query's footprint fold + routing/progress state.
    /// Sent right after `cluster.query(query_id, ..)` returns the hosting `worker`; a
    /// `Hydrated` that races ahead of it is buffered and replayed on register.
    pub fn register_query(
        &self,
        query_id: QueryId,
        worker: usize,
        fold: NormalizeFold,
        normalized_fp: u64,
        proj: ProjMap,
        hydrated_cv: u64,
    ) {
        let _ = self.tx.send(DrainMsg::Control(DrainControl::RegisterQuery {
            query_id,
            worker,
            fold: Box::new(fold),
            normalized_fp,
            proj,
            hydrated_cv,
        }));
    }

    /// Register a parameterized query family's routing/progress state and partition key
    /// (design 310 §5.2). Sent right after `cluster.family(..)` returns the hosting worker;
    /// its partitions follow via [`bind_partition`](Self::bind_partition).
    pub fn register_family(
        &self,
        query_id: QueryId,
        worker: usize,
        param_cols: Vec<ColId>,
        hydrated_cv: u64,
    ) {
        let _ = self
            .tx
            .send(DrainMsg::Control(DrainControl::RegisterFamily {
                query_id,
                worker,
                param_cols,
                hydrated_cv,
            }));
    }

    /// Add one partition (its footprint fold, built from the concrete member AST) to a
    /// registered family.
    pub fn bind_partition(
        &self,
        query_id: QueryId,
        binding: Binding,
        fold: NormalizeFold,
        normalized_fp: u64,
        proj: ProjMap,
        hydrated_cv: u64,
    ) {
        let _ = self.tx.send(DrainMsg::Control(DrainControl::BindPartition {
            query_id,
            binding,
            fold: Box::new(fold),
            normalized_fp,
            proj,
            hydrated_cv,
        }));
    }

    /// Drop one partition's drain-side state (after `cluster.unbind`).
    pub fn unbind_partition(&self, query_id: QueryId, binding: Binding) {
        let _ = self
            .tx
            .send(DrainMsg::Control(DrainControl::UnbindPartition {
                query_id,
                binding,
            }));
    }

    /// Attach a subscriber (its own conn route + epoch) to a registered engine query.
    pub fn attach_subscriber(&self, query_id: QueryId, sub: SubId, conn: ConnId, epoch: u64) {
        let _ = self
            .tx
            .send(DrainMsg::Control(DrainControl::AttachSubscriber {
                query_id,
                sub,
                conn,
                epoch,
                binding: None,
            }));
    }

    /// Attach a subscriber to one partition of a registered family.
    pub fn attach_partition_subscriber(
        &self,
        query_id: QueryId,
        binding: Binding,
        sub: SubId,
        conn: ConnId,
        epoch: u64,
    ) {
        let _ = self
            .tx
            .send(DrainMsg::Control(DrainControl::AttachSubscriber {
                query_id,
                sub,
                conn,
                epoch,
                binding: Some(binding),
            }));
    }

    /// Detach one subscriber, leaving the shared engine query (and its peers) running.
    pub fn detach_subscriber(&self, sub: SubId) {
        let _ = self
            .tx
            .send(DrainMsg::Control(DrainControl::DetachSubscriber { sub }));
    }

    /// Fault every subscriber on an engine query (terminal `faulted` each) and detach them,
    /// leaving the query itself in place for the caller to [`remove_query`](Self::remove_query)
    /// next — e.g. an explicit dematerialize that has active subscribers to notify.
    pub fn fault_subscribers(&self, query_id: QueryId, reason: String) {
        let _ = self
            .tx
            .send(DrainMsg::Control(DrainControl::FaultSubscribers {
                query_id,
                reason,
            }));
    }

    /// Tear down an engine query's drain-side state (after `cluster.destroy_query`).
    pub fn remove_query(&self, query_id: QueryId) {
        let _ = self
            .tx
            .send(DrainMsg::Control(DrainControl::RemoveQuery { query_id }));
    }

    /// Register a connection under a consumer-assigned id.
    pub fn connect(&self, conn: ConnId) {
        let _ = self
            .tx
            .send(DrainMsg::Control(DrainControl::Connect { conn }));
    }

    /// Drop a connection and all its queries' drain-side state.
    pub fn disconnect(&self, conn: ConnId) {
        let _ = self
            .tx
            .send(DrainMsg::Control(DrainControl::Disconnect { conn }));
    }
}

/// The running drain: owns the two worker/forwarder/loop threads. Dropping (or
/// [`shutdown`](Self::shutdown)) stops the loop; the forwarder exits when the cluster's
/// channel-out closes (i.e. when the `Cluster` is dropped).
pub struct Drain {
    tx: SyncSender<DrainMsg>,
    loop_join: Option<JoinHandle<()>>,
    _forward_join: Option<JoinHandle<()>>,
}

impl Drain {
    /// Spawn the drain over a cluster's channel-out `events`, with `initial_cv` the
    /// replica's committed watermark at open (the progress frontier) and `sink` the
    /// delivery target. Returns the handle plus a [`DrainHandle`] for the coordinator.
    pub fn spawn<S: DrainSink + 'static>(
        events: Receiver<ClusterEvent>,
        initial_cv: u64,
        sink: S,
    ) -> Drain {
        // BOUNDED (FOLLOWER-LAG-SHED §4, rung 0b): this loop queue is the second unbounded hop
        // that used to hide behind the worker channel-out — bounding the first alone just
        // relocates the backlog here. A full queue blocks the forwarder, which stops draining
        // the channel-out, which blocks the workers: pressure propagates upstream by
        // construction. The loop's sink never blocks (a slow client is evicted), so the loop
        // always drains. Control senders (the engine thread) share the bound — a backlogged
        // drain briefly backpressures control too, which is the intended lock-step.
        let (tx, rx) = mpsc::sync_channel::<DrainMsg>(LOOP_CHANNEL_BOUND);

        // Forwarder: relay every worker event into the unified ordered stream. Exits when
        // the cluster's channel-out closes (Cluster dropped) or the loop has gone away.
        let fwd_tx = tx.clone();
        let forward_join = thread::Builder::new()
            .name("ivm-drain-forward".into())
            .spawn(move || {
                while let Ok(ev) = events.recv() {
                    if fwd_tx.send(DrainMsg::Event(ev)).is_err() {
                        break;
                    }
                }
            })
            .expect("spawn drain forwarder");

        let mut core = DrainCore::new(Box::new(sink), initial_cv);
        let loop_join = thread::Builder::new()
            .name("ivm-drain".into())
            .spawn(move || {
                while let Ok(msg) = rx.recv() {
                    match msg {
                        DrainMsg::Event(ev) => core.on_event(ev),
                        DrainMsg::Control(DrainControl::Shutdown) => break,
                        DrainMsg::Control(c) => core.on_control(c),
                    }
                }
            })
            .expect("spawn drain loop");

        Drain {
            tx,
            loop_join: Some(loop_join),
            _forward_join: Some(forward_join),
        }
    }

    /// A coordinator-side control handle (cheap clone).
    pub fn handle(&self) -> DrainHandle {
        DrainHandle {
            tx: self.tx.clone(),
        }
    }

    /// Stop the loop and join it. The forwarder is left to exit on cluster close.
    pub fn shutdown(mut self) {
        self.stop();
    }

    fn stop(&mut self) {
        let _ = self.tx.send(DrainMsg::Control(DrainControl::Shutdown));
        if let Some(j) = self.loop_join.take() {
            let _ = j.join();
        }
    }
}

impl Drop for Drain {
    fn drop(&mut self) {
        self.stop();
    }
}

// ---------------------------------------------------------------------------
// The drain core (single-thread state machine; runs on the drain thread)
// ---------------------------------------------------------------------------

/// One footprint fold and the subscribers fed from it: a singleton query's whole
/// state, or ONE partition of a parameterized query family (design 310 §5.2 / impl plan
/// D8 — a partition holds its own fold built from the concrete member AST, so its
/// `hello`, `normalized_fp`, and projection are byte-identical to the standalone's).
struct Footprint {
    /// The committed watermark this footprint hydrated at — a floor on its known-current
    /// `cv` (the hydrate reflects every commit ≤ this, even before the worker reports a
    /// position).
    hydrated_cv: u64,
    /// The latest commit version the footprint reflects — what a late subscriber's
    /// snapshot frame is stamped with (≥ `hydrated_cv`). Advances on each delivered tick;
    /// NOT used for release (that stays gated on the worker position).
    footprint_cv: u64,
    /// `true` once the hydrate snapshot has folded in — late subscribers can baseline off
    /// the cached footprint, earlier ones were baselined when it arrived.
    hydrated: bool,
    /// The shared footprint serializer: ONE per engine query / partition (the dedup win —
    /// identical ASTs fold once, §11). Snapshots a late subscriber via
    /// [`NormalizeFold::footprint_snapshot`].
    fold: NormalizeFold,
    normalized_fp: u64,
    /// Project-at-emit map (`PROJECTION-SUPPORT-DESIGN.md` §5.2): table → projected base-column
    /// indices. The shared fold runs over FULL rows; each emitted snapshot/batch is projected
    /// through this before fan-out. Empty ⇒ pass-through (a `'*'` query).
    proj: ProjMap,
    /// The subscribers fed from this footprint.
    subscribers: BTreeSet<SubId>,
}

impl Footprint {
    fn new(fold: NormalizeFold, normalized_fp: u64, proj: ProjMap, hydrated_cv: u64) -> Footprint {
        Footprint {
            hydrated_cv,
            footprint_cv: hydrated_cv,
            hydrated: false,
            fold,
            normalized_fp,
            proj,
            subscribers: BTreeSet::new(),
        }
    }
}

/// What kind of engine query a [`QueryReg`] routes: a singleton owns one footprint; a
/// family owns one per bound partition, keyed by the binding the root row's partition
/// columns canonicalize to.
enum QueryKind {
    Singleton(Footprint),
    Family {
        /// The partition key — the parameter columns on the root row (design §4.2).
        param_cols: Vec<ColId>,
        partitions: HashMap<Binding, Footprint>,
        /// Events for a binding whose `BindPartition` control has not landed yet (the
        /// worker's `PartitionHydrated` and any `Changed` bucket behind it can race the
        /// coordinator's control), replayed in order on the control. Only a
        /// `PartitionHydrated` opens an entry: a `Changed` bucket for a binding with
        /// neither a footprint nor a pending entry is the unbind race — dropped.
        pending: HashMap<Binding, Vec<Update>>,
    },
}

/// Per-registered **engine query** on the drain — its routing/progress state and its
/// footprint(s), shared across every subscriber on it.
struct QueryReg {
    worker: usize,
    /// The committed watermark the query registered at — the `cv` floor for a subscriber
    /// whose footprint is not (yet) known.
    hydrated_cv: u64,
    kind: QueryKind,
}

impl QueryReg {
    fn footprint(&self, binding: Option<&Binding>) -> Option<&Footprint> {
        match (&self.kind, binding) {
            (QueryKind::Singleton(f), _) => Some(f),
            (QueryKind::Family { partitions, .. }, Some(b)) => partitions.get(b),
            (QueryKind::Family { .. }, None) => None,
        }
    }

    fn footprint_mut(&mut self, binding: Option<&Binding>) -> Option<&mut Footprint> {
        match (&mut self.kind, binding) {
            (QueryKind::Singleton(f), _) => Some(f),
            (QueryKind::Family { partitions, .. }, Some(b)) => partitions.get_mut(b),
            (QueryKind::Family { .. }, None) => None,
        }
    }

    /// Every subscriber across every footprint.
    fn all_subscribers(&self) -> Vec<SubId> {
        match &self.kind {
            QueryKind::Singleton(f) => f.subscribers.iter().copied().collect(),
            QueryKind::Family { partitions, .. } => partitions
                .values()
                .flat_map(|f| f.subscribers.iter().copied())
                .collect(),
        }
    }

    /// Detach every subscriber across every footprint, returning them.
    fn take_all_subscribers(&mut self) -> Vec<SubId> {
        match &mut self.kind {
            QueryKind::Singleton(f) => std::mem::take(&mut f.subscribers).into_iter().collect(),
            QueryKind::Family { partitions, .. } => partitions
                .values_mut()
                .flat_map(|f| std::mem::take(&mut f.subscribers).into_iter())
                .collect(),
        }
    }

    fn remove_subscriber(&mut self, sub: SubId, binding: Option<&Binding>) {
        if let Some(f) = self.footprint_mut(binding) {
            f.subscribers.remove(&sub);
        }
    }
}

/// Per-subscriber framing + routing — many of these share one [`QueryReg`].
struct SubReg {
    query_id: QueryId,
    /// The partition this subscriber is on when the query is a family; `None` for a
    /// singleton.
    binding: Option<Binding>,
    conn: ConnId,
    epoch: u64,
    /// Gap-free seq over *emitted* batches; `0` until the seq-0 snapshot is sent.
    next_seq: u64,
    /// `false` until this subscriber's seq-0 snapshot has been emitted (a pre-hydrate
    /// attach waits for the engine query to hydrate).
    baselined: bool,
}

/// Per-connection state for the poke rule + coherent-release tracking.
struct ConnReg {
    /// The subscriptions on this connection (its `cv_min` floor is the min over them).
    subs: BTreeSet<SubId>,
    /// `cv_min` of the last progress frame sent (0 if none).
    released_cv: u64,
    /// Highest `cv` of a data frame (snapshot / batch) delivered to this connection — the
    /// release target. A progress frame is due only while this exceeds `released_cv`.
    max_delivered: u64,
}

struct DrainCore {
    sink: Box<dyn DrainSink>,
    /// The committed watermark (max `tx_id` seen) — the `cv_min` of a query-less connection.
    frontier: u64,
    /// worker → highest `tx_id` it has fully delivered (`Progressed`).
    worker_pos: HashMap<usize, u64>,
    queries: HashMap<QueryId, QueryReg>,
    subs: HashMap<SubId, SubReg>,
    conns: HashMap<ConnId, ConnReg>,
    /// Events that arrived for a query before its `RegisterQuery` (defensive; the
    /// coordinator registers right after `cluster.query`, but a `Hydrated` can race ahead).
    pending_events: HashMap<QueryId, Vec<Update>>,
}

impl DrainCore {
    fn new(sink: Box<dyn DrainSink>, initial_cv: u64) -> DrainCore {
        DrainCore {
            sink,
            frontier: initial_cv,
            worker_pos: HashMap::new(),
            queries: HashMap::new(),
            subs: HashMap::new(),
            conns: HashMap::new(),
            pending_events: HashMap::new(),
        }
    }

    // --- worker-event handling --------------------------------------------

    fn on_event(&mut self, ev: ClusterEvent) {
        match ev {
            ClusterEvent::Update { query_id, update } => self.on_update(query_id, update),
            ClusterEvent::Progressed { worker, tx_id } => self.on_progressed(worker, tx_id.0),
            ClusterEvent::Faulted {
                query_id,
                reason,
                cause,
            } => self.on_faulted(query_id, reason, cause),
        }
    }

    fn on_update(&mut self, query_id: QueryId, update: Update) {
        if !self.queries.contains_key(&query_id) {
            // Not yet registered — buffer and replay on RegisterQuery (never drop a baseline).
            self.pending_events
                .entry(query_id)
                .or_default()
                .push(update);
            return;
        }
        match update {
            Update::Hydrated { tx_id, changes } => {
                self.hydrate_footprint(query_id, None, tx_id.0, &changes);
            }
            Update::PartitionHydrated {
                tx_id,
                binding,
                changes,
            } => {
                let reg = self.queries.get_mut(&query_id).expect("present");
                match &mut reg.kind {
                    // A singleton never receives one; ignore rather than mis-fold.
                    QueryKind::Singleton(_) => {}
                    QueryKind::Family {
                        partitions,
                        pending,
                        ..
                    } => {
                        if partitions.contains_key(&binding) {
                            self.hydrate_footprint(query_id, Some(&binding), tx_id.0, &changes);
                        } else {
                            // The bind race: the worker's hydrate beat the coordinator's
                            // `BindPartition` control. Open the pending entry; replayed then.
                            pending.entry(binding.clone()).or_default().push(
                                Update::PartitionHydrated {
                                    tx_id,
                                    binding,
                                    changes,
                                },
                            );
                        }
                    }
                }
            }
            Update::Changed { tx_id, changes } => {
                let reg = self.queries.get(&query_id).expect("present");
                let param_cols = match &reg.kind {
                    QueryKind::Singleton(_) => {
                        self.change_footprint(query_id, None, tx_id.0, changes);
                        return;
                    }
                    QueryKind::Family { param_cols, .. } => param_cols.clone(),
                };
                for (binding, bucket) in demux_by_partition(changes, &param_cols) {
                    let reg = self.queries.get_mut(&query_id).expect("present");
                    let QueryKind::Family {
                        partitions,
                        pending,
                        ..
                    } = &mut reg.kind
                    else {
                        unreachable!("kind checked above")
                    };
                    if partitions.contains_key(&binding) {
                        self.change_footprint(query_id, Some(&binding), tx_id.0, bucket);
                    } else if let Some(p) = pending.get_mut(&binding) {
                        p.push(Update::Changed {
                            tx_id,
                            changes: bucket,
                        });
                    }
                    // else: the unbind race — a delta for a partition no longer bound.
                }
            }
        }
    }

    /// Fold a hydrate batch into one footprint (a singleton's, or a family partition's),
    /// mark it hydrated, and baseline every subscriber already waiting on it from that one
    /// footprint. The footprint-only fold skips the wire-op materialization — late
    /// subscribers baseline off `footprint_snapshot`, never the hydrate fold's return value.
    fn hydrate_footprint(
        &mut self,
        query_id: QueryId,
        binding: Option<&Binding>,
        cv: u64,
        changes: &[ChangeEvent],
    ) {
        let waiting: Vec<SubId> = {
            let reg = self.queries.get_mut(&query_id).expect("present");
            let Some(fp) = reg.footprint_mut(binding) else {
                return;
            };
            fp.fold.fold_footprint_only(changes);
            fp.hydrated = true;
            fp.footprint_cv = fp.footprint_cv.max(cv);
            fp.subscribers.iter().copied().collect()
        };
        for sub in waiting {
            self.baseline_subscriber(sub);
        }
    }

    /// Fold one commit's changes ONCE through a footprint; fan the net deltas out to every
    /// baselined subscriber on it, each stamped with its own epoch + seq. A fold that nets
    /// to nothing yields no batch (and no seq, keeping seq gap-free).
    fn change_footprint(
        &mut self,
        query_id: QueryId,
        binding: Option<&Binding>,
        cv: u64,
        changes: Vec<ChangeEvent>,
    ) {
        let (ops, subscribers, fp_id) = {
            let reg = self.queries.get_mut(&query_id).expect("present");
            let Some(fp) = reg.footprint_mut(binding) else {
                return;
            };
            fp.footprint_cv = fp.footprint_cv.max(cv);
            // No subscribers ⇒ keep the shared footprint current for late joiners
            // (their snapshot reads it), but skip building + projecting wire ops
            // nobody will receive. This is the steady-state pinned-board case — the
            // bulk of the per-commit allocation churn the wiki-demo RSS profiling
            // traced the glibc retention to.
            if fp.subscribers.is_empty() {
                fp.fold.fold_footprint_only(&changes);
                return;
            }
            // Fold over FULL rows once (the shared dedup win), then project the net
            // deltas once before fanning out — projection is per-query, so every
            // subscriber of this AST gets the same projected ops (§2.1/§5.2).
            let ops = project_ops(fp.fold.fold(&changes), &fp.proj);
            (
                ops,
                fp.subscribers.iter().copied().collect::<Vec<_>>(),
                fp.normalized_fp,
            )
        };
        if ops.is_empty() {
            return;
        }
        for sub in subscribers {
            let Some(s) = self.subs.get_mut(&sub) else {
                continue;
            };
            if !s.baselined {
                continue;
            }
            let batch = NormalizedBatch {
                epoch: s.epoch,
                seq: s.next_seq,
                cv,
                normalized_fp: fp_id,
                ops: ops.clone(),
            };
            s.next_seq += 1;
            let conn = s.conn;
            self.sink.batch(conn, QueryId(sub), batch);
            if let Some(c) = self.conns.get_mut(&conn) {
                c.max_delivered = c.max_delivered.max(cv);
            }
        }
    }

    /// Emit a subscriber's seq-0 snapshot from its footprint's cached fold and mark it
    /// live. No-op if the footprint is not yet hydrated (the subscriber waits) or the
    /// subscriber already baselined. Mirrors the single-thread server's
    /// "subscribe → hello, snapshot, progress": the snapshot always gets an initial frame.
    fn baseline_subscriber(&mut self, sub: SubId) {
        let Some(s) = self.subs.get(&sub) else {
            return;
        };
        if s.baselined {
            return;
        }
        let Some(fp) = self
            .queries
            .get(&s.query_id)
            .and_then(|reg| reg.footprint(s.binding.as_ref()))
        else {
            return;
        };
        if !fp.hydrated {
            return;
        }
        let batch = NormalizedBatch {
            epoch: s.epoch,
            seq: 0,
            cv: fp.footprint_cv,
            normalized_fp: fp.normalized_fp,
            // Project the cached full-row footprint to this query's synced columns (§5.2).
            ops: project_ops(fp.fold.footprint_snapshot(), &fp.proj),
        };
        let (conn, cv) = (s.conn, fp.footprint_cv);
        let s = self.subs.get_mut(&sub).expect("present");
        s.baselined = true;
        s.next_seq = 1;
        self.sink.batch(conn, QueryId(sub), batch);
        if let Some(c) = self.conns.get_mut(&conn) {
            c.max_delivered = c.max_delivered.max(cv);
        }
        self.emit_frame(conn, true);
    }

    fn on_progressed(&mut self, worker: usize, tx_id: u64) {
        let slot = self.worker_pos.entry(worker).or_insert(0);
        if tx_id <= *slot {
            return;
        }
        *slot = tx_id;
        self.frontier = self.frontier.max(tx_id);
        // Every connection with a subscription on this worker may have advanced its cv_min.
        let conns = self.conns_on_worker(worker);
        for conn in conns {
            self.emit_frame(conn, false);
        }
    }

    fn on_faulted(&mut self, query_id: QueryId, reason: String, cause: crate::FaultCause) {
        // Terminal teardown of one engine query: drop it + every subscriber on it (each
        // re-subscribes), so it stops pinning cv_min.
        self.pending_events.remove(&query_id);
        let Some(reg) = self.queries.remove(&query_id) else {
            return;
        };
        // Signal the query owner FIRST (before notifying subscribers), so any re-registration
        // it triggers is enqueued ahead of the re-subscribes the subscriber faults provoke —
        // the re-registered query is then present when those re-subscribes attach. The cause
        // rides along so the owner's mode machine can consult it BEFORE deciding to
        // re-register (FOLLOWER-LAG-SHED §6.2 item 5 / §6.6 item 4).
        self.sink.query_faulted(query_id, cause);
        for sub in reg.all_subscribers() {
            if let Some(s) = self.subs.remove(&sub) {
                if let Some(c) = self.conns.get_mut(&s.conn) {
                    c.subs.remove(&sub);
                }
                self.sink.faulted(s.conn, QueryId(sub), reason.clone());
                // A pinned-back cv_min may now be free to advance for the connection.
                self.emit_frame(s.conn, false);
            }
        }
    }

    /// Fault + detach every subscriber on a query without removing the query (the caller
    /// removes it next). Unlike [`on_faulted`](Self::on_faulted) this does NOT signal the
    /// query owner — the query is being deliberately revoked, not recovered.
    fn fault_subscribers(&mut self, query_id: QueryId, reason: &str) {
        let subs: Vec<SubId> = match self.queries.get_mut(&query_id) {
            Some(reg) => reg.take_all_subscribers(),
            None => return,
        };
        self.fault_subs(subs, reason);
    }

    /// Fault + detach the given (already-taken) subscribers.
    fn fault_subs(&mut self, subs: Vec<SubId>, reason: &str) {
        for sub in subs {
            if let Some(s) = self.subs.remove(&sub) {
                if let Some(c) = self.conns.get_mut(&s.conn) {
                    c.subs.remove(&sub);
                }
                self.sink.faulted(s.conn, QueryId(sub), reason.to_string());
                self.emit_frame(s.conn, false);
            }
        }
    }

    // --- control handling -------------------------------------------------

    fn on_control(&mut self, c: DrainControl) {
        match c {
            DrainControl::RegisterQuery {
                query_id,
                worker,
                fold,
                normalized_fp,
                proj,
                hydrated_cv,
            } => {
                self.queries.insert(
                    query_id,
                    QueryReg {
                        worker,
                        hydrated_cv,
                        kind: QueryKind::Singleton(Footprint::new(
                            *fold,
                            normalized_fp,
                            proj,
                            hydrated_cv,
                        )),
                    },
                );
                self.replay_pending(query_id);
            }
            DrainControl::RegisterFamily {
                query_id,
                worker,
                param_cols,
                hydrated_cv,
            } => {
                self.queries.insert(
                    query_id,
                    QueryReg {
                        worker,
                        hydrated_cv,
                        kind: QueryKind::Family {
                            param_cols,
                            partitions: HashMap::new(),
                            pending: HashMap::new(),
                        },
                    },
                );
                // A `PartitionHydrated` that beat the register lands in the per-binding
                // pending map through the ordinary route.
                self.replay_pending(query_id);
            }
            DrainControl::BindPartition {
                query_id,
                binding,
                fold,
                normalized_fp,
                proj,
                hydrated_cv,
            } => {
                let buffered = {
                    let Some(reg) = self.queries.get_mut(&query_id) else {
                        return;
                    };
                    let QueryKind::Family {
                        partitions,
                        pending,
                        ..
                    } = &mut reg.kind
                    else {
                        return; // a bind on a singleton — ignore
                    };
                    partitions.insert(
                        binding.clone(),
                        Footprint::new(*fold, normalized_fp, proj, hydrated_cv),
                    );
                    pending.remove(&binding).unwrap_or_default()
                };
                // Replay the partition's own hydrate (and any deltas behind it) in order.
                for u in buffered {
                    self.on_update(query_id, u);
                }
            }
            DrainControl::UnbindPartition { query_id, binding } => {
                let subs = {
                    let Some(reg) = self.queries.get_mut(&query_id) else {
                        return;
                    };
                    let QueryKind::Family {
                        partitions,
                        pending,
                        ..
                    } = &mut reg.kind
                    else {
                        return;
                    };
                    pending.remove(&binding);
                    match partitions.remove(&binding) {
                        Some(fp) => fp.subscribers.into_iter().collect::<Vec<_>>(),
                        None => return,
                    }
                };
                self.fault_subs(subs, "partition unbound");
            }
            DrainControl::AttachSubscriber {
                query_id,
                sub,
                conn,
                epoch,
                binding,
            } => {
                let Some(reg) = self.queries.get_mut(&query_id) else {
                    return; // attach to an unknown query — ignore
                };
                let Some(fp) = reg.footprint_mut(binding.as_ref()) else {
                    return; // attach to an unbound partition — ignore
                };
                fp.subscribers.insert(sub);
                self.subs.insert(
                    sub,
                    SubReg {
                        query_id,
                        binding,
                        conn,
                        epoch,
                        next_seq: 0,
                        baselined: false,
                    },
                );
                if let Some(c) = self.conns.get_mut(&conn) {
                    c.subs.insert(sub);
                }
                // Already hydrated ⇒ snapshot off the cached footprint now; else it lands
                // when the engine query / partition hydrates.
                self.baseline_subscriber(sub);
            }
            DrainControl::DetachSubscriber { sub } => {
                self.drop_subscriber(sub);
            }
            DrainControl::FaultSubscribers { query_id, reason } => {
                self.fault_subscribers(query_id, &reason);
            }
            DrainControl::RemoveQuery { query_id } => {
                self.pending_events.remove(&query_id);
                if let Some(reg) = self.queries.remove(&query_id) {
                    for sub in reg.all_subscribers() {
                        if let Some(s) = self.subs.remove(&sub) {
                            if let Some(c) = self.conns.get_mut(&s.conn) {
                                c.subs.remove(&sub);
                            }
                        }
                    }
                }
            }
            DrainControl::Connect { conn } => {
                self.conns.insert(
                    conn,
                    ConnReg {
                        subs: BTreeSet::new(),
                        released_cv: 0,
                        max_delivered: 0,
                    },
                );
            }
            DrainControl::Disconnect { conn } => {
                // Drop the connection's subscriptions (its shared engine queries persist —
                // the manager owns their lifecycle).
                if let Some(c) = self.conns.remove(&conn) {
                    for sub in c.subs {
                        if let Some(s) = self.subs.remove(&sub) {
                            if let Some(reg) = self.queries.get_mut(&s.query_id) {
                                reg.remove_subscriber(sub, s.binding.as_ref());
                            }
                        }
                    }
                }
            }
            DrainControl::Shutdown => {}
        }
    }

    /// Replay any events (a `Hydrated` / `PartitionHydrated`) that beat a register.
    fn replay_pending(&mut self, query_id: QueryId) {
        if let Some(buffered) = self.pending_events.remove(&query_id) {
            for u in buffered {
                self.on_update(query_id, u);
            }
        }
    }

    /// Detach one subscriber from its (persisting) engine query and connection.
    fn drop_subscriber(&mut self, sub: SubId) {
        if let Some(s) = self.subs.remove(&sub) {
            if let Some(reg) = self.queries.get_mut(&s.query_id) {
                reg.remove_subscriber(sub, s.binding.as_ref());
            }
            if let Some(c) = self.conns.get_mut(&s.conn) {
                c.subs.remove(&sub);
            }
        }
    }

    // --- the poke decision ------------------------------------------------

    /// Recompute `conn`'s `cv_min` and emit a progress frame iff the poke rule fires:
    /// there is undelivered-but-now-releasable data. `force` (the initial snapshot
    /// frame) always emits. Mutation confirmation needs no extra trigger — a client's
    /// own lmid advance is a data frame on its system query, so it IS releasable data.
    fn emit_frame(&mut self, conn: ConnId, force: bool) {
        let Some(c) = self.conns.get(&conn) else {
            return;
        };
        let cv_min = self.conn_cv_min(c);
        let data_to_release = c.max_delivered > c.released_cv && cv_min > c.released_cv;

        if !(force || data_to_release) {
            return;
        }
        let c = self.conns.get_mut(&conn).expect("present");
        c.released_cv = c.released_cv.max(cv_min);
        self.sink.progress(conn, ProgressFrame { cv_min });
    }

    /// The commit version every one of `conn`'s live subscriptions is known-current through —
    /// the min over them of `max(hydrated_cv, host-worker position)`; the frontier for a
    /// subscription-less connection.
    fn conn_cv_min(&self, c: &ConnReg) -> u64 {
        if c.subs.is_empty() {
            return self.frontier;
        }
        c.subs
            .iter()
            .map(|s| self.sub_cv(*s))
            .min()
            .unwrap_or(self.frontier)
    }

    fn sub_cv(&self, sub: SubId) -> u64 {
        let Some(s) = self.subs.get(&sub) else {
            return self.frontier;
        };
        match self.queries.get(&s.query_id) {
            Some(reg) => {
                // A partition hydrated later than its family registered carries its own
                // (higher) floor.
                let floor = reg
                    .footprint(s.binding.as_ref())
                    .map(|f| f.hydrated_cv)
                    .unwrap_or(reg.hydrated_cv);
                floor.max(self.worker_pos.get(&reg.worker).copied().unwrap_or(0))
            }
            None => self.frontier,
        }
    }

    fn conns_on_worker(&self, worker: usize) -> Vec<ConnId> {
        let mut set: BTreeSet<ConnId> = BTreeSet::new();
        for s in self.subs.values() {
            if self
                .queries
                .get(&s.query_id)
                .is_some_and(|reg| reg.worker == worker)
            {
                set.insert(s.conn);
            }
        }
        set.into_iter().collect()
    }
}

/// Distinct partitions a batch may span before the demux stops scanning its buckets
/// linearly and builds a hash index over them. Measured, not guessed (follow-up 310 B3, on
/// a 4-core Xeon @ 2.80 GHz, two parameter columns, every change a distinct partition):
/// bucketing 32 changes costs 3.3 µs linearly against 8.4 µs through a `HashMap`, so the
/// scan wins outright at the width a real commit has — but 512 costs 345 µs against
/// 117 µs, 2 048 costs 5.2 ms against 0.49 ms, and 8 192 costs 80 ms against 2.7 ms. The
/// crossover sits near 100 partitions; 64 keeps the common case byte-for-byte the scan it
/// always was and caps the quadratic before it can matter.
const LINEAR_DEMUX_BUCKETS: usize = 64;

/// Demux one `Changed` batch by the root row's partition key (impl plan D8): every change
/// is routable by a column read, buckets keep the batch's arrival order, and the changes
/// within a bucket keep theirs.
///
/// Batch width is a property of the WRITE, not of the family: one commit updating many
/// users' rows — the `zero-hot` bulk-update shape — lands as one batch spanning as many
/// partitions. A plain scan over the buckets seen so far is O(partitions²) in that width,
/// which is why it hands off to a hash index past [`LINEAR_DEMUX_BUCKETS`]; the index maps
/// a binding to its slot in `buckets`, so first-seen order survives the switch.
fn demux_by_partition(
    changes: Vec<ChangeEvent>,
    param_cols: &[ColId],
) -> Vec<(Binding, Vec<ChangeEvent>)> {
    let mut buckets: Vec<(Binding, Vec<ChangeEvent>)> = Vec::new();
    // Built only once the scan stops being the cheaper lookup; `None` is the scan.
    let mut index: Option<HashMap<Binding, usize>> = None;
    for c in changes {
        let key = canonical_key(c.root_row(), param_cols);
        let slot = match &index {
            Some(ix) => ix.get(&key).copied(),
            None => buckets.iter().position(|(b, _)| *b == key),
        };
        match slot {
            Some(i) => buckets[i].1.push(c),
            None => {
                let i = buckets.len();
                buckets.push((key.clone(), vec![c]));
                match &mut index {
                    Some(ix) => {
                        ix.insert(key, i);
                    }
                    None if buckets.len() > LINEAR_DEMUX_BUCKETS => {
                        index = Some(
                            buckets
                                .iter()
                                .enumerate()
                                .map(|(i, (b, _))| (b.clone(), i))
                                .collect(),
                        );
                    }
                    None => {}
                }
            }
        }
    }
    buckets
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::TxId;
    use rindle::value::{owned_row, OwnedValue};
    use rindle::{CaughtChange, CaughtNode};
    use std::sync::{Arc, Mutex};

    use crate::normalize_protocol::TableWireSchema;

    // NOTE (regression context): the drain used to carry per-commit lmid/rejected
    // metadata on the control channel, racing the worker event stream into the one
    // queue — a `Progressed(N)` that beat `CommitMeta(N)` emitted a frame releasing
    // commit N's data with a STALE lmid, making the client re-invoke its own
    // already-confirmed mutation (the dup-ADD `debug_assert`,
    // `CLUSTER-FOLD-IN-FINDINGS.md`). That class of race is now structurally
    // impossible: lmid is a row delivered through the client's own system query, so
    // it shares the data stream's ordering and the frame carries only `cv_min`.

    /// A collector usable as the drain sink — DrainCore owns it (boxed), so it shares its
    /// state through an `Arc<Mutex<_>>` the test also holds.
    #[derive(Clone, Default)]
    struct Out {
        batches: Arc<Mutex<Vec<(ConnId, u64, NormalizedBatch)>>>,
        frames: Arc<Mutex<Vec<(ConnId, ProgressFrame)>>>,
        faults: Arc<Mutex<Vec<(ConnId, u64)>>>,
        query_faults: Arc<Mutex<Vec<u64>>>,
    }
    impl DrainSink for Out {
        fn batch(&mut self, conn: ConnId, q: QueryId, b: NormalizedBatch) {
            self.batches.lock().unwrap().push((conn, q.0, b));
        }
        fn progress(&mut self, conn: ConnId, f: ProgressFrame) {
            self.frames.lock().unwrap().push((conn, f));
        }
        fn faulted(&mut self, conn: ConnId, q: QueryId, _r: String) {
            self.faults.lock().unwrap().push((conn, q.0));
        }
        fn query_faulted(&mut self, q: QueryId, _cause: crate::FaultCause) {
            self.query_faults.lock().unwrap().push(q.0);
        }
    }

    use crate::normalize::NormalizeFold;
    use crate::normalize_protocol::build_query_parts;

    fn query_parts() -> (NormalizeFold, u64) {
        let ast = rindle::table("t").build();
        let schemas = vec![TableWireSchema {
            name: "t".into(),
            columns: vec!["id".into(), "val".into()],
            primary_key: vec![0],
        }];
        let (fold, _tables, fp, _proj) = build_query_parts(&ast, schemas);
        (fold, fp)
    }

    /// Register engine query `q` and (by default) attach a single subscriber with sub id
    /// `== q` on `conn` — the 1:1 shape the old `install` helper drove.
    fn register(core: &mut DrainCore, q: u64, worker: usize, cv: u64) {
        let (fold, fp) = query_parts();
        core.on_control(DrainControl::RegisterQuery {
            query_id: QueryId(q),
            worker,
            fold: Box::new(fold),
            normalized_fp: fp,
            proj: ProjMap::new(),
            hydrated_cv: cv,
        });
    }

    fn attach(core: &mut DrainCore, q: u64, sub: u64, conn: ConnId, epoch: u64) {
        core.on_control(DrainControl::AttachSubscriber {
            query_id: QueryId(q),
            sub,
            conn,
            epoch,
            binding: None,
        });
    }

    fn add(id: i64) -> CaughtChange {
        add_val(id, "x")
    }

    fn add_val(id: i64, val: &str) -> CaughtChange {
        CaughtChange::Add(CaughtNode {
            row: owned_row(vec![OwnedValue::Int(id), OwnedValue::str(val)]),
            relationships: Default::default(),
        })
    }

    // --- parameterized query families (design 310 §5.2 / impl plan D8) ------------

    fn sb(val: &str) -> Binding {
        vec![rindle::canon::CanonVal::Str(val.into())]
    }

    /// Register family `q` partitioned on column 1 (`val`).
    fn register_family(core: &mut DrainCore, q: u64, worker: usize, cv: u64) {
        core.on_control(DrainControl::RegisterFamily {
            query_id: QueryId(q),
            worker,
            param_cols: vec![1],
            hydrated_cv: cv,
        });
    }

    fn bind(core: &mut DrainCore, q: u64, val: &str, cv: u64) {
        let (fold, fp) = query_parts();
        core.on_control(DrainControl::BindPartition {
            query_id: QueryId(q),
            binding: sb(val),
            fold: Box::new(fold),
            normalized_fp: fp,
            proj: ProjMap::new(),
            hydrated_cv: cv,
        });
    }

    fn attach_partition(core: &mut DrainCore, q: u64, val: &str, sub: u64, conn: ConnId) {
        core.on_control(DrainControl::AttachSubscriber {
            query_id: QueryId(q),
            sub,
            conn,
            epoch: 1,
            binding: Some(sb(val)),
        });
    }

    fn partition_hydrated(
        core: &mut DrainCore,
        q: u64,
        val: &str,
        cv: u64,
        rows: Vec<(i64, &str)>,
    ) {
        core.on_event(ClusterEvent::Update {
            query_id: QueryId(q),
            update: Update::PartitionHydrated {
                tx_id: TxId(cv),
                binding: sb(val),
                changes: rows.into_iter().map(|(id, v)| add_val(id, v)).collect(),
            },
        });
    }

    fn changed_rows(core: &mut DrainCore, q: u64, tx: u64, rows: Vec<(i64, &str)>) {
        core.on_event(ClusterEvent::Update {
            query_id: QueryId(q),
            update: Update::Changed {
                tx_id: TxId(tx),
                changes: rows.into_iter().map(|(id, v)| add_val(id, v)).collect(),
            },
        });
    }

    #[test]
    fn family_changed_is_demuxed_per_partition() {
        let out = Out::default();
        let mut core = DrainCore::new(Box::new(out.clone()), 0);
        core.on_control(DrainControl::Connect { conn: 7 });
        register_family(&mut core, 1, 0, 0);
        bind(&mut core, 1, "x", 0);
        bind(&mut core, 1, "y", 0);
        attach_partition(&mut core, 1, "x", 10, 7);
        attach_partition(&mut core, 1, "y", 11, 7);
        partition_hydrated(&mut core, 1, "x", 0, vec![]);
        partition_hydrated(&mut core, 1, "y", 0, vec![]);
        // Both baselined (seq 0 snapshots).
        assert_eq!(batches_for(&out, 10), vec![(1, 0, 0)]);
        assert_eq!(batches_for(&out, 11), vec![(1, 0, 0)]);
        // One family `Changed` carrying three partitions' rows: x gets two, y one, and
        // the unbound `z` row reaches nobody.
        changed_rows(
            &mut core,
            1,
            5,
            vec![(1, "x"), (2, "y"), (3, "x"), (4, "z")],
        );
        assert_eq!(batches_for(&out, 10), vec![(1, 0, 0), (1, 1, 2)]);
        assert_eq!(batches_for(&out, 11), vec![(1, 0, 0), (1, 1, 1)]);
        // A commit touching only y leaves x's seq untouched (gap-free per partition).
        changed_rows(&mut core, 1, 6, vec![(5, "y")]);
        assert_eq!(batches_for(&out, 10), vec![(1, 0, 0), (1, 1, 2)]);
        assert_eq!(batches_for(&out, 11), vec![(1, 0, 0), (1, 1, 1), (1, 2, 1)]);
    }

    /// A batch's width is a property of the WRITE, not of the family: one bulk update
    /// lands as one `Changed` spanning as many partitions as it touched. Past
    /// `LINEAR_DEMUX_BUCKETS` the demux stops scanning its buckets and builds a hash index
    /// over them (follow-up 310 B3), so what has to hold across that handoff is that the
    /// output is identical either way: one bucket per partition, buckets in first-seen
    /// order, and each bucket's changes in arrival order.
    #[test]
    fn a_wide_batch_demuxes_exactly_as_a_narrow_one_does() {
        let id_of = |c: &ChangeEvent| match c.root_row().get(0) {
            Some(rindle::value::Value::Int(i)) => i,
            other => panic!("expected an Int id, got {other:?}"),
        };
        for n in [
            4,
            LINEAR_DEMUX_BUCKETS,
            LINEAR_DEMUX_BUCKETS + 1,
            4 * LINEAR_DEMUX_BUCKETS,
        ] {
            // Two interleaved passes over the same n partitions, so both the bucket order
            // and the within-bucket order are observable: partition i sees ids i and n+i.
            let changes: Vec<ChangeEvent> = (0..2 * n)
                .map(|j| add_val(j as i64, &format!("p{}", j % n)))
                .collect();
            let got = demux_by_partition(changes, &[1]);
            assert_eq!(got.len(), n, "one bucket per partition at n={n}");
            for (i, (binding, bucket)) in got.iter().enumerate() {
                assert_eq!(*binding, sb(&format!("p{i}")), "first-seen order at n={n}");
                let ids: Vec<i64> = bucket.iter().map(id_of).collect();
                assert_eq!(
                    ids,
                    vec![i as i64, (n + i) as i64],
                    "arrival order at n={n}"
                );
            }
        }
    }

    #[test]
    fn partition_hydrate_racing_its_bind_is_replayed_in_order() {
        let out = Out::default();
        let mut core = DrainCore::new(Box::new(out.clone()), 0);
        core.on_control(DrainControl::Connect { conn: 7 });
        register_family(&mut core, 1, 0, 0);
        // The worker's hydrate and a following delta beat the coordinator's control.
        partition_hydrated(&mut core, 1, "x", 3, vec![(1, "x")]);
        changed_rows(&mut core, 1, 4, vec![(2, "x")]);
        bind(&mut core, 1, "x", 3);
        // A late subscriber baselines from the footprint that folded BOTH.
        attach_partition(&mut core, 1, "x", 10, 7);
        let b = out.batches.lock().unwrap();
        let (_, sub, batch) = &b[0];
        assert_eq!(*sub, 10);
        assert_eq!(batch.seq, 0);
        assert_eq!(batch.ops.len(), 2, "hydrate + the replayed delta");
        assert_eq!(batch.cv, 4);
    }

    #[test]
    fn unbind_partition_faults_only_its_subscribers_and_drops_its_deltas() {
        let out = Out::default();
        let mut core = DrainCore::new(Box::new(out.clone()), 0);
        core.on_control(DrainControl::Connect { conn: 7 });
        register_family(&mut core, 1, 0, 0);
        bind(&mut core, 1, "x", 0);
        bind(&mut core, 1, "y", 0);
        attach_partition(&mut core, 1, "x", 10, 7);
        attach_partition(&mut core, 1, "y", 11, 7);
        partition_hydrated(&mut core, 1, "x", 0, vec![]);
        partition_hydrated(&mut core, 1, "y", 0, vec![]);
        core.on_control(DrainControl::UnbindPartition {
            query_id: QueryId(1),
            binding: sb("x"),
        });
        assert_eq!(*out.faults.lock().unwrap(), vec![(7, 10)]);
        // A late delta for x (the unbind race) is dropped; y keeps flowing.
        changed_rows(&mut core, 1, 5, vec![(1, "x"), (2, "y")]);
        assert_eq!(batches_for(&out, 10), vec![(1, 0, 0)]);
        assert_eq!(batches_for(&out, 11), vec![(1, 0, 0), (1, 1, 1)]);
        // An attach to the unbound partition is ignored.
        attach_partition(&mut core, 1, "x", 12, 7);
        assert!(batches_for(&out, 12).is_empty());
    }

    #[test]
    fn family_fault_faults_every_partition_subscriber() {
        let out = Out::default();
        let mut core = DrainCore::new(Box::new(out.clone()), 0);
        core.on_control(DrainControl::Connect { conn: 7 });
        register_family(&mut core, 1, 0, 0);
        bind(&mut core, 1, "x", 0);
        bind(&mut core, 1, "y", 0);
        attach_partition(&mut core, 1, "x", 10, 7);
        attach_partition(&mut core, 1, "y", 11, 7);
        core.on_event(ClusterEvent::Faulted {
            query_id: QueryId(1),
            reason: "boom".into(),
            cause: crate::FaultCause::Derive,
        });
        let mut faults = out.faults.lock().unwrap().clone();
        faults.sort();
        assert_eq!(faults, vec![(7, 10), (7, 11)]);
        assert_eq!(*out.query_faults.lock().unwrap(), vec![1]);
    }

    /// Register query `q` on `worker`, attach subscriber `q` to `conn`, then hydrate (empty).
    fn install(core: &mut DrainCore, q: u64, conn: ConnId, worker: usize, cv: u64) {
        register(core, q, worker, cv);
        attach(core, q, q, conn, 1);
        core.on_event(ClusterEvent::Update {
            query_id: QueryId(q),
            update: Update::Hydrated {
                tx_id: TxId(cv),
                changes: vec![],
            },
        });
    }

    fn changed(core: &mut DrainCore, q: u64, tx: u64, id: i64) {
        core.on_event(ClusterEvent::Update {
            query_id: QueryId(q),
            update: Update::Changed {
                tx_id: TxId(tx),
                changes: vec![add(id)],
            },
        });
    }

    /// The batches a routing id (sub id) received, as `(epoch, seq, op-count)` in order.
    fn batches_for(out: &Out, sub: u64) -> Vec<(u64, u64, usize)> {
        out.batches
            .lock()
            .unwrap()
            .iter()
            .filter(|(_, q, _)| *q == sub)
            .map(|(_, _, b)| (b.epoch, b.seq, b.ops.len()))
            .collect()
    }

    /// The frames a connection received (cv_min values, in order).
    fn cvs(out: &Out, conn: ConnId) -> Vec<u64> {
        out.frames
            .lock()
            .unwrap()
            .iter()
            .filter(|(c, _)| *c == conn)
            .map(|(_, f)| f.cv_min)
            .collect()
    }

    /// A slow worker holds back only its own clients: advancing worker 0 releases a
    /// connection whose query lives on worker 0, while a connection on worker 1 (not yet
    /// progressed) stays pinned — the read-side isolation win (§8).
    #[test]
    fn per_worker_position_isolates_clients() {
        let out = Out::default();
        let mut core = DrainCore::new(Box::new(out.clone()), 0);
        core.on_control(DrainControl::Connect { conn: 1 });
        core.on_control(DrainControl::Connect { conn: 2 });
        install(&mut core, 1, 1, 0, 0); // Q1 on worker 0, conn 1
        install(&mut core, 2, 2, 1, 0); // Q2 on worker 1, conn 2

        // A commit at tx 1 changes both queries → both get a data frame at cv 1 (buffered).
        changed(&mut core, 1, 1, 10);
        changed(&mut core, 2, 1, 20);
        assert_eq!(
            cvs(&out, 1),
            vec![0],
            "no release before worker 0 progresses"
        );
        assert_eq!(
            cvs(&out, 2),
            vec![0],
            "no release before worker 1 progresses"
        );

        // Worker 0 reports tx 1 → ONLY conn 1 is released to cv_min 1; conn 2 stays pinned.
        core.on_event(ClusterEvent::Progressed {
            worker: 0,
            tx_id: TxId(1),
        });
        assert_eq!(cvs(&out, 1), vec![0, 1], "conn 1 released by its worker");
        assert_eq!(
            cvs(&out, 2),
            vec![0],
            "conn 2 still held back by its own (slow) worker 1"
        );

        // Worker 1 catches up → conn 2 releases too.
        core.on_event(ClusterEvent::Progressed {
            worker: 1,
            tx_id: TxId(1),
        });
        assert_eq!(
            cvs(&out, 2),
            vec![0, 1],
            "conn 2 released when worker 1 caught up"
        );
    }

    /// The poke rule: a foreign write that doesn't touch a connection's data produces no
    /// progress frame for it, even as its worker's position (and cv_min) advances.
    #[test]
    fn idle_connection_not_poked_by_foreign_write() {
        let out = Out::default();
        let mut core = DrainCore::new(Box::new(out.clone()), 0);
        core.on_control(DrainControl::Connect { conn: 1 });
        install(&mut core, 1, 1, 0, 0); // Q1 on worker 0
        assert_eq!(cvs(&out, 1), vec![0], "the initial snapshot frame");

        // A foreign write at tx 5 advances worker 0 but never delivered Q1 a data frame.
        core.on_event(ClusterEvent::Progressed {
            worker: 0,
            tx_id: TxId(5),
        });
        assert_eq!(
            cvs(&out, 1),
            vec![0],
            "idle connection sees no extra frame for an irrelevant write"
        );
    }

    /// An lmid-as-data confirmation: the client's own system-query batch (its lmid row)
    /// is releasable data like any other, so a "mutation-only" commit still pokes the
    /// mutating connection — via the normal data path, with no metadata side-channel.
    #[test]
    fn lmid_query_batch_releases_like_any_data() {
        let out = Out::default();
        let mut core = DrainCore::new(Box::new(out.clone()), 0);
        core.on_control(DrainControl::Connect { conn: 1 });
        core.on_control(DrainControl::Connect { conn: 2 });
        install(&mut core, 1, 1, 0, 0); // conn 1's lmid system query, worker 0
        install(&mut core, 2, 2, 1, 0); // conn 2's data query, worker 1
        let before2 = cvs(&out, 2).len();

        // Commit tx 1 = conn 1's mutation whose only derived change is its lmid row.
        changed(&mut core, 1, 1, 10); // the lmid-query batch (data frame, cv 1)
        core.on_event(ClusterEvent::Progressed {
            worker: 0,
            tx_id: TxId(1),
        });
        core.on_event(ClusterEvent::Progressed {
            worker: 1,
            tx_id: TxId(1),
        });

        assert_eq!(
            cvs(&out, 1),
            vec![0, 1],
            "the mutating connection's lmid batch released at cv 1"
        );
        assert_eq!(
            cvs(&out, 2).len(),
            before2,
            "the other connection saw no frame (no data delivered to it)"
        );
    }

    /// Dedup fan-out: ONE engine query feeds two subscribers on different connections. The
    /// shared footprint folds a commit once, but each subscriber gets its own framed batch
    /// (its own epoch, its own seq cursor). There is only one engine query id in play.
    #[test]
    fn shared_query_fans_one_engine_query_to_many_subscribers() {
        let out = Out::default();
        let mut core = DrainCore::new(Box::new(out.clone()), 0);
        core.on_control(DrainControl::Connect { conn: 1 });
        core.on_control(DrainControl::Connect { conn: 2 });
        // One engine query (eqid 100) on worker 0; two subscribers with distinct epochs.
        register(&mut core, 100, 0, 0);
        attach(&mut core, 100, 10, 1, 7);
        attach(&mut core, 100, 20, 2, 9);
        core.on_event(ClusterEvent::Update {
            query_id: QueryId(100),
            update: Update::Hydrated {
                tx_id: TxId(0),
                changes: vec![],
            },
        });
        // Each subscriber got its own seq-0 snapshot stamped with its own epoch.
        assert_eq!(batches_for(&out, 10), vec![(7, 0, 0)]);
        assert_eq!(batches_for(&out, 20), vec![(9, 0, 0)]);

        // A commit folds ONCE but fans to both, each at seq 1 under its own epoch.
        changed(&mut core, 100, 1, 42);
        assert_eq!(batches_for(&out, 10), vec![(7, 0, 0), (7, 1, 1)]);
        assert_eq!(batches_for(&out, 20), vec![(9, 0, 0), (9, 1, 1)]);

        // Detaching one leaves the other (and the shared engine query) live.
        core.on_control(DrainControl::DetachSubscriber { sub: 10 });
        changed(&mut core, 100, 2, 43);
        assert_eq!(
            batches_for(&out, 10),
            vec![(7, 0, 0), (7, 1, 1)],
            "detached subscriber gets no further batches"
        );
        assert_eq!(batches_for(&out, 20), vec![(9, 0, 0), (9, 1, 1), (9, 2, 1)]);
    }

    /// A late subscriber attaching to an already-hydrated shared query is baselined from the
    /// cached footprint — its seq-0 snapshot carries every live row, with NO second engine
    /// query / re-hydrate.
    #[test]
    fn late_subscriber_baselines_from_cached_footprint() {
        let out = Out::default();
        let mut core = DrainCore::new(Box::new(out.clone()), 0);
        core.on_control(DrainControl::Connect { conn: 1 });
        core.on_control(DrainControl::Connect { conn: 2 });
        register(&mut core, 100, 0, 0);
        attach(&mut core, 100, 10, 1, 1);
        // Hydrate with one row, then a commit adds a second — footprint now has 2 rows.
        core.on_event(ClusterEvent::Update {
            query_id: QueryId(100),
            update: Update::Hydrated {
                tx_id: TxId(0),
                changes: vec![add(1)],
            },
        });
        changed(&mut core, 100, 1, 2);
        assert_eq!(batches_for(&out, 10), vec![(1, 0, 1), (1, 1, 1)]);

        // A late subscriber joins after both commits — one seq-0 snapshot with BOTH rows.
        attach(&mut core, 100, 20, 2, 5);
        assert_eq!(
            batches_for(&out, 20),
            vec![(5, 0, 2)],
            "late subscriber snapshots the full 2-row footprint from cache"
        );

        // From here both share the same stream; the late one continues at seq 1.
        changed(&mut core, 100, 2, 3);
        assert_eq!(batches_for(&out, 10), vec![(1, 0, 1), (1, 1, 1), (1, 2, 1)]);
        assert_eq!(batches_for(&out, 20), vec![(5, 0, 2), (5, 1, 1)]);
    }

    /// A worker fault signals the QUERY OWNER first (so it can re-register), then faults
    /// every subscriber on it, then removes the engine query.
    #[test]
    fn fault_signals_owner_before_faulting_subscribers() {
        let out = Out::default();
        let mut core = DrainCore::new(Box::new(out.clone()), 0);
        core.on_control(DrainControl::Connect { conn: 1 });
        core.on_control(DrainControl::Connect { conn: 2 });
        register(&mut core, 100, 0, 0);
        attach(&mut core, 100, 10, 1, 1);
        attach(&mut core, 100, 20, 2, 1);
        core.on_event(ClusterEvent::Update {
            query_id: QueryId(100),
            update: Update::Hydrated {
                tx_id: TxId(0),
                changes: vec![],
            },
        });

        core.on_faulted(QueryId(100), "boom".into(), crate::FaultCause::Derive);
        assert_eq!(
            *out.query_faults.lock().unwrap(),
            vec![100],
            "owner signalled once with the ENGINE query id"
        );
        let mut faulted_subs: Vec<u64> =
            out.faults.lock().unwrap().iter().map(|(_, s)| *s).collect();
        faulted_subs.sort();
        assert_eq!(faulted_subs, vec![10, 20], "both subscribers faulted");
        assert!(
            !core.queries.contains_key(&QueryId(100)),
            "engine query removed"
        );
        assert!(core.subs.is_empty(), "subscribers dropped");
    }

    /// Revoking subscribers (the dematerialize-with-active-subscribers path) faults + detaches
    /// them WITHOUT signalling the owner (no re-registration) and leaves the query for the
    /// caller's following RemoveQuery.
    #[test]
    fn fault_subscribers_revokes_without_owner_signal() {
        let out = Out::default();
        let mut core = DrainCore::new(Box::new(out.clone()), 0);
        core.on_control(DrainControl::Connect { conn: 1 });
        register(&mut core, 100, 0, 0);
        attach(&mut core, 100, 10, 1, 1);
        core.on_event(ClusterEvent::Update {
            query_id: QueryId(100),
            update: Update::Hydrated {
                tx_id: TxId(0),
                changes: vec![],
            },
        });

        core.on_control(DrainControl::FaultSubscribers {
            query_id: QueryId(100),
            reason: "materialization removed".into(),
        });
        assert_eq!(
            out.faults.lock().unwrap().as_slice(),
            &[(1, 10)],
            "the subscriber was faulted"
        );
        assert!(
            out.query_faults.lock().unwrap().is_empty(),
            "owner NOT signalled — this is a revoke, not a recover"
        );
        assert!(
            core.queries.contains_key(&QueryId(100)),
            "query left in place for the caller's RemoveQuery"
        );
        assert!(core.subs.is_empty(), "subscriber detached");
    }

    /// Streaming fan-out (`CLUSTER-INCREMENTAL-FANOUT-DESIGN.md` §4): one transaction arrives as
    /// **several `Changed(N)` slices** (a chunked `TxPush` stream). The drain fans each out as
    /// its own `cv=N` data frame eagerly, but holds the release until the `Progressed(N)` commit
    /// marker — so the whole transaction becomes visible at once, never a partial slice.
    #[test]
    fn multi_slice_txn_buffers_data_frames_and_releases_together_on_marker() {
        let out = Out::default();
        let mut core = DrainCore::new(Box::new(out.clone()), 0);
        core.on_control(DrainControl::Connect { conn: 1 });
        install(&mut core, 1, 1, 0, 0); // Q1 on worker 0, conn 1, hydrated at cv 0
        assert_eq!(cvs(&out, 1), vec![0], "the initial snapshot frame");

        // Tx 1 streams in as TWO speculative slices (a >1-chunk fan-out), both stamped cv 1.
        changed(&mut core, 1, 1, 10);
        changed(&mut core, 1, 1, 11);
        assert_eq!(
            batches_for(&out, 1),
            vec![(1, 0, 0), (1, 1, 1), (1, 2, 1)],
            "seq-0 snapshot, then one data frame per slice (both at cv 1)"
        );
        assert_eq!(
            cvs(&out, 1),
            vec![0],
            "no progress frame yet — slices are buffered until the commit marker"
        );

        // The commit marker releases the whole tx (both slices) at once.
        core.on_event(ClusterEvent::Progressed {
            worker: 0,
            tx_id: TxId(1),
        });
        assert_eq!(
            cvs(&out, 1),
            vec![0, 1],
            "the marker releases cv 1 — both slices become visible together, atomically"
        );
    }

    /// Streaming fault, consumer side (`CLUSTER-INCREMENTAL-FANOUT-DESIGN.md` §4.1, abort≙rehydrate):
    /// a worker that dies mid-stream stops reporting `Progressed`, so a connection with a query on
    /// it is pinned at that worker's last position — even as a query on a HEALTHY worker advances
    /// (its data is buffered, held). The terminal `Faulted` for the dead worker's query (its
    /// re-subscribe signal) removes it, so the connection's `cv_min` jumps to the healthy worker's
    /// position and the held data releases. This is the read-side recovery the worker-side
    /// tear-down + re-hydrate pairs with.
    #[test]
    fn faulted_worker_unpins_connection_and_releases_held_data() {
        let out = Out::default();
        let mut core = DrainCore::new(Box::new(out.clone()), 0);
        core.on_control(DrainControl::Connect { conn: 1 });
        install(&mut core, 1, 1, 0, 0); // Q1 on worker 0, conn 1
        install(&mut core, 2, 1, 1, 0); // Q2 on worker 1, SAME conn 1
        assert_eq!(
            cvs(&out, 1),
            vec![0, 0],
            "an initial snapshot frame per install"
        );

        // Tx 3 touches both; both workers progress → conn releases to cv 3.
        changed(&mut core, 1, 3, 10);
        changed(&mut core, 2, 3, 20);
        core.on_event(ClusterEvent::Progressed {
            worker: 0,
            tx_id: TxId(3),
        });
        core.on_event(ClusterEvent::Progressed {
            worker: 1,
            tx_id: TxId(3),
        });
        assert_eq!(
            cvs(&out, 1),
            vec![0, 0, 3],
            "both workers at 3 → conn released to 3"
        );

        // Worker 1 now dies mid-stream. A later tx 7 touches ONLY Q1 (worker 0), and worker 0
        // progresses to 7 — but the connection stays pinned at 3 (Q2's dead worker never moved),
        // so Q1's tx-7 data is buffered, NOT released.
        changed(&mut core, 1, 7, 11);
        core.on_event(ClusterEvent::Progressed {
            worker: 0,
            tx_id: TxId(7),
        });
        assert_eq!(
            cvs(&out, 1),
            vec![0, 0, 3],
            "Q1's tx-7 data is held — the dead worker's Q2 pins cv_min at 3"
        );

        // Faulting Q2 (the dead worker's query) unpins the connection: cv_min jumps to worker 0's
        // position (7) and the held tx-7 data releases.
        core.on_faulted(
            QueryId(2),
            "worker 1 died mid-stream".into(),
            crate::FaultCause::WorkerLost,
        );
        assert_eq!(
            cvs(&out, 1),
            vec![0, 0, 3, 7],
            "the fault unpins cv_min → the held Q1 data releases at cv 7"
        );
        assert_eq!(
            out.faults.lock().unwrap().as_slice(),
            &[(1, 2)],
            "Q2's subscriber got the terminal fault (its re-subscribe signal)"
        );
    }
}