rindle_replica/

engine.rs

//! The per-thread IVM unit: ONE `Graph`, one `TableSource` per registered table
//! (shared by every query on this thread), and the registered queries (each a
//! change-stream sink). One push of a captured batch fans out to every dependent
//! query, so a commit needs just one derivation snapshot per thread — a plain read
//! transaction; batch state rides in each source's in-memory [`BatchDelta`] overlay
//! (design 306 batch-overlay derivation, the sole mode since S5 removed the
//! historical write-then-abort replay).

use std::cell::{Cell, RefCell};
use std::collections::{BTreeMap, HashMap};
use std::path::PathBuf;
use std::rc::Rc;
use std::sync::Arc;
use std::time::Instant;

use rindle::change::SourceChange;
use rindle::family::{BindingSet, FamilyPipeline};
use rindle::graph::{
    CountingSource, EmitCounter, Graph, JoinPrecheckState, NodeId, PipelineManifest,
};
use rindle::storage::StorageFactory;
use rindle::value::{values_identical, ColId, OwnedRow, Schema, Sort, SourceSchema};
use rindle::{
    build_family_pipeline, build_pipeline, has_scalar_subquery, resolve_scalars, view_schema, Ast,
    CaughtChange, RindleError, ScalarCatalog, ScalarSource,
};
use rindle_cdc::Captured;
use rindle_planner::{has_flippable_exists, plan_ast, CachingCostModel, ConnectionCostModel};
use rindle_sqlite::{
    BatchDelta, DatabaseStorage, DeltaBudget, GraphTableSourceExt, SqliteCostModel, TableSource,
};
use rindle_wire::family_key::Binding;
use rusqlite::Connection;

use crate::analyze::{
    advisories, collect_joins, AnalyzePlan, AnalyzeReport, AnalyzeSource, AnalyzeTiming,
    AnalyzeTotals,
};
use crate::parallel::open_journal;
use crate::schema::TableSchema;
use crate::JournalMode;
use crate::QueryId;
use crate::ReplicaError;

/// Per-registered-query bookkeeping in the shared graph: the [`PipelineManifest`]
/// (the exact node/storage slots its build created, for teardown) and the caller's
/// opaque [`QueryId`] tag. Keyed by the query's change-sink `NodeId`. The engine does
/// nothing with the tag (no de-dup / no refcount — that is the caller's concern); it is
/// retained for correlation/observability and echoed back via the `Query` handle.
/// What [`Engine::join_precheck_report`] returns: the join membership pre-check's bounds
/// and per-query join states on one engine (design 311 §8 inspection).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct JoinPrecheckReport {
    /// The bounds in force on the engine's graph.
    pub bounds: rindle::graph::JoinPrecheckBounds,
    /// Distinct parent keys charged against the per-graph budget right now.
    pub tracked_keys: usize,
    /// Per registered query (sorted by id): the state of each join its pipeline owns, in
    /// build order.
    pub queries: Vec<(QueryId, Vec<JoinPrecheckState>)>,
}

/// What [`Engine::storage_report`] returns: how much operator scratch state one
/// registered query's pipeline holds right now (design 310 impl plan D5's leak probe,
/// lifted to the cluster boundary — see
/// [`Cluster::__test_storage_entries`](crate::Cluster::__test_storage_entries)).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct StorageReport {
    /// `(key, value)` entries across every storage slot the pipeline owns.
    pub entries: usize,
    /// Every `(slot, key, value)`, when the caller asked for the dump; empty otherwise.
    /// `slot` indexes the pipeline manifest's storage list (a build-order operator name).
    pub dump: Vec<(usize, String, String)>,
}

struct QueryReg {
    manifest: PipelineManifest,
    /// The caller's opaque tag, echoed back via [`Engine::query_id_of`] when a
    /// derived delta is emitted (the engine still never *interprets* it).
    query_id: QueryId,
    /// `Some` for a parameterized query family (design 310): the pipeline handle the
    /// per-partition bind / unbind / hydrate entry points take.
    family: Option<FamilyPipeline>,
}

/// Where stateful operators (`take` / `cap` / `reduce`) keep their scratch state.
///
/// The graph's [`StorageFactory`] decides this once, at construction, for every
/// operator the engine's pipelines build. Both backends are equivalent in result —
/// the SQLite-backed `OpStorage` agrees with `MemoryStorage` op-for-op — so the
/// choice is purely an operational memory/throughput trade-off.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum OperatorStorage {
    /// In-process `BTreeMap` scratch state (the default). Fastest, but the working
    /// set is bounded by RAM.
    #[default]
    Memory,
    /// Spill operator state to a private, on-disk temporary SQLite database (one per
    /// engine, journal/sync off, exclusive locking — see
    /// [`DatabaseStorage::new_temp_file`]). Lets operator state grow past RAM at the
    /// cost of per-op SQLite calls; the scratch DB is deleted when the engine drops.
    SqliteSpill,
}

/// A hosted table resolved to what a push needs: the shared source's `NodeId` plus the
/// `Schema` whose primary key decides whether an `Edit` is pk-changing. The value half of
/// [`Engine::sources`], named so the borrowed forms that thread it through the push path
/// ([`Engine::push_resolved`] and its caller's memo) stay readable.
type HostedSource = (NodeId, SourceSchema);

/// The per-thread IVM unit (see the module docs). Internal to the `Db`/`Cluster`
/// openers, and public via the crate's **embedded-engine seam** (design 308): a host
/// that owns its own write path — the SQLite extension's commit hook — constructs one
/// over a derivation connection, registers tables/queries, and drives
/// [`apply_batch`](Engine::apply_batch) itself.
pub struct Engine {
    /// The single dataflow graph for this thread. `!Send`; never leaves the thread.
    graph: Graph,
    /// The connection the engine reads/derives against (its `TableSource`s hold it).
    /// During a commit it holds the derivation snapshot transaction, closed by
    /// [`rollback`](Engine::rollback).
    worker: Rc<Connection>,
    /// table name → (shared source `NodeId`, its `Schema`). The schema is what the
    /// pipeline builder's `resolve` closure hands back for each referenced table.
    sources: HashMap<String, HostedSource>,
    /// table name → a build-time **read** handle ([`TableSource::fork`] over the same
    /// worker connection + cached unique-index metadata) used by [`resolve_scalars`] to
    /// fold a `scalar` correlated subquery before planning. Separate from the live IVM
    /// source in the graph; it only does point reads / unique-key lookups.
    scalar_sources: HashMap<String, TableSource>,
    /// Every registered query, keyed by its change-sink `NodeId` (drained each commit).
    /// The generational sink id is unique per registration, so a torn-down-then-rebuilt
    /// query never collides with its former self.
    queries: HashMap<NodeId, QueryReg>,
    /// When set, [`register_query`](Self::register_query) runs the cost-based join-flip
    /// planner over the AST before lowering (a [`SqliteCostModel`] on `worker`, wrapped
    /// in [`CachingCostModel`]). Opt-in (default off) so the unplanned path stays the
    /// default; the plan is frozen at registration (never re-planned). Planning is
    /// result-preserving, so this only changes execution strategy, never the rows.
    plan_queries: bool,
    /// The operator-scratch-state backend this engine's graph was built with. Held so a
    /// post-fault rebuild ([`fault_recover`](crate::parallel)) re-creates the graph with
    /// the *same* backend rather than silently downgrading to memory.
    operator_storage: OperatorStorage,
    /// table name → the shared [`BatchDelta`] its source (and every fork) reads through.
    /// Recorded so `open_snapshot`/`rollback` can drive the lifecycle (plan D1).
    deltas: HashMap<String, Rc<BatchDelta>>,
    /// The D4 derivation memory budget, shared by EVERY table's delta so one
    /// transaction is bounded by this number and not by (tables × this number). Held on
    /// the engine — and installed per `register_table` — so a shed/fault rebuild carries
    /// the tuned ceiling instead of silently resetting to the default.
    ///
    /// One budget per engine means one per *thread*: a `Cluster`'s worker engines each
    /// hold their own, so a process ceiling is `workers × max_bytes`.
    delta_budget: Rc<DeltaBudget>,
}

impl Engine {
    pub fn new(
        worker: Rc<Connection>,
        plan_queries: bool,
        operator_storage: OperatorStorage,
    ) -> Result<Engine, ReplicaError> {
        let graph = build_graph(operator_storage)?;
        Ok(Engine {
            graph,
            worker,
            sources: HashMap::new(),
            scalar_sources: HashMap::new(),
            queries: HashMap::new(),
            plan_queries,
            operator_storage,
            deltas: HashMap::new(),
            delta_budget: Rc::new(DeltaBudget::default()),
        })
    }

    pub fn has_source(&self, table: &str) -> bool {
        self.sources.contains_key(table)
    }

    /// Build the shared `TableSource` for `table` on the worker connection, add it to
    /// the graph, and remember its `Schema`. Idempotent.
    pub fn register_table(&mut self, table: &str, ts: TableSchema) -> Result<(), ReplicaError> {
        if self.sources.contains_key(table) {
            return Ok(());
        }
        let columns = ts.columns;
        let primary_key = ts.primary_key;
        // The reported schema (relationships are derived query-locally by the builder,
        // so a relationship-free schema is correct here): names + PK + primary sort.
        let col_names: Vec<&str> = columns.iter().map(|c| &*c.name).collect();
        let sort: Sort = primary_key.iter().map(|&c| (c, true)).collect();
        let schema = SourceSchema::new(col_names, primary_key.clone(), sort)
            .with_column_types(columns.iter().map(|c| c.ty).collect());

        let source = TableSource::try_new(self.worker.clone(), table, columns, primary_key)?;
        // Record the source's batch delta (design 306) BEFORE forking, so the scalar
        // fork — and every later fork — shares it (plan D1: the delta models storage,
        // which forks share). `open_snapshot`/`rollback` drive its lifecycle.
        let delta = source.batch_delta();
        // Every table charges the SAME budget — the ceiling is the transaction's, not
        // each table's (design 306 D4).
        delta.set_budget(Rc::clone(&self.delta_budget));
        self.deltas.insert(table.to_string(), delta);
        // A read-only fork (same connection + cached unique-index metadata) for
        // build-time scalar resolution; the original drives the live IVM pipeline.
        self.scalar_sources.insert(table.to_string(), source.fork());
        let id = self.graph.add_table_source(source);
        self.sources.insert(table.to_string(), (id, schema));
        Ok(())
    }

    /// Lower `ast` into the shared graph, terminate it in a fresh change-stream sink,
    /// and hydrate it. `query_id` is the caller's opaque tag (the engine stores it for
    /// correlation but never interprets it — de-dup is the caller's concern). Returns
    /// the sink's `NodeId` and the initial change set (the hydration `Add`s).
    /// `BuildError` (unknown table/column, unsupported shape) is surfaced synchronously.
    pub fn register_query(
        &mut self,
        query_id: QueryId,
        ast: &Ast,
    ) -> Result<(NodeId, Vec<CaughtChange>), ReplicaError> {
        use std::panic::{catch_unwind, AssertUnwindSafe};

        // Record the exact node/storage slots this build creates so the pipeline can be
        // torn down precisely later (and so a *failed* build is cleaned up, not leaked).
        self.graph.begin_recording();

        // Build → wire the sink → hydrate under a panic boundary. The recording stays
        // active THROUGH hydrate (which adds no nodes), so on ANY unwind the still-`Some`
        // manifest captures everything built and we reclaim it below — no orphaned,
        // wired, un-tear-down-able pipeline is left fanning pushes forever. Build/engine
        // errors surface as `Err`; an internal-invariant *panic* (an engine bug, never
        // valid input — malformed ASTs are `BuildError`) is caught here. Mirrors the push
        // path's `source_push_isolated`; effective under `panic = "unwind"`
        // (`release-server`), a passthrough under the client `abort` profile.
        let outcome = {
            let graph = &mut self.graph;
            let sources = &self.sources;
            let scalar_sources = &self.scalar_sources;
            let worker = &self.worker;
            let plan_queries = self.plan_queries;
            catch_unwind(AssertUnwindSafe(move || {
                let resolve = |t: &str| sources.get(t).map(|(id, sc)| (*id, sc.clone()));

                // 1. Scalar-subquery resolution, BEFORE planning (design §8): fold a
                //    statically-unique `scalar` EXISTS into a literal and delete the
                //    join, so the planner enumerates over the already-simplified graph.
                //    Reads the build-time snapshot through the fork catalog; the inlined
                //    value is frozen for this pipeline (§3). Skips the clone when there
                //    is nothing flagged.
                let resolved;
                let ast = if has_scalar_subquery(ast) {
                    let catalog = ForkCatalog(scalar_sources);
                    resolved = resolve_scalars(ast, &catalog)?;
                    &resolved
                } else {
                    ast
                };

                // 2. Opt-in cost-based planning: annotate `flip` before lowering. A fresh
                // `CachingCostModel` per registration (stats may drift between queries;
                // the plan is frozen for this one). Runs INSIDE the panic boundary so a
                // cost-model assert (e.g. scanstatus) surfaces as the internal-panic
                // error rather than crashing the worker. Result-preserving by construction.
                let planned;
                let ast = if plan_queries && has_flippable_exists(ast) {
                    let model: Rc<dyn ConnectionCostModel> =
                        Rc::new(CachingCostModel::new(SqliteCostModel::new(worker.clone())));
                    planned = plan_ast(ast, model);
                    &planned
                } else {
                    // No flippable EXISTS (or planning off) ⇒ planning is a guaranteed
                    // no-op; lower the AST directly (skips the plan-graph build + clone).
                    ast
                };
                let top = build_pipeline(graph, ast, &resolve)?;
                let sink = graph.add_change_sink(top);
                graph.set_sink_edge(top, sink);
                // Fallible cold drain: registering a `sum` over data whose set total is
                // ALREADY outside i64 must fail typed here (§5.3 — a fresh SQLite
                // `SELECT sum(…)` over this state errors too), not return Ok with a
                // `NULL` aggregate while the armed overflow state fails every later
                // write on the shared graph. The `Err` arm below tears the partial
                // pipeline down, which also clears that state.
                let initial = graph.try_hydrate_change_sink(sink)?;
                Ok::<(NodeId, Vec<CaughtChange>), ReplicaError>((sink, initial))
            }))
        };

        match outcome {
            Ok(Ok((sink, initial))) => {
                let manifest = self.graph.take_recording();
                self.queries.insert(
                    sink,
                    QueryReg {
                        manifest,
                        query_id,
                        family: None,
                    },
                );
                Ok((sink, initial))
            }
            Ok(Err(e)) => {
                // A surfaced build/engine error: tear out the partial pipeline.
                let manifest = self.graph.take_recording();
                self.graph.destroy_pipeline(&manifest);
                Err(e)
            }
            Err(_panic) => {
                // An internal-invariant panic mid build/hydrate: the recording still
                // holds everything built before the unwind — reclaim it so no live,
                // undrained orphan is left wired to the shared sources.
                let manifest = self.graph.take_recording();
                self.graph.destroy_pipeline(&manifest);
                Err(ReplicaError::Rindle(RindleError::Storage(
                    "internal panic during query registration".to_string(),
                )))
            }
        }
    }

    /// Register a **parameterized query family** (design 310 §4 / impl plan §6.1) under
    /// `query_id`: one pipeline over the family's stripped template whose `params`
    /// columns are a partition dimension. Mirrors [`register_query`](Self::register_query)
    /// — recording, the panic boundary, scalar-subquery resolution and planning on the
    /// template — but builds through `build_family_pipeline` and does **not** hydrate:
    /// partitions hydrate individually as they are bound
    /// ([`bind_partition`](Self::bind_partition)). Returns the change-sink id.
    pub fn register_family(
        &mut self,
        query_id: QueryId,
        stripped: &Ast,
        params: &[Box<str>],
    ) -> Result<NodeId, ReplicaError> {
        use std::panic::{catch_unwind, AssertUnwindSafe};

        self.graph.begin_recording();
        let outcome = {
            let graph = &mut self.graph;
            let sources = &self.sources;
            let scalar_sources = &self.scalar_sources;
            let worker = &self.worker;
            let plan_queries = self.plan_queries;
            catch_unwind(AssertUnwindSafe(move || {
                let resolve = |t: &str| sources.get(t).map(|(id, sc)| (*id, sc.clone()));
                let resolved;
                let ast = if has_scalar_subquery(stripped) {
                    let catalog = ForkCatalog(scalar_sources);
                    resolved = resolve_scalars(stripped, &catalog)?;
                    &resolved
                } else {
                    stripped
                };
                let planned;
                let ast = if plan_queries && has_flippable_exists(ast) {
                    let model: Rc<dyn ConnectionCostModel> =
                        Rc::new(CachingCostModel::new(SqliteCostModel::new(worker.clone())));
                    planned = plan_ast(ast, model);
                    &planned
                } else {
                    ast
                };
                let fam = build_family_pipeline(graph, ast, params, BindingSet::new(), &resolve)?;
                let sink = graph.add_change_sink(fam.top);
                graph.set_sink_edge(fam.top, sink);
                Ok::<(NodeId, FamilyPipeline), ReplicaError>((sink, fam))
            }))
        };
        match outcome {
            Ok(Ok((sink, fam))) => {
                let manifest = self.graph.take_recording();
                self.queries.insert(
                    sink,
                    QueryReg {
                        manifest,
                        query_id,
                        family: Some(fam),
                    },
                );
                Ok(sink)
            }
            Ok(Err(e)) => {
                let manifest = self.graph.take_recording();
                self.graph.destroy_pipeline(&manifest);
                Err(e)
            }
            Err(_panic) => {
                let manifest = self.graph.take_recording();
                self.graph.destroy_pipeline(&manifest);
                Err(ReplicaError::Rindle(RindleError::Storage(
                    "internal panic during family registration".to_string(),
                )))
            }
        }
    }

    /// Bind one partition of the family registered under `query_id` and hydrate only it
    /// (design 310 §4.4): the partition's initial `Add` set. `Ok(None)` if `query_id` is not
    /// a registered family; `Err` if the binding is already bound or the hydrate fails
    /// (the binding is then rolled back).
    pub fn bind_partition(
        &self,
        query_id: QueryId,
        binding: &Binding,
    ) -> Result<Option<Vec<CaughtChange>>, ReplicaError> {
        let Some((sink, fam)) = self.family_of(query_id) else {
            return Ok(None);
        };
        Ok(Some(self.graph.bind_family_partition(fam, sink, binding)?))
    }

    /// Unbind one partition of the family registered under `query_id` (design 310 §4.4,
    /// impl plan D5). `Ok(false)` if `query_id` is not a registered family or the binding
    /// was not bound.
    pub fn unbind_partition(
        &self,
        query_id: QueryId,
        binding: &Binding,
    ) -> Result<bool, ReplicaError> {
        let Some((sink, fam)) = self.family_of(query_id) else {
            return Ok(false);
        };
        if !fam.bindings.contains(binding) {
            return Ok(false);
        }
        self.graph.unbind_family_partition(fam, sink, binding)?;
        Ok(true)
    }

    fn family_of(&self, query_id: QueryId) -> Option<(NodeId, &FamilyPipeline)> {
        self.queries
            .iter()
            .find(|(_, reg)| reg.query_id == query_id)
            .and_then(|(&sink, reg)| reg.family.as_ref().map(|f| (sink, f)))
    }

    /// Run `ast` **cold** on a throwaway, instrumented pipeline over a pinned read
    /// snapshot and report where its time and rows go (`ANALYZE-QUERY-DESIGN.md`): the
    /// chosen plan, the per-phase timing, and — per source — the rows it **emitted** into
    /// the pipeline beside the SQLite scan **work** it cost.
    ///
    /// **Read-only and isolated.** It builds and drops a *fresh* [`Graph`] with *fresh*
    /// sources (forks over the worker connection), so it never registers a materialization,
    /// never dedups on the manager's `QueryKey`, never mutates, and never touches
    /// `self.graph` / `self.sources` / the live materialization set. Every leaf fetch runs
    /// inside one deferred read transaction (`BEGIN … ROLLBACK`) so the whole run sees one
    /// consistent snapshot; under wal2 that reader never blocks the writer.
    ///
    /// It runs the **same** planner gate as [`register_query`](Self::register_query)
    /// (`plan_queries && has_flippable_exists`), so the plan it reports is the one a *fresh*
    /// registration would choose — the "current-stats plan," which for a long-lived
    /// materialization whose stats have drifted can legitimately differ from the frozen plan
    /// actually running (§3.1). `BuildError` (unknown table/column, unsupported shape)
    /// surfaces synchronously; an internal-invariant *panic* (e.g. a cost-model assert —
    /// planning runs before `build_pipeline` validates table names) is contained and
    /// surfaced as an error, exactly like [`register_query`](Self::register_query).
    pub fn analyze_query(&self, ast: &Ast) -> Result<AnalyzeReport, ReplicaError> {
        // One fresh read handle + schema per registered table for this run. The forks
        // share the worker connection — safe, because the engine is idle here (analyze
        // is a between-commits FIFO command), so its connection is free for the pinned
        // snapshot. The live sources are byte-for-byte untouched.
        let tables: HashMap<String, (TableSource, SourceSchema)> = self
            .sources
            .iter()
            .map(|(table, (_, schema))| {
                let fork = self
                    .scalar_sources
                    .get(table)
                    .expect("every registered table has a build-time read fork")
                    .fork();
                (table.clone(), (fork, schema.clone()))
            })
            .collect();
        analyze_on(
            &self.worker,
            &tables,
            self.plan_queries,
            self.operator_storage,
            ast,
        )
    }

    /// The hierarchical view [`Schema`] `ast` materializes to — derived from the registered
    /// sources exactly as [`register_query`](Engine::register_query) resolves them (so the
    /// shape, sort, `singular` flag, and in-view relationships line up with the change
    /// stream). Protocol-agnostic; the flat-change / remote layer turns it into a wire
    /// schema + fingerprint. `BuildError` (unknown table/column) surfaces as `Err`.
    pub fn view_schema(&self, ast: &Ast) -> Result<Schema, ReplicaError> {
        let sources = &self.sources;
        let resolve = |t: &str| sources.get(t).map(|(id, sc)| (*id, sc.clone()));
        Ok(view_schema(ast, &resolve)?)
    }

    /// Tear down a registered query: drop its bookkeeping and reclaim its pipeline
    /// (disconnect from the shared sources + free its operator/storage slots). Returns
    /// `false` if `sink` was not a registered query (already gone). Idempotent and safe
    /// against a stale sink id: a non-matching key is simply absent from `queries`.
    pub fn deregister_query(&mut self, sink: NodeId) -> bool {
        match self.queries.remove(&sink) {
            Some(reg) => {
                self.graph.destroy_pipeline(&reg.manifest);
                true
            }
            None => false,
        }
    }

    /// Tear down every query registered under `query_id` and reclaim its pipeline.
    /// The parallel `Cluster` shards one `query_id` to one worker and tears down by
    /// the stable `query_id` (a sink `NodeId` does not survive a fault rebuild);
    /// duplicates registered under the same id are all removed. Returns the count
    /// reclaimed (0 if none — idempotent / unknown id).
    pub fn deregister_by_query_id(&mut self, query_id: QueryId) -> usize {
        let sinks: Vec<NodeId> = self
            .queries
            .iter()
            .filter(|(_, reg)| reg.query_id == query_id)
            .map(|(sink, _)| *sink)
            .collect();
        sinks
            .into_iter()
            .filter(|&sink| self.deregister_query(sink))
            .count()
    }

    /// The caller's [`QueryId`] for a registered query's sink, or `None` if `sink`
    /// is not a live query (used to tag derived deltas for channel-out delivery).
    pub fn query_id_of(&self, sink: NodeId) -> Option<QueryId> {
        self.queries.get(&sink).map(|r| r.query_id)
    }

    /// The [`QueryId`]s of every query currently registered on this engine (used by
    /// the parallel runtime to notify subscribers when the whole engine is torn down
    /// and rebuilt after a derivation fault).
    pub fn query_ids(&self) -> Vec<QueryId> {
        self.queries.values().map(|r| r.query_id).collect()
    }

    /// Fetch `query_id` through its pipeline as a fresh set of hydration `Add`s.
    /// This traverses current sources rather than reading a cached change-stream fold.
    /// [`try_hydrate_change_sink`](rindle::graph::Graph::try_hydrate_change_sink)
    /// leaves the push buffer unchanged, so it is safe to call
    /// on an already-hydrated query between commits. `Ok(None)` if `query_id` is not
    /// registered on this engine; `Err` when the read boundary raises (a parked leaf
    /// error, or the §5.3 sum-overflow state a prior — itself typed-errored — write left
    /// out of range: reads over unrepresentable state error every time, never render a
    /// `NULL` aggregate as data).
    pub fn read_snapshot(
        &self,
        query_id: QueryId,
    ) -> Result<Option<Vec<CaughtChange>>, ReplicaError> {
        let sink = self
            .queries
            .iter()
            .find(|(_, reg)| reg.query_id == query_id)
            .map(|(&sink, _)| sink);
        match sink {
            None => Ok(None),
            // A family's snapshot is every bound partition's rows (design 310 §4.4).
            Some(sink) => match self.queries.get(&sink).and_then(|r| r.family.as_ref()) {
                Some(fam) => Ok(Some(self.graph.hydrate_family(fam, sink)?)),
                None => Ok(Some(self.graph.try_hydrate_change_sink(sink)?)),
            },
        }
    }

    /// Whether opt-in cost-based query planning is enabled (so a post-fault rebuild can
    /// inherit the setting).
    pub fn plan_queries(&self) -> bool {
        self.plan_queries
    }

    /// The operator-scratch-state backend this engine's graph uses (so a post-fault
    /// rebuild re-creates the graph with the same backend).
    pub fn operator_storage(&self) -> OperatorStorage {
        self.operator_storage
    }

    /// Set the D4 derivation memory budget — the estimated bytes of folded rows ONE
    /// transaction may hold across every table it touches. Every registered delta
    /// already shares this budget, so the new ceiling is live immediately; tables
    /// registered later — including by a shed/fault rebuild — inherit it.
    pub fn set_max_delta_bytes(&self, max_bytes: usize) {
        self.delta_budget.set_max_bytes(max_bytes);
    }

    /// The engine's current D4 budget ceiling (so a rebuild can carry it over).
    pub fn max_delta_bytes(&self) -> usize {
        self.delta_budget.max_bytes()
    }

    /// Set the join membership pre-check bounds on this engine's graph
    /// (`designs/311-JOIN-MEMBERSHIP-PRECHECK-DESIGN.md` §8): the per-join distinct-key
    /// bound (`None` = off, the default) and the per-graph key budget. A change resets
    /// every join's set, which then rebuilds by observing its next hydrate-shaped fetch.
    pub fn set_join_precheck_bounds(&self, per_join: Option<usize>, per_graph: usize) {
        self.graph.set_join_precheck_bounds(per_join, per_graph);
    }

    /// The engine's current pre-check bounds (so a rebuild can carry them over).
    pub fn join_precheck_bounds(&self) -> rindle::graph::JoinPrecheckBounds {
        self.graph.join_precheck_bounds()
    }

    /// The operator scratch storage held by the pipeline(s) registered under `query_id`
    /// — the entry count always, and every `(slot, key, value)` when `dump`. `None` if
    /// nothing is registered under that id. Inspection-only (a full `scan` per storage
    /// slot); the leak probe of design 310 impl plan D5 — after unbinding every partition
    /// of a family, its pipeline's scratch state must be back at the count it held when
    /// the bindings were all bound minus their slots, not carrying zombie partitions.
    pub fn storage_report(&self, query_id: QueryId, dump: bool) -> Option<StorageReport> {
        let mut found = false;
        let mut out = StorageReport::default();
        for reg in self.queries.values().filter(|r| r.query_id == query_id) {
            found = true;
            out.entries += self.graph.storage_entries_of(&reg.manifest);
            if dump {
                out.dump.extend(self.graph.storage_dump_of(&reg.manifest));
            }
        }
        found.then_some(out)
    }

    /// A read-only report of the join membership pre-check on this engine: the bounds in
    /// force, the graph-wide tracked-key charge, and — per registered query, sorted by
    /// [`QueryId`] — the state of every join its pipeline owns. The inspection hook the
    /// schedule tests use to compare a live worker against a fresh graph (design 311
    /// §14.1: the knob used to land only at the next commit barrier, after the hydrate
    /// that builds the sets, which no lane could see).
    pub fn join_precheck_report(&self) -> JoinPrecheckReport {
        let mut queries: Vec<(QueryId, Vec<JoinPrecheckState>)> = self
            .queries
            .values()
            .map(|reg| {
                (
                    reg.query_id,
                    self.graph.join_precheck_states_of(reg.manifest.nodes()),
                )
            })
            .collect();
        queries.sort_by_key(|(q, _)| q.0);
        JoinPrecheckReport {
            bounds: self.graph.join_precheck_bounds(),
            tracked_keys: self.graph.join_precheck_tracked_keys(),
            queries,
        }
    }

    /// The connection this engine derives against, as the shared handle a shed/fault
    /// rebuild needs to construct the replacement engine.
    pub fn worker_conn(&self) -> Rc<Connection> {
        self.worker.clone()
    }

    /// The connection this engine derives against (for ad-hoc reads/probes).
    pub fn conn(&self) -> &Connection {
        &self.worker
    }

    // --- The derivation, split into its three phases so a worker thread can
    // interleave the snapshot/commit handshake between them (open + ack, derive,
    // [writer commits], emit + roll back). The single-thread `Db` runs all three
    // back-to-back via `apply_batch`. ---

    /// Arm (or clear) the wall-clock deadline for the next push (FOLLOWER-LAG-SHED §6.6 —
    /// the runaway-push bail). The worker brackets each `apply_and_drain` with it; the
    /// graph's fan-out checkpoints park a `PushDeadlineExceeded` on expiry.
    pub fn set_push_deadline(&self, deadline: Option<std::time::Instant>) {
        self.graph.set_push_deadline(deadline);
    }

    /// **Phase 1.** Open the derivation snapshot: a transaction plus a forcing read
    /// that PINS the read snapshot at the current (pre-commit = T-1) state.
    /// A WAL read snapshot is established at the *first read*, not at `BEGIN`, so
    /// the pin must happen here — before the writer commits the transaction being
    /// derived — or a read during [`apply_and_drain`](Self::apply_and_drain) (after
    /// the writer commits) could slide forward to the post-commit state and
    /// double-count. Reads `sqlite_schema`, so it needs no registered table.
    ///
    /// The transaction is a plain deferred `BEGIN` — the derivation never writes;
    /// batch state lives in each source's [`BatchDelta`], activated here (design 306).
    ///
    /// **Self-healing:** nothing but this method opens a transaction on the worker
    /// connection, so a `BEGIN` that fails with one already open can only mean an
    /// earlier derivation leaked its snapshot. Close it and retry once rather than
    /// leaving the connection pinned at that stale WAL snapshot for the life of the
    /// process (every later `open_snapshot` failing the same way, the shard serving
    /// ever-older hydrations, and the WAL never checkpointing past the held read mark).
    pub fn open_snapshot(&self) -> Result<(), ReplicaError> {
        if let Err(e) = self.worker.execute_batch("BEGIN") {
            if self.worker.is_autocommit() {
                // Not a leaked transaction (busy, I/O, …) — the caller's fault to handle.
                return Err(ReplicaError::sqlite(
                    "worker BEGIN (derivation snapshot)",
                    e,
                ));
            }
            self.rollback(); // discard the leaked transaction + its overlays, then retry once
            self.worker.execute_batch("BEGIN").map_err(|e| {
                ReplicaError::sqlite(
                    "worker BEGIN (derivation snapshot, after clearing a leaked transaction)",
                    e,
                )
            })?;
        }
        if let Err(e) = self
            .worker
            .query_row("SELECT count(*) FROM sqlite_schema", [], |_| Ok(()))
        {
            // Don't leak the just-opened transaction if the forcing read failed.
            let _ = self.worker.execute_batch("ROLLBACK");
            return Err(ReplicaError::sqlite("worker snapshot forcing read", e));
        }
        for delta in self.deltas.values() {
            delta.begin();
        }
        Ok(())
    }

    /// **Phase 2.** Push every *hosted* change in `batch` through the shared
    /// sources — fanning each out to every dependent query — then drain each
    /// affected sink. A change for a table this engine does not host is **skipped**
    /// (another worker hosts it; the single-thread engine hosts every captured
    /// table, so the skip is unreachable there). Returns `(sink, events)` per
    /// changed query. On error, partial sink buffers are discarded so a failed tx
    /// can't leak into the next. The snapshot must be open (Phase 1).
    ///
    /// The whole push+drain is run under a panic boundary so an internal-invariant
    /// *panic* (an engine bug, never valid input) becomes an `Err` rather than
    /// unwinding the worker thread: the per-push `source_push_isolated` already
    /// catches the push fan-out, and this outer guard additionally covers the drain
    /// loop. An `Err` here routes the worker to its in-place tear-down-on-fault
    /// (discard + rebuild the engine) — keeping the worker thread alive instead of
    /// escalating to a full thread death + respawn. Effective under `panic = "unwind"`.
    pub fn apply_and_drain(
        &self,
        batch: &[Captured],
    ) -> Result<Vec<(NodeId, Vec<CaughtChange>)>, ReplicaError> {
        use std::panic::{catch_unwind, AssertUnwindSafe};
        let result = match catch_unwind(AssertUnwindSafe(|| self.push_and_drain(batch))) {
            Ok(r) => r,
            Err(_) => Err(ReplicaError::Rindle(RindleError::Storage(
                "internal panic during derive".to_string(),
            ))),
        };
        if result.is_err() {
            for sink in self.queries.keys().copied().collect::<Vec<_>>() {
                let _ = self.graph.take_sink_changes(sink);
            }
        }
        result
    }

    /// Apply **one** captured change to the shared sources, fanning it out to every
    /// dependent query. A change for a table this engine does not host is **skipped**
    /// (another worker hosts it; the single-thread engine hosts every captured table, so
    /// the skip is unreachable there). Does **not** drain — the deltas this change
    /// produces accumulate in the affected sinks until [`drain`](Self::drain). The
    /// snapshot must be open (Phase 1).
    ///
    /// CONTRACT: a `SourceChange::Edit` is pk-stable — the primary key is row identity,
    /// never mutated in place. The CDC captures an arbitrary SQL statement, so an
    /// `UPDATE … SET <pk> = …` (legal, if pathological) arrives here as a single Edit
    /// whose old/new pks differ. Normalize that ONE case into Remove(old)+Add(new) at
    /// this single ingress, so every consumer — especially the pk-keyed NormalizeFold,
    /// which would otherwise silently drop it — sees the honest row-identity change
    /// (adversarial-review #13). All other producers honor the contract by construction.
    /// The batch is shared (`Arc`) across workers; cloning a `SourceChange` is a refcount
    /// bump, not a row copy. `source_push_isolated` catch_unwinds the operator-graph
    /// fan-out + drains parked errors, so an internal-invariant panic surfaces as `Err`.
    pub fn push(&self, cap: &Captured) -> Result<(), ReplicaError> {
        // Skip-if-not-hosted: only push changes for tables we have a source for.
        let Some(src) = self.sources.get(&*cap.table) else {
            return Ok(());
        };
        self.push_resolved(src, cap)
    }

    /// [`push`](Self::push) with the table already resolved to its source — the seam that
    /// lets [`push_and_drain`](Self::push_and_drain) hoist the name lookup out of the
    /// per-row loop. Behavior is identical; only *who* did the `sources` lookup differs.
    fn push_resolved(
        &self,
        (src_id, schema): &HostedSource,
        cap: &Captured,
    ) -> Result<(), ReplicaError> {
        match &cap.change {
            SourceChange::Edit { old, row } if pk_changed(&schema.primary_key, old, row) => {
                self.graph
                    .source_push_isolated(*src_id, SourceChange::Remove(old.clone()))?;
                self.graph
                    .source_push_isolated(*src_id, SourceChange::Add(row.clone()))?;
            }
            _ => {
                self.graph
                    .source_push_isolated(*src_id, cap.change.clone())?;
            }
        }
        Ok(())
    }

    /// Take whatever deltas are pending in the sinks: drain every registered query's
    /// change-sink and return `(sink, events)` for each that produced output (empty
    /// sinks are skipped). Called once after a batch of [`push`](Self::push)es in the
    /// batch model; per-change in the streaming model (where each push is followed
    /// immediately by a drain so the worker holds nothing across a transaction).
    pub fn drain(&self) -> Vec<(NodeId, Vec<CaughtChange>)> {
        let sinks: Vec<NodeId> = self.queries.keys().copied().collect();
        let mut out = Vec::new();
        for sink in sinks {
            let changes = self.graph.take_sink_changes(sink);
            if !changes.is_empty() {
                out.push((sink, changes));
            }
        }
        out
    }

    fn push_and_drain(
        &self,
        batch: &[Captured],
    ) -> Result<Vec<(NodeId, Vec<CaughtChange>)>, ReplicaError> {
        // THE batch boundary for `rindle_apply_batch_seconds`. Scoped to cover the drain
        // as well as the pushes: what an operator feels is "apply this batch and hand me
        // its deltas", not the fan-out alone. Dropped on the `?` path too, so a failed
        // apply still records the time it burned. Free with `metrics` off (ZST).
        //
        // NOT one observation per transaction, and the metric is named so it cannot be
        // read as one: the coordinator cuts a transaction into `PUSH_CHUNK_ROWS` (1024)
        // chunks and BROADCASTS each chunk to every worker, so a transaction of N rows
        // under W workers produces `W · ceil(N / 1024)` observations here. Quantiles
        // survive that (the W copies are identical); `_count`/`_sum` are ×W. See
        // `rindle::metrics::ApplyBatch`.
        //
        // `batch.len()` is the row count OFFERED to this worker, which is ≥ what it
        // pushes: the loop below skips changes for tables this worker does not host, while
        // a pk-changing Edit fans into two pushes. `rindle_changes_processed_total` is the
        // pushes-actually-made counter; these two are deliberately different questions.
        let _batch = rindle::metrics::apply_batch(batch.len() as u64);
        // One-entry memo over the table resolve, which is otherwise paid `rows × workers`
        // times per transaction: every chunk is broadcast to EVERY worker (`parallel.rs`
        // `StreamTx::push`) though only the shards holding a query on that table host it, so
        // on all the others this name hash IS the per-row cost. The memo caches the NEGATIVE
        // too (note the inner `Option`) — "not hosted here" is the common answer, and
        // re-deriving it per row is exactly what is being removed.
        //
        // Exact, not heuristic: `Captured::table` is the interned key out of the CDC
        // registry, so every row of a table carries the SAME `Arc` and `Arc::ptr_eq` hits
        // for the whole run. Chunks are cut between statements (`cluster.rs` `maybe_pump`)
        // and a statement writes one table, so a bulk chunk collapses to one hash instead
        // of 1024. A miss (interleaved tables, e.g. a trigger cascade) costs one pointer
        // compare and falls through to the lookup. A hit cannot be a false match: pointer
        // equality implies the same allocation, and `batch` keeps every name it holds alive
        // across the loop, so no address can be recycled underneath the memo.
        let mut memo: Option<(&Arc<str>, Option<&HostedSource>)> = None;
        for cap in batch {
            let src = match memo {
                Some((name, src)) if Arc::ptr_eq(name, &cap.table) => src,
                _ => {
                    let hit = self.sources.get(&*cap.table);
                    memo = Some((&cap.table, hit));
                    hit
                }
            };
            if let Some(src) = src {
                self.push_resolved(src, cap)?;
            }
        }
        Ok(self.drain())
    }

    /// **Phase 3.** Discard the derivation state (the writer's COMMIT is the durable
    /// copy): clear every table's [`BatchDelta`], then close the snapshot
    /// transaction. Safe to call
    /// once the snapshot was opened (Phase 1).
    pub fn rollback(&self) {
        for delta in self.deltas.values() {
            delta.end();
        }
        let _ = self.worker.execute_batch("ROLLBACK");
    }

    /// Single-thread convenience (the `Db` path): the three phases back-to-back on
    /// one thread — open the snapshot, derive, roll back. Returns `(sink, events)`
    /// per changed query.
    pub fn apply_batch(
        &self,
        captured: Vec<Captured>,
    ) -> Result<Vec<(NodeId, Vec<CaughtChange>)>, ReplicaError> {
        self.open_snapshot()?;
        let result = self.apply_and_drain(&captured);
        self.rollback();
        result
    }
}

/// Did any primary-key column actually change between an Edit's `old` and `new`? Uses the
/// total, null-identical [`values_identical`] (a pk null in both is unchanged), so this is true
/// only for a genuine pk mutation — which is then modeled as Remove(old)+Add(new), never an
/// in-place Edit.
fn pk_changed(pk: &[ColId], old: &OwnedRow, new: &OwnedRow) -> bool {
    pk.iter()
        .any(|&c| !values_identical(old.col(c), new.col(c)))
}

/// A [`ScalarCatalog`] over the engine's fork read-handles — the seam
/// [`resolve_scalars`] uses to look up a statically-unique child row at build time.
/// Borrows the `scalar_sources` map for the duration of one registration.
struct ForkCatalog<'a>(&'a HashMap<String, TableSource>);

impl ScalarCatalog for ForkCatalog<'_> {
    fn source(&self, table: &str) -> Option<&dyn ScalarSource> {
        self.0.get(table).map(|s| s as &dyn ScalarSource)
    }
}

// ---------------------------------------------------------------------------
// `analyze_query` support
// ---------------------------------------------------------------------------

/// Build a [`Graph`] with the operator-scratch backend `storage` selects — shared by
/// [`Engine::new`] and [`analyze_on`], so an analyze run exercises (and costs) the same
/// `take`/`reduce` scratch backend the live engine is configured with rather than
/// silently buffering in memory.
fn build_graph(storage: OperatorStorage) -> Result<Graph, ReplicaError> {
    Ok(match storage {
        OperatorStorage::Memory => Graph::new(),
        OperatorStorage::SqliteSpill => {
            // A private temp-file scratch DB confined to this thread (the connection is
            // `!Send`, created here, dropped with the graph). Each stateful operator
            // gets a unique `op_id`-namespaced keyspace inside the database's shared
            // `storage` table.
            let db = DatabaseStorage::new_temp_file()
                .map_err(|e| ReplicaError::sqlite("open operator scratch db", e))?;
            Graph::with_storage_factory(StorageFactory::custom(Rc::new(db)))
        }
    })
}

/// A [`ScalarCatalog`] over one analyze run's own read handles (the per-run twin of
/// [`ForkCatalog`], which borrows the engine's long-lived fork map).
struct AnalyzeCatalog<'a>(&'a HashMap<String, (TableSource, SourceSchema)>);

impl ScalarCatalog for AnalyzeCatalog<'_> {
    fn source(&self, table: &str) -> Option<&dyn ScalarSource> {
        self.0.get(table).map(|(s, _)| s as &dyn ScalarSource)
    }
}

/// Everything one standalone `analyze query` run needs, snapshotted off the live
/// engine so the run itself can happen on **any** thread: the DB path, the registered
/// tables, and the engine configuration. Built by
/// [`Cluster::analyze_spec`](crate::Cluster::analyze_spec); consumed by
/// [`analyze_standalone`]. Plain owned data (`Send`), no engine handles.
pub struct AnalyzeSpec {
    pub(crate) path: PathBuf,
    pub(crate) tables: Vec<(String, TableSchema)>,
    pub(crate) plan_queries: bool,
    pub(crate) operator_storage: OperatorStorage,
}

/// Run one `analyze query` COLD on the **calling thread** over a **private** connection
/// to `spec.path`, touching no engine, no worker, and no live statement cache.
///
/// This is the daemon's `/analyze` entry point (run on a dedicated thread there): because
/// everything is analysis-private — a fresh connection with the standard pragmas
/// (`open_journal` — the same opener the workers use), fresh `TableSource`s, a throwaway
/// graph — a genuinely slow cold hydrate blocks only its own thread; commits,
/// materializations, and control-plane commands proceed. Isolation also keeps the
/// numbers honest: the `scan-stats` counters live on this connection's private prepared
/// statements, so a live cursor's history can never inflate them. Under wal2 the pinned
/// deferred read snapshot never blocks the writer.
pub fn analyze_standalone(spec: &AnalyzeSpec, ast: &Ast) -> Result<AnalyzeReport, ReplicaError> {
    // Analysis-only: this connection hydrates and counts, and never writes a row, so the
    // foreign-key posture is inert here — take the default rather than inventing one.
    let conn = Rc::new(open_journal(
        &spec.path,
        JournalMode::default(),
        crate::ForeignKeys::default(),
    )?);
    let mut tables: HashMap<String, (TableSource, SourceSchema)> = HashMap::new();
    for (table, ts) in &spec.tables {
        // Mirror `Engine::register_table`: the same source construction and the same
        // source-schema mapping, so the pipeline this run builds is the one the daemon
        // would run.
        let col_names: Vec<&str> = ts.columns.iter().map(|c| &*c.name).collect();
        let sort: Sort = ts.primary_key.iter().map(|&c| (c, true)).collect();
        let schema = SourceSchema::new(col_names, ts.primary_key.clone(), sort)
            .with_column_types(ts.columns.iter().map(|c| c.ty).collect());
        let source = TableSource::try_new(
            conn.clone(),
            table,
            ts.columns.clone(),
            ts.primary_key.clone(),
        )?;
        tables.insert(table.clone(), (source, schema));
    }
    analyze_on(
        &conn,
        &tables,
        spec.plan_queries,
        spec.operator_storage,
        ast,
    )
}

/// The core of one `analyze query` run over `conn` (`ANALYZE-QUERY-DESIGN.md`): pin a
/// read snapshot, lower `ast` through the same scalar-resolve → planner gate as
/// `register_query` into a throwaway instrumented pipeline over `tables`, hydrate it
/// cold, and assemble the report. Shared by [`Engine::analyze_query`] (worker connection
/// and engine catalog) and [`analyze_standalone`] (private connection, off every engine
/// thread).
///
/// The whole run — planning included — sits under a panic boundary mirroring
/// `register_query`'s: an internal-invariant panic (e.g. a cost-model assert on an AST
/// that names an unknown table, which planning hits *before* `build_pipeline` validates
/// it) surfaces as an error instead of killing the calling thread. The throwaway graph
/// is function-local, so unwind cleanup is just its drop; the read transaction rolls
/// back via the RAII guard on every exit path.
fn analyze_on(
    conn: &Rc<Connection>,
    tables: &HashMap<String, (TableSource, SourceSchema)>,
    plan_queries: bool,
    operator_storage: OperatorStorage,
    ast: &Ast,
) -> Result<AnalyzeReport, ReplicaError> {
    use std::panic::{catch_unwind, AssertUnwindSafe};

    // Pin one consistent snapshot across every leaf fetch; ROLLBACK on ANY exit path
    // (early `?`, panic, normal return) via the RAII guard.
    conn.execute_batch("BEGIN")
        .map_err(|e| ReplicaError::sqlite("analyze: begin read snapshot", e))?;
    let _read_txn = RollbackOnDrop(conn);

    let outcome = catch_unwind(AssertUnwindSafe(|| {
        // 1. A throwaway graph (same operator backend as the live one), one fresh
        //    instrumented source per table: a `CountingSource` counts the rows the
        //    fork vends, and (under the `analyze` feature) a `ScanSink` records its
        //    SQLite scan work.
        let mut graph = build_graph(operator_storage)?;
        let mut metas: HashMap<String, AnalyzeSourceMeta> = HashMap::new();
        for (table, (source, schema)) in tables {
            let base = source.fork();
            #[cfg(feature = "analyze")]
            let scan = {
                let sink: rindle_sqlite::ScanSink =
                    Rc::new(Cell::new(rindle_sqlite::ScanStats::default()));
                base.set_scan_sink(sink.clone());
                sink
            };
            #[cfg(feature = "analyze")]
            let plan: rindle_sqlite::PlanSink = {
                let sink: rindle_sqlite::PlanSink = Rc::new(RefCell::new(None));
                base.set_plan_sink(sink.clone());
                sink
            };
            let counter: EmitCounter = Rc::new(Cell::new(0));
            let counting = CountingSource::new(Box::new(base), counter.clone());
            let node = graph.add_dyn_source(Box::new(counting));
            metas.insert(
                table.clone(),
                AnalyzeSourceMeta {
                    node,
                    counter,
                    schema: schema.clone(),
                    #[cfg(feature = "analyze")]
                    scan,
                    #[cfg(feature = "analyze")]
                    plan,
                },
            );
        }

        // 2. `resolve` records the tables the pipeline actually references (in first-fetch
        //    order), so the report lists exactly the leaves the query touches — a referenced
        //    empty table still appears; an unreferenced registered table does not.
        let referenced = RefCell::new(Vec::<String>::new());
        let resolve = |t: &str| {
            metas.get(t).map(|m| {
                referenced.borrow_mut().push(t.to_string());
                (m.node, m.schema.clone())
            })
        };

        // 3. Scalar resolution → plan gate: the register_query path verbatim, so the
        //    reported plan is the one the daemon would run. Timed (plan phase).
        let resolved;
        let ast: &Ast = if has_scalar_subquery(ast) {
            let catalog = AnalyzeCatalog(tables);
            resolved = resolve_scalars(ast, &catalog)?;
            &resolved
        } else {
            ast
        };
        let plan_start = Instant::now();
        let planning = plan_queries && has_flippable_exists(ast);
        let planned;
        let ast: &Ast = if planning {
            let model: Rc<dyn ConnectionCostModel> =
                Rc::new(CachingCostModel::new(SqliteCostModel::new(conn.clone())));
            planned = plan_ast(ast, model);
            &planned
        } else {
            ast
        };
        let plan_ms = elapsed_ms(plan_start);

        // 4. Build the throwaway pipeline + wire its sink. Timed (build phase).
        let build_start = Instant::now();
        let top = build_pipeline(&mut graph, ast, &resolve)?;
        let sink = graph.add_change_sink(top);
        graph.set_sink_edge(top, sink);
        let build_ms = elapsed_ms(build_start);

        // 5. Hydrate — the cold drain that drives every source fetch (and so all the
        //    emit/scan counting). Timed (hydrate phase, the headline).
        let hydrate_start = Instant::now();
        // Fallible drain: a leaf error parked mid-fetch (an unsafe integer, a failed
        // `sqlite3_step`) or a §5.3 overflow ends the run rather than reporting the
        // partial counts as truth.
        let initial = graph.try_hydrate_change_sink(sink)?;
        let hydrate_ms = elapsed_ms(hydrate_start);
        let output_rows = initial.len();

        // 6. Read the counters back for exactly the referenced leaves (deduped — a
        //    self-join resolves a table twice into one source node) and assemble.
        let mut seen = std::collections::HashSet::new();
        let ordered: Vec<String> = referenced
            .into_inner()
            .into_iter()
            .filter(|t| seen.insert(t.clone()))
            .collect();

        // Concrete `CREATE INDEX` lines for high-amplification leaves, derived once from
        // the planned AST + each table's PK column names (the SQLite-free static advisory).
        let pk_names: HashMap<String, Vec<String>> = tables
            .iter()
            .map(|(t, (_, sc))| {
                let names = sc
                    .primary_key
                    .iter()
                    .map(|&c| sc.columns[c].to_string())
                    .collect();
                (t.clone(), names)
            })
            .collect();
        let pks: BTreeMap<&str, &[String]> = pk_names
            .iter()
            .map(|(t, v)| (t.as_str(), v.as_slice()))
            .collect();
        let advisory_lines = advisories(ast, &pks);

        let mut sources = Vec::with_capacity(ordered.len());
        let mut total_emitted = 0u64;
        let mut total_scanned: Option<u64> = None;
        for table in ordered {
            let m = &metas[&table];
            let emitted = m.counter.get();
            total_emitted += emitted;
            let (scanned, fullscan_steps, scanned_kind) = read_scan(m);
            if let Some(s) = scanned {
                total_scanned = Some(total_scanned.unwrap_or(0) + s);
            }
            let ratio = match scanned {
                Some(s) if emitted > 0 => Some(s as f64 / emitted as f64),
                _ => None,
            };
            // Advise only a leaf whose un-indexed scan work is pathologically *amplified*
            // vs. what it emitted — a per-parent re-scan, not a benign one-time full scan
            // (which a child-driven flip does inherently). The scanned/emitted amplification
            // is the design's diagnostic; `fullscan_steps` is the reliable v1 proxy for it.
            let advisory = match fullscan_steps {
                Some(fs) if is_amplified(fs, emitted) => advisory_lines.get(&table).cloned(),
                _ => None,
            };
            let sqlite_plan = read_plan(m, conn);
            sources.push(AnalyzeSource {
                node: m.node.idx,
                table,
                emitted,
                scanned,
                fullscan_steps,
                ratio,
                advisory,
                sqlite_plan,
                scanned_kind,
            });
        }

        Ok(AnalyzeReport {
            plan: AnalyzePlan {
                planning,
                joins: collect_joins(ast),
                ast: serde_json::to_value(ast).unwrap_or(serde_json::Value::Null),
            },
            timing: AnalyzeTiming {
                plan_ms,
                build_ms,
                hydrate_ms,
            },
            sources,
            totals: AnalyzeTotals {
                scanned: total_scanned,
                emitted: total_emitted,
                output_rows,
            },
        })
        // `graph` drops → the throwaway pipeline is freed.
    }));

    match outcome {
        Ok(result) => result,
        // An internal-invariant panic mid plan/build/hydrate: the throwaway graph was
        // dropped during the unwind and `_read_txn` still rolls back — report the error
        // instead of letting the unwind kill the calling thread.
        Err(_panic) => Err(ReplicaError::Rindle(RindleError::Storage(
            "internal panic during analyze".to_string(),
        ))),
    }
    // `_read_txn` drops → ROLLBACK.
}

/// Per-source scratch for one [`Engine::analyze_query`] run: the throwaway source's
/// `NodeId`, its live emit counter, its schema (for the `resolve` closure), and — under
/// the `analyze` feature — its shared SQLite scan-work sink.
struct AnalyzeSourceMeta {
    node: NodeId,
    counter: EmitCounter,
    schema: SourceSchema,
    #[cfg(feature = "analyze")]
    scan: rindle_sqlite::ScanSink,
    /// Captures this leaf's representative fetch SQL, re-`EXPLAIN`ed after hydration for
    /// the per-leaf access-path text.
    #[cfg(feature = "analyze")]
    plan: rindle_sqlite::PlanSink,
}

/// Read a source's SQLite scan work back after hydration: `(scanned, fullscan_steps,
/// kind)`. The `scanned` proxy is `vm_steps` (design §3.3); a v1 **work** proxy, not a
/// row count. Without the `analyze` feature (or on a memory source) there is nothing to
/// read, so both numbers are `None`.
#[cfg(feature = "analyze")]
fn read_scan(m: &AnalyzeSourceMeta) -> (Option<u64>, Option<u64>, &'static str) {
    let s = m.scan.get();
    (
        Some(s.vm_steps),
        Some(s.fullscan_steps),
        "vm_steps (work proxy)",
    )
}

#[cfg(not(feature = "analyze"))]
fn read_scan(_m: &AnalyzeSourceMeta) -> (Option<u64>, Option<u64>, &'static str) {
    (None, None, "unavailable")
}

/// The per-leaf SQLite access-path text: re-`EXPLAIN` the representative fetch SQL this
/// source captured during hydration (`conn` is the analyze read connection). `None` if the
/// leaf ran no SQL (e.g. never fetched) or the plan is empty. Feature-off ⇒ nothing to read.
#[cfg(feature = "analyze")]
fn read_plan(m: &AnalyzeSourceMeta, conn: &Connection) -> Option<String> {
    let sql = m.plan.borrow().clone()?;
    rindle_sqlite::explain_plan(conn, &sql)
}

#[cfg(not(feature = "analyze"))]
fn read_plan(_m: &AnalyzeSourceMeta, _conn: &Connection) -> Option<String> {
    None
}

/// Milliseconds elapsed since `start`, as an `f64` (sub-ms resolution for the timing
/// report).
fn elapsed_ms(start: Instant) -> f64 {
    start.elapsed().as_secs_f64() * 1000.0
}

/// Whether a leaf's un-indexed scan work is pathologically amplified relative to what it
/// emitted — the v1 "a missing index made SQLite re-scan this table" trigger
/// (`ANALYZE-QUERY-DESIGN.md` §3.5). A present index keeps `fullscan_steps` ~0; a benign
/// one-time full scan keeps it ≈ what it emits; only a **per-parent** re-scan drives it
/// many-fold above `emitted`. The floor keeps a tiny table (where a full scan is cheap and
/// an index wouldn't pay) from tripping it.
fn is_amplified(fullscan_steps: u64, emitted: u64) -> bool {
    /// A full scan under this many steps is cheaper than the index write cost — never advise.
    const FLOOR: u64 = 64;
    /// The scanned-work-per-emitted-row multiple that reads as "an index would collapse this."
    const FACTOR: f64 = 8.0;
    fullscan_steps >= FLOOR && (fullscan_steps as f64) >= (emitted.max(1) as f64) * FACTOR
}

/// RAII: `ROLLBACK` the analyze read transaction on any exit path (early `?`, panic in an
/// unwinding build, or normal return). Analyze never writes, so a rollback is always the
/// right close.
struct RollbackOnDrop<'a>(&'a Connection);

impl Drop for RollbackOnDrop<'_> {
    fn drop(&mut self) {
        let _ = self.0.execute_batch("ROLLBACK");
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A derivation snapshot that was never closed (its worker was dropped from the
    /// transaction stream, so no terminal command ever reached it) must not wedge the
    /// connection: the next `open_snapshot` clears the leaked transaction and retries,
    /// and what it opens is a **real** pin, not a no-op inherited from the leak.
    ///
    /// Without the retry, `BEGIN` fails "cannot start a transaction within a transaction"
    /// for the life of the process — every push on that shard is skipped, every commit
    /// faults, and hydrations serve an ever-older base.
    #[test]
    fn open_snapshot_clears_a_leaked_transaction_and_retries() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("leak.db");
        {
            let c =
                open_journal(&path, JournalMode::default(), crate::ForeignKeys::default()).unwrap();
            c.execute_batch("CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER)")
                .unwrap();
        }
        let conn = Rc::new(
            open_journal(&path, JournalMode::default(), crate::ForeignKeys::default()).unwrap(),
        );
        let engine = Engine::new(conn.clone(), false, OperatorStorage::default()).unwrap();

        engine.open_snapshot().unwrap();
        assert!(
            !conn.is_autocommit(),
            "the snapshot holds a transaction open"
        );
        // …and now LEAK it: no `rollback`, straight into the next transaction's open.
        engine
            .open_snapshot()
            .expect("the leaked transaction is cleared and the BEGIN retried");
        assert!(!conn.is_autocommit(), "the retry left a transaction open");

        // The retried snapshot really is pinned at *this* state: a row another connection
        // commits after it is invisible until the rollback.
        {
            let w =
                open_journal(&path, JournalMode::default(), crate::ForeignKeys::default()).unwrap();
            w.execute("INSERT INTO t (id, v) VALUES (1, 10)", [])
                .unwrap();
        }
        let count = |c: &Connection| {
            c.query_row("SELECT count(*) FROM t", [], |r| r.get::<_, i64>(0))
                .unwrap()
        };
        assert_eq!(
            count(&conn),
            0,
            "the re-opened snapshot pins the pre-insert state"
        );
        engine.rollback();
        assert!(conn.is_autocommit(), "rollback closes the transaction");
        assert_eq!(
            count(&conn),
            1,
            "and the connection moves on to current state"
        );
    }
}