rindle_sqlite/

batch_delta.rs

//! The in-memory **batch delta** (`designs/306-BATCH-OVERLAY-DESIGN.md` §4, impl plan §3):
//! a per-`TableSource` fold of the current transaction's `SourceChange`s, standing in for
//! the throwaway derivation *writes* the historical write-then-abort path performed
//! (deleted with 306 S5). During a
//! derivation, change *k+1* must read a base state that already includes changes *1..k*;
//! this structure holds exactly that divergence from the pinned base snapshot, so the
//! derivation connection can stay read-only.
//!
//! Three parts, each a [`rindle::btree::BTree`] (the engine's own COW B+tree, reused
//! as-is so the delta inherits the already-fuzzed `compare_rows` ordering — plan §3.5):
//!
//! - **`touched`** — every pk the batch touched (present OR removed), keyed under the
//!   pk sort. A base row whose pk is here is *suppressed* from the scan: its
//!   authoritative content now lives in `primary`, or nowhere if the batch removed it.
//! - **`primary`** — the current derived content of each touched-and-present pk.
//! - **`secondary`** — the same rows under each index sort a fetch has asked for,
//!   created lazily by [`BatchDelta::rows_for_fetch`] and maintained eagerly by
//!   [`BatchDelta::apply`] for the rest of the transaction (plan §3.3).
//!
//! The delta is **shared, not forked**: [`TableSource::fork`](crate::TableSource::fork)
//! clones the `Rc<BatchDelta>` because the delta models *storage*, which forks share
//! (plan D1). Range reads vend the tree's **owning** [`BTreeCursor`] (an `Rc<BNode>`
//! spine — the OQ-1 design), constructed under a momentary `RefCell` borrow, so no
//! borrow is ever held across a vend (plan §3.4 as amended 2026-08-10: the original
//! materialize-to-`Vec` rule assumed a borrowing cursor and cost O(|delta|) per fetch
//! — O(N²) per batch under per-change ordered re-fetches). A later `apply` cannot
//! invalidate an in-flight stream either way: it runs at write-after-drain (the same
//! quiet point `write_change` occupied, with the in-flight change carried by the push
//! overlay), and if a cursor ever were alive, `Rc::make_mut` path-copies around every
//! node it pins, leaving the cursor its as-of-fetch snapshot.

use std::cell::{Cell, RefCell};
use std::cmp::Ordering;
use std::collections::HashMap;
use std::rc::Rc;

use rindle::btree::{row_bound_of, BTree, BTreeCursor, RowBound};
use rindle::change::{constraint_matches, Constraint, FetchRequest, SourceChange};
use rindle::source_common::{compare_rows_rev, RowPredicate};
use rindle::value::{compare_rows, ColId, OwnedRow as Row, RowRef, RowStream, Sort};
use rindle::RindleError;
use rindle::{build_scan_start, constrained_index_sort};

/// D4's default derivation memory budget: 256 MiB of *accounted* delta (see
/// [`DeltaBudget`]), shared by every table a transaction touches. High enough that no
/// OLTP transaction or existing test fires it; a bulk load that does trips
/// [`RindleError::DeltaOverflow`] — the load-shed signal (the host commits the data and
/// re-hydrates the affected queries) — instead of an OOM. The spill (design §7 / plan
/// S6) lifts it later.
///
/// Sized to land near the row cap it replaced (1M rows) for a *small* row: at ~100 byte
/// rows with a couple of secondary sorts this trips around 1M rows. The point of the
/// change is the other end — a 1 KiB row now sheds at ~230k rows instead of holding
/// ~1.1 GiB, which the row cap permitted.
pub const DEFAULT_MAX_DELTA_BYTES: usize = 256 * 1024 * 1024;

/// Heap overhead of one `OwnedRow`'s `Arc<[u8]>` allocation: the `ArcInner` header's two
/// `usize` refcounts, on top of the row's own `byte_len()`.
const ARC_HEADER_BYTES: usize = 16;

/// What one row costs in **one** B+tree, beyond its shared buffer: a 16-byte
/// `OwnedRow` fat pointer in the leaf's `Vec<Row>`, doubled to cover `Vec` capacity
/// slack (a leaf holds 32..=64 rows at a capacity rounded up by doubling), plus the
/// internal spine (~1/`MAX_NODE_SIZE` of the entries per level) and each `Rc<BNode>`
/// allocation header.
///
/// Measured at ~37 B/slot by `tests/delta_budget_alloc_probe.rs` (which fails the build
/// if this drifts under what the heap actually does). Rounded up from there on purpose:
/// over-charging sheds a little early, under-charging is the OOM this budget exists to
/// prevent.
const TREE_ENTRY_BYTES: usize = 48;

/// Round an allocation up to the 16-byte granularity a general-purpose allocator
/// hands out.
const fn alloc_granular(bytes: usize) -> usize {
    (bytes + 15) & !15
}

/// The heap an [`OwnedRow`](Row)'s buffer costs, charged once no matter how many of the
/// delta's trees reference it — cloning a row between them is an `Arc` bump.
fn row_bytes(r: &Row) -> usize {
    alloc_granular(ARC_HEADER_BYTES + r.byte_len())
}

/// The **derivation memory budget** (design 306 D4) — what bounds a batch delta, in
/// place of a row count.
///
/// A row count is not a memory bound: rows are variable-length `Arc<[u8]>`, so 1M rows
/// is ~190 MiB of 100-byte rows or ~1.2 GiB of 1 KiB rows, and the cap that protects
/// the first OOMs on the second. This tracks estimated **bytes**, charged as rows enter
/// the delta's trees and released as they leave, so the ceiling means the same thing
/// for every schema.
///
/// **One budget spans every table in the transaction.** `rindle-replica`'s `Engine` hands
/// the same `Rc<DeltaBudget>` to each registered table's delta, so a write touching 20
/// tables is bounded by the budget, not by 20× it. (Threads do not share: a cluster's
/// worker engines each get their own, so a process ceiling is `workers × max_bytes` —
/// tune per-worker, not per-process.)
///
/// **The number is an accounted upper bound, not a measurement.** It sums, per delta:
/// `row_bytes` once per distinct row buffer the delta holds (in `touched`, the
/// superset — `primary` and the secondaries share those same `Arc`s), plus
/// `TREE_ENTRY_BYTES` per tree slot across `touched`, `primary`, and each secondary
/// index. It does not model allocator fragmentation or the connection's own SQLite
/// pager, and it deliberately rounds against itself.
/// `tests/delta_budget_alloc_probe.rs` pins it against a live-bytes global allocator and
/// fails the build if it ever drops below what the heap actually retains.
pub struct DeltaBudget {
    max_bytes: Cell<usize>,
    used: Cell<usize>,
}

impl DeltaBudget {
    /// A budget of `max_bytes`, nothing charged against it yet.
    pub fn new(max_bytes: usize) -> DeltaBudget {
        DeltaBudget {
            max_bytes: Cell::new(max_bytes),
            used: Cell::new(0),
        }
    }

    /// Re-tune the ceiling. Takes effect at the next overflow check; safe on a live,
    /// shared budget — the limit is only ever read, never part of any structure.
    pub fn set_max_bytes(&self, max_bytes: usize) {
        self.max_bytes.set(max_bytes);
    }

    /// The current ceiling.
    pub fn max_bytes(&self) -> usize {
        self.max_bytes.get()
    }

    /// Estimated bytes currently held across every delta sharing this budget.
    pub fn used(&self) -> usize {
        self.used.get()
    }

    fn charge(&self, bytes: usize) {
        self.used.set(self.used.get().saturating_add(bytes));
    }

    fn release(&self, bytes: usize) {
        self.used.set(self.used.get().saturating_sub(bytes));
    }

    fn exceeded(&self) -> bool {
        self.used.get() > self.max_bytes.get()
    }
}

impl Default for DeltaBudget {
    fn default() -> Self {
        DeltaBudget::new(DEFAULT_MAX_DELTA_BYTES)
    }
}

/// What the delta knows about a primary key ([`BatchDelta::lookup_pk`]).
#[derive(Clone, Debug)]
pub enum DeltaLookup {
    /// The batch never touched this pk — fall through to the base snapshot.
    Untouched,
    /// The batch removed this pk; the base row (if any) is stale. Authoritative: gone.
    Absent,
    /// The batch's current derived content for this pk.
    Present(Row),
}

struct Inner {
    /// Every pk the batch touched (present OR removed). Stores whole rows keyed under
    /// `pk_sort` — the comparator reads only the pk columns, so any version of the row
    /// is an equivalent key and no probe-row synthesis is needed (plan §3.1).
    touched: BTree,
    /// Current derived content of each touched-and-present pk. Ordered by `pk_sort`.
    /// Invariant: `primary ⊆ touched` (under pk equality).
    primary: BTree,
    /// The same rows under each `index_sort` a fetch has asked for — keyed by the SAME
    /// `constraint columns ++ connection sort` computation `MemorySource::fetch` uses
    /// (plan §3.2), never by the bare connection sort. `pk_sort` itself is served by
    /// `primary` directly and never lands here.
    secondary: HashMap<Sort, BTree>,
    pk_sort: Sort,
    primary_key: Vec<ColId>,
    /// `false` ⇒ every read path short-circuits (the hot fast path: outside a
    /// derivation the delta must cost nothing).
    active: bool,
    /// The batch outgrew the [`DeltaBudget`] and the delta gave up: its trees were
    /// dropped (the memory bound is the point) and every further [`BatchDelta::apply`]
    /// re-raises [`RindleError::DeltaOverflow`] without accumulating. Reads while
    /// overflowed see an empty delta — nothing may derive from it; the host's shed
    /// path (teardown + re-hydrate) is the defined recovery. Reset by `begin`/`end`.
    overflowed: bool,
    /// The budget this delta charges against — shared with every other table in the
    /// same derivation (design 306 D4).
    budget: Rc<DeltaBudget>,
    /// What THIS delta has charged to `budget`, so `begin`/`end`/overflow can release
    /// exactly its own contribution and leave its peers' alone.
    owed: usize,
}

/// The per-source batch delta. Interior-`RefCell` so one `Rc<BatchDelta>` can be shared
/// between a `TableSource`, its forks, and the engine that drives the lifecycle
/// (plan D1 — the same idiom as `cursors_open`/`fetch_error`).
pub struct BatchDelta {
    inner: RefCell<Inner>,
}

impl BatchDelta {
    /// An **inactive** delta over `primary_key` with its own private
    /// [`DEFAULT_MAX_DELTA_BYTES`] budget. Every embedded caller gets one of these
    /// implicitly and never pays for it; a host driving several tables replaces it with
    /// one shared budget via [`set_budget`](Self::set_budget).
    pub fn new(primary_key: Vec<ColId>) -> BatchDelta {
        Self::with_budget(primary_key, Rc::new(DeltaBudget::default()))
    }

    /// An inactive delta charging against `budget` (D4) — pass the same `Rc` to every
    /// table in a derivation and the ceiling bounds the whole transaction.
    pub fn with_budget(primary_key: Vec<ColId>, budget: Rc<DeltaBudget>) -> BatchDelta {
        let pk_sort: Sort = primary_key.iter().map(|&c| (c, true)).collect();
        BatchDelta {
            inner: RefCell::new(Inner {
                touched: BTree::new(),
                primary: BTree::new(),
                secondary: HashMap::new(),
                pk_sort,
                primary_key,
                active: false,
                overflowed: false,
                budget,
                owed: 0,
            }),
        }
    }

    /// An inactive delta with a private budget of `max_bytes` — the single-table
    /// convenience over [`with_budget`](Self::with_budget).
    pub fn with_max_bytes(primary_key: Vec<ColId>, max_bytes: usize) -> BatchDelta {
        Self::with_budget(primary_key, Rc::new(DeltaBudget::new(max_bytes)))
    }

    /// Move this delta onto `budget` — how a host installs one shared ceiling across
    /// every table it registers, on a delta the source already created (and may already
    /// have forked: the budget lives behind the shared `RefCell`, so forks follow).
    /// Anything currently charged moves with it.
    pub fn set_budget(&self, budget: Rc<DeltaBudget>) {
        let mut inner = self.inner.borrow_mut();
        let owed = inner.owed;
        inner.budget.release(owed);
        budget.charge(owed);
        inner.budget = budget;
    }

    /// The budget this delta charges against (shared — re-tune it with
    /// [`DeltaBudget::set_max_bytes`], read it with [`DeltaBudget::used`]).
    pub fn budget(&self) -> Rc<DeltaBudget> {
        Rc::clone(&self.inner.borrow().budget)
    }

    /// Estimated bytes **this** delta is holding (its share of `budget().used()`).
    pub fn used_bytes(&self) -> usize {
        self.inner.borrow().owed
    }

    // -- lifecycle (plan §3.3) --

    /// Start a derivation: clear all state and mark active. Replaces the write path's
    /// `BEGIN CONCURRENT` throwaway transaction. Drops the `secondary` **map** entirely
    /// (not just its rows) so `M` in the fold cost is bounded by the sorts actually
    /// fetched during *this* transaction.
    pub fn begin(&self) {
        let mut inner = self.inner.borrow_mut();
        inner.clear();
        inner.active = true;
        inner.overflowed = false;
    }

    /// End a derivation: clear and deactivate. Replaces `execute_batch("ROLLBACK")` and
    /// is strictly more reliable — there is no path on which a failed or skipped
    /// rollback leaves derivation state behind.
    pub fn end(&self) {
        let mut inner = self.inner.borrow_mut();
        inner.clear();
        inner.active = false;
        inner.overflowed = false;
    }

    pub fn is_active(&self) -> bool {
        self.inner.borrow().active
    }

    pub fn is_empty(&self) -> bool {
        self.inner.borrow().touched.is_empty()
    }

    // -- the fold (plan §3.1) --

    /// Fold one change into the delta — the stand-in for `TableSource::write_change`,
    /// called from exactly where the write is called (after the fan-out drains), so the
    /// I1/I2 push invariants hold by construction (design §3.1).
    ///
    /// Erroring on an inactive delta is deliberate (plan §3.3): after the write closure
    /// becomes `delta.apply` unconditionally, a push outside an open snapshot must fail
    /// loudly rather than silently diverge.
    ///
    /// A batch whose folded rows outgrow the shared [`DeltaBudget`] trips
    /// [`RindleError::DeltaOverflow`] (design 306 D4): the delta drops its trees on the
    /// spot — the budget is a memory bound, so overflowing must not keep holding the
    /// rows — and every further `apply` re-raises the same error without accumulating.
    /// The derivation for this batch is unrecoverable; the host sheds (commits the data,
    /// tears down + re-hydrates the affected queries) rather than failing the write.
    ///
    /// The charge is **net**: re-touching rows already in the delta costs nothing beyond
    /// the change in their content size, so a transaction that rewrites the same working
    /// set a million times never sheds — only one that holds an unbounded amount of it
    /// does.
    ///
    /// An `Add` of a pk already present is [`RindleError::ConsistencyViolation`] in
    /// every build (see `Inner::add_row`) — the loud fault the write-then-abort
    /// path's UNIQUE constraint used to raise, never a silent overwrite.
    pub fn apply(&self, change: &SourceChange) -> Result<(), RindleError> {
        let mut inner = self.inner.borrow_mut();
        if !inner.active {
            return Err(RindleError::Storage(
                "batch delta: apply outside an active derivation (no open snapshot)".into(),
            ));
        }
        if inner.overflowed {
            return Err(inner.overflow_error());
        }
        match change {
            SourceChange::Add(r) => {
                inner.add_row(r)?;
            }
            SourceChange::Remove(r) => {
                inner.touch(r);
                inner.delete_stored(r);
            }
            SourceChange::Edit { row, old } => {
                // `Engine::push` normalizes a pk-*changing* Edit to Remove(old) +
                // Add(row) before the fan-out — but the delta is a public seam, so
                // fold the same normalization here rather than trusting the caller:
                // BOTH pks must end up touched, or the old base row stays visible in
                // merged scans (it used to be a debug_assert with a silently-corrupt
                // release fallthrough).
                if compare_rows(&inner.pk_sort, old, row) != Ordering::Equal {
                    inner.touch(old);
                }
                inner.delete_stored(old);
                inner.add_row(row)?;
            }
        }
        if inner.budget.exceeded() {
            let err = inner.overflow_error();
            // Drop the rows AND hand their bytes back: a shed that kept charging would
            // poison the budget for the peers whose deltas are still within it.
            inner.clear();
            inner.overflowed = true;
            return Err(err);
        }
        Ok(())
    }

    // -- reads (plan §3.4) --

    /// What the delta says about `probe`'s primary key. `probe` is any row carrying the
    /// pk columns — the pk-sort comparator reads nothing else.
    pub fn lookup_pk(&self, probe: &Row) -> DeltaLookup {
        let inner = self.inner.borrow();
        if !inner.active || !inner.touched.has(probe, &inner.pk_sort) {
            return DeltaLookup::Untouched;
        }
        match inner.primary.get(probe, &inner.pk_sort) {
            Some(r) => DeltaLookup::Present(r.clone()),
            None => DeltaLookup::Absent,
        }
    }

    /// Whether a vended base row must be suppressed from the scan: its pk was touched,
    /// so its authoritative content lives in the delta (or nowhere). One pk-sort B-tree
    /// probe per vended base row — the merge's per-row cost (design §5).
    pub fn suppresses(&self, row: &Row) -> bool {
        let inner = self.inner.borrow();
        inner.active && inner.touched.has(row, &inner.pk_sort)
    }

    /// The delta row matching every `(col, value)` equality of `key`, if any — the
    /// delta-first half of a `get_row`/`lookup_unique` point read (plan §4.4). `key` may
    /// be any unique key, not just the pk, so this scans `primary` linearly
    /// ([`constraint_matches`] semantics, `values_equal`: null never matches) — bounded
    /// by the delta, and these are build-time / consistency reads, not the hot path.
    pub fn find_by_key(&self, key: &[(ColId, rindle::value::OwnedValue)]) -> Option<Row> {
        let inner = self.inner.borrow();
        if !inner.active || inner.primary.is_empty() {
            return None;
        }
        let key: Constraint = key.to_vec();
        let mut cur = inner.primary.values_from(None, true, &inner.pk_sort);
        while let Some(r) = cur.next_row() {
            if constraint_matches(r, &key) {
                return Some(r.clone());
            }
        }
        None
    }

    /// The delta side of one `fetch` (plan §4.2/§4.3): the rows this transaction's
    /// delta contributes to a request, narrowed exactly as [`compute_overlays`] narrows
    /// the single-change overlay — constraint, multi-constraint `any`, `predicate`
    /// (§2.1: delta rows, like overlay rows, are never seen by SQL, so `predicate` —
    /// which can be strictly stronger than `sql_condition` — is the rule).
    ///
    /// **Lazy** (plan §3.4 as amended): returns a [`DeltaRows`] pull stream over the
    /// tree's owning cursor rather than a materialized `Vec`, so a fetch costs one
    /// O(log |delta|) seek plus the rows the merge actually consumes — never
    /// O(|delta|) per fetch (the O(N²)-per-batch trap under per-change ordered
    /// re-fetches). The `RefCell` borrow lives only inside this call; the cursor owns
    /// its `Rc<BNode>` spine (OQ-1), so nothing of the borrow escapes, and a later
    /// `apply` path-copies around any node a live stream pins.
    ///
    /// Vended in **connection-sort order** (`sort`, honoring `req.reverse`): the seek
    /// index is `constraint columns ++ sort` (§3.2), and restricted to the group an
    /// equality constraint selects, that order *is* the connection sort — which is what
    /// makes the output mergeable against SQL's `ORDER BY sort` stream.
    ///
    /// `start` is a seek optimization only; the exact `At`/`After` cut is re-applied
    /// downstream by `generate_with_start`, exactly as the tiebreak path relies on.
    ///
    /// [`compute_overlays`]: rindle::source_common::compute_overlays
    pub fn rows_for_fetch(
        &self,
        req: &FetchRequest,
        sort: &Sort,
        predicate: Option<&RowPredicate>,
    ) -> DeltaRows {
        let constraint = req.constraint.as_ref();
        let mut inner = self.inner.borrow_mut();
        if !inner.active || inner.touched.is_empty() {
            return DeltaRows::empty();
        }

        // INDEX SORT: the shared `MemorySource::fetch` rule, computed from
        // `req.constraint` only (never from `multi_constraints` — their fetches
        // scan the bare-sort index and filter, because the merge order must match
        // the base stream, plan §4.2).
        let index_sort = constrained_index_sort(constraint, sort, &inner.primary_key);
        inner.ensure_index(&index_sort);

        // Seek: constraint values with reverse-aware sentinels for unconstrained
        // columns; with no constraint, seek to the plain start row (the basis is
        // applied downstream, not by the seek). Always inclusive.
        let scan_start: Option<RowBound> = match constraint {
            Some(c) => Some(build_scan_start(c, &index_sort, req.reverse)),
            None => req
                .start
                .as_ref()
                .map(|s| row_bound_of(&s.row, &index_sort)),
        };

        let tree = inner.tree_for(&index_sort);
        let cur = if req.reverse {
            tree.values_from_reversed(scan_start.as_ref(), true, &index_sort)
        } else {
            tree.values_from(scan_start.as_ref(), true, &index_sort)
        };
        DeltaRows {
            cur: Some(cur),
            constraint: req.constraint.clone(),
            multi: req
                .multi_constraints
                .iter()
                .filter(|mc| !mc.is_empty())
                .cloned()
                .collect(),
            predicate: predicate.cloned(),
        }
        // `inner`'s borrow ends here; `cur` owns its spine.
    }
}

/// The lazily-pulled delta side of one fetch ([`BatchDelta::rows_for_fetch`]): the
/// owning [`BTreeCursor`] plus the request's narrowing, applied **per pull** —
/// constraint group-break, multi-constraint `any`, then the predicate (arbitrary
/// user-compiled code, safe to run here because no `RefCell` borrow is held; the
/// same guarantee materializing used to buy, now free).
pub struct DeltaRows {
    /// `None` after exhaustion — and from birth for the inactive/empty
    /// short-circuit, which must build no index and touch no tree.
    cur: Option<BTreeCursor>,
    /// The group-break: the seek index leads with these columns, so the first
    /// non-matching row ends the stream (a `break`, not a filter).
    constraint: Option<Constraint>,
    /// The non-empty multiConstraints only (an empty entry constrains nothing):
    /// keep a row iff it matches SOME entry of EVERY one.
    multi: Vec<Vec<Constraint>>,
    predicate: Option<RowPredicate>,
}

impl DeltaRows {
    fn empty() -> DeltaRows {
        DeltaRows {
            cur: None,
            constraint: None,
            multi: Vec::new(),
            predicate: None,
        }
    }
}

impl Iterator for DeltaRows {
    type Item = Row;

    /// Pull the next matching delta row, owned (an `Arc` bump per vended row).
    fn next(&mut self) -> Option<Row> {
        let cur = self.cur.as_mut()?;
        while let Some(r) = cur.next_row() {
            if let Some(c) = &self.constraint {
                if !constraint_matches(r, c) {
                    // Past the matching group — every later row fails too.
                    break;
                }
            }
            if !self
                .multi
                .iter()
                .all(|mc| mc.iter().any(|c| constraint_matches(r, c)))
            {
                continue;
            }
            if let Some(p) = &self.predicate {
                if !p(r) {
                    continue;
                }
            }
            return Some(r.clone());
        }
        self.cur = None;
        None
    }
}

impl Inner {
    /// Charge `bytes` to the shared budget, remembering this delta's own share.
    fn charge(&mut self, bytes: usize) {
        self.owed = self.owed.saturating_add(bytes);
        self.budget.charge(bytes);
    }

    /// Hand `bytes` back.
    fn release(&mut self, bytes: usize) {
        self.owed = self.owed.saturating_sub(bytes);
        self.budget.release(bytes);
    }

    /// Recompute what this delta *should* owe, straight from the trees — the model in
    /// [`DeltaBudget`]'s docs, expressed once more as an oracle. `charge`/`release` are
    /// spread across `touch`/`add_row`/`delete_stored`/`ensure_index`, each of which has
    /// to mirror the others exactly (a secondary index created between an `add_row` and
    /// its `delete_stored` changes how many slots that row occupies); this is how that
    /// symmetry gets machine-checked instead of argued. O(|delta|), debug builds only,
    /// once per derivation — so every test in the suite validates the accounting.
    #[cfg(debug_assertions)]
    fn audit(&self) -> usize {
        let mut bytes = 0;
        let mut cur = self.touched.values_from(None, true, &self.pk_sort);
        while let Some(r) = cur.next_row() {
            bytes += row_bytes(r);
        }
        let slots = self.touched.len()
            + self.primary.len()
            + self.secondary.values().map(|t| t.len()).sum::<usize>();
        bytes + slots * TREE_ENTRY_BYTES
    }

    /// Drop every tree and return this delta's whole charge to the budget. The one
    /// place trees are emptied, so no path can drop rows while still holding their
    /// bytes against the peers sharing the budget.
    fn clear(&mut self) {
        // A `#[cfg]` block, not a bare `debug_assert_eq!`: the macro still type-checks
        // its arguments in release, so `audit` would have to be compiled there too.
        // This way the O(|delta|) oracle does not exist in a release build at all.
        #[cfg(debug_assertions)]
        {
            let audited = self.audit();
            debug_assert_eq!(
                self.owed, audited,
                "batch delta: budget accounting drifted from what the trees hold"
            );
        }
        self.touched = BTree::new();
        self.primary = BTree::new();
        self.secondary = HashMap::new();
        let owed = self.owed;
        self.release(owed);
    }

    /// The D4 shed signal, reporting the shared budget's state at the trip.
    fn overflow_error(&self) -> RindleError {
        RindleError::DeltaOverflow {
            max_bytes: self.budget.max_bytes(),
            used: self.budget.used(),
        }
    }

    /// `touched.add` under the pk sort. Duplicate touches replace the slot (`BTree::add`
    /// overwrites — any version of the row is an equivalent key).
    ///
    /// **This is where row content is charged to the budget, and the only place.**
    /// `touched` is the superset — every pk the batch touched is here, and after an
    /// `add_row` its slot holds the very `Arc` `primary` and the secondaries hold — so
    /// charging here counts each buffer exactly once, and `primary`/secondary
    /// membership costs only [`TREE_ENTRY_BYTES`]. It is also what makes a bulk DELETE
    /// accounted at all: those rows live in `touched` and nowhere else.
    fn touch(&mut self, r: &Row) {
        let cost = row_bytes(r);
        match self.touched.add_replacing(r.clone(), &self.pk_sort) {
            // Replaced: this pk's buffer changed, so charge the difference only —
            // net accounting is what keeps a re-touched working set from drifting up.
            Some(prev) => {
                let prev_cost = row_bytes(&prev);
                self.charge(cost);
                self.release(prev_cost);
            }
            None => {
                self.charge(cost + TREE_ENTRY_BYTES);
            }
        }
    }

    /// Insert `r` into `primary` and every secondary index (a row clone is one `Arc`
    /// bump). Also records the touch.
    ///
    /// An ADD of a pk whose content is already present is a malformed change stream —
    /// a real error in every build, not a debug assert: the write-then-abort path it
    /// replaced hit the table's UNIQUE pk constraint here (a loud derive fault →
    /// teardown + re-hydrate), and silently overwriting instead would double-count
    /// the row in the fan-out while `apply()` reports success. The delta is left
    /// abandoned (primary overwritten, secondaries untouched) — the error unwinds the
    /// derivation, and `end()` clears everything before anything reads it.
    fn add_row(&mut self, r: &Row) -> Result<(), RindleError> {
        self.touch(r);
        let newly = self.primary.add(r.clone(), &self.pk_sort);
        if !newly {
            return Err(RindleError::ConsistencyViolation {
                kind: "batch delta: ADD of a pk already present in the delta",
            });
        }
        for (s, tree) in self.secondary.iter_mut() {
            tree.add(r.clone(), s);
        }
        // `primary` + one slot per secondary. The row's *buffer* is already charged by
        // `touch` and shared with all of them (a clone is an `Arc` bump), so these
        // memberships cost only their tree slots.
        let slots = 1 + self.secondary.len();
        self.charge(slots * TREE_ENTRY_BYTES);
        Ok(())
    }

    /// Delete the pk of `probe` from `primary` and every secondary — **by the currently
    /// stored row, never by the incoming one** (plan §3.1, the one trap): secondary keys
    /// include non-pk columns, and when a pk is touched twice in one batch the incoming
    /// `old` is the *original base* row, not what the delta holds. No-op if the pk has
    /// no stored content (the batch is removing/editing a pure base row).
    fn delete_stored(&mut self, probe: &Row) {
        let prev = self.primary.get(probe, &self.pk_sort).cloned();
        if let Some(prev) = prev {
            self.primary.delete(&prev, &self.pk_sort);
            for (s, tree) in self.secondary.iter_mut() {
                let removed = tree.delete(&prev, s);
                debug_assert!(removed, "batch delta: secondary index missing a stored row");
            }
            // Mirror `add_row`: give back the slots, never the buffer — `touched` still
            // holds this pk (that is what suppresses the base row) and still owns the
            // content charge.
            let slots = 1 + self.secondary.len();
            self.release(slots * TREE_ENTRY_BYTES);
        }
    }

    /// Build the index for `index_sort` from `primary` if it does not exist yet —
    /// `MemorySource::ensure_index` almost verbatim: own each primary row (`Arc` bumps),
    /// re-sort under the new comparator, bulk-load (`from_sorted`, O(|D| log |D|),
    /// bounded by the delta and never by the table). The pk sort itself is served by
    /// `primary` directly ([`Inner::tree_for`]) and is never duplicated here.
    fn ensure_index(&mut self, index_sort: &Sort) {
        if *index_sort == self.pk_sort || self.secondary.contains_key(index_sort) {
            return;
        }
        let mut rows: Vec<Row> = {
            let mut cur = self.primary.values_from(None, true, &self.pk_sort);
            let mut v = Vec::with_capacity(self.primary.len());
            while let Some(r) = cur.next_row() {
                v.push(r.clone());
            }
            v
        };
        rows.sort_by(|a, b| compare_rows(index_sort, a, b));
        // One new slot per primary row (the buffers are the same `Arc`s). Charged, but
        // NOT checked here: `ensure_index` is on the read path, which has no way to
        // report an overflow — the next `apply` sees the raised total and sheds.
        self.charge(rows.len() * TREE_ENTRY_BYTES);
        self.secondary.insert(
            index_sort.clone(),
            BTree::from_sorted(rows.into_iter(), index_sort),
        );
    }

    /// The tree serving `index_sort`: `primary` for the pk sort, else the (already
    /// ensured) secondary.
    fn tree_for(&self, index_sort: &Sort) -> &BTree {
        if *index_sort == self.pk_sort {
            &self.primary
        } else {
            self.secondary
                .get(index_sort)
                .expect("ensure_index built it")
        }
    }
}

// ---------------------------------------------------------------------------
// The merge (plan §4.2): base ⊎ delta, under the connection sort
// ---------------------------------------------------------------------------

/// Two-way sorted merge of the SQL leaf stream and the delta's rows for one fetch
/// (design §5, plan §4.2). Sits *below* `generate_with_overlay_checked`, exactly where
/// `TiebreakStream` sits, so every downstream consumer (the overlay splice, the start
/// cut) is unchanged.
///
/// - **base side** — the SQL cursor, dropping any row whose pk the delta touched
///   ([`BatchDelta::suppresses`] — its authoritative content is on the delta side, or
///   nowhere).
/// - **delta side** — [`BatchDelta::rows_for_fetch`], already narrowed (constraint /
///   multi-constraint / predicate) and in connection-sort order.
/// - **merged** under the **connection `sort`** — the comparator the overlay splice
///   receives. There is no `index_sort` in `TableSource::fetch`; the delta's
///   constraint-prefixed index is a *seek* structure whose group order coincides with
///   the connection sort (plan §4.2).
///
/// Owning each base row here costs nothing net: `LeafRows` above owns every row it
/// pulls anyway, and its `to_owned_row` on the vended `&OwnedRow` degrades to an `Arc`
/// bump. Like `TiebreakStream`, this inherits `LeafRows`' error duty — a
/// `try_to_owned_row` failure is parked in the shared fetch error sink.
pub(crate) struct BatchMerge<S: RowStream> {
    base: S,
    delta: Rc<BatchDelta>,
    delta_rows: DeltaRows,
    base_peek: Option<Row>,
    delta_peek: Option<Row>,
    sort: Sort,
    reverse: bool,
    /// The row being lent out (the lending `RowStream` contract).
    current: Option<Row>,
    base_done: bool,
    error_sink: Rc<RefCell<Option<RindleError>>>,
}

impl<S: RowStream> BatchMerge<S> {
    pub(crate) fn new(
        base: S,
        delta: Rc<BatchDelta>,
        req: &FetchRequest,
        sort: &Sort,
        predicate: Option<&RowPredicate>,
        error_sink: Rc<RefCell<Option<RindleError>>>,
    ) -> BatchMerge<S> {
        let delta_rows = delta.rows_for_fetch(req, sort, predicate);
        BatchMerge {
            base,
            delta,
            delta_rows,
            base_peek: None,
            delta_peek: None,
            sort: sort.clone(),
            reverse: req.reverse,
            current: None,
            base_done: false,
            error_sink,
        }
    }

    /// Pull the next non-suppressed base row, owned (the shared [`pull_base_row`]).
    fn pull_base(&mut self) -> Option<Row> {
        pull_base_row(
            &mut self.base,
            &mut self.base_done,
            &self.delta,
            &self.error_sink,
        )
    }
}

/// Pull the next non-suppressed base row, owned — the base-side loop both merge
/// shapes share. Parks a materialization error into the shared sink and reports
/// end-of-stream (matching `LeafRows::next`).
fn pull_base_row<S: RowStream>(
    base: &mut S,
    base_done: &mut bool,
    delta: &BatchDelta,
    error_sink: &RefCell<Option<RindleError>>,
) -> Option<Row> {
    if *base_done {
        return None;
    }
    loop {
        match base.next_row() {
            None => {
                *base_done = true;
                return None;
            }
            Some(r) => match r.try_to_owned_row() {
                Ok(row) => {
                    if delta.suppresses(&row) {
                        continue;
                    }
                    return Some(row);
                }
                Err(err) => {
                    *error_sink.borrow_mut() = Some(err);
                    *base_done = true;
                    return None;
                }
            },
        }
    }
}

impl<S: RowStream> RowStream for BatchMerge<S> {
    type Row<'a>
        = &'a Row
    where
        Self: 'a;

    fn next_row(&mut self) -> Option<&Row> {
        if self.base_peek.is_none() {
            self.base_peek = self.pull_base();
        }
        if self.delta_peek.is_none() {
            self.delta_peek = self.delta_rows.next();
        }
        let take_delta = match (&self.base_peek, &self.delta_peek) {
            (None, None) => return None,
            (Some(_), None) => false,
            (None, Some(_)) => true,
            // A tie is impossible (sort includes the full pk, and a base row whose pk
            // the delta holds was suppressed); delta-first on a tie is inert.
            (Some(b), Some(d)) => {
                compare_rows_rev(&self.sort, self.reverse, d, b) != Ordering::Greater
            }
        };
        self.current = if take_delta {
            self.delta_peek.take()
        } else {
            self.base_peek.take()
        };
        self.current.as_ref()
    }
}

/// The unordered peer of [`BatchMerge`] for `unordered` connections: delta rows are
/// vended first (order is not a contract here — mirroring the unordered overlay's
/// eager head injection), then the base stream with touched pks suppressed.
pub(crate) struct BatchMergeUnordered<S: RowStream> {
    base: S,
    delta: Rc<BatchDelta>,
    delta_rows: DeltaRows,
    current: Option<Row>,
    base_done: bool,
    error_sink: Rc<RefCell<Option<RindleError>>>,
}

impl<S: RowStream> BatchMergeUnordered<S> {
    /// `req` must carry no `start` (nothing downstream of the unordered splice
    /// re-applies a cut); the caller strips it.
    pub(crate) fn new(
        base: S,
        delta: Rc<BatchDelta>,
        req: &FetchRequest,
        sort: &Sort,
        predicate: Option<&RowPredicate>,
        error_sink: Rc<RefCell<Option<RindleError>>>,
    ) -> BatchMergeUnordered<S> {
        debug_assert!(req.start.is_none(), "unordered merge with a start bound");
        let delta_rows = delta.rows_for_fetch(req, sort, predicate);
        BatchMergeUnordered {
            base,
            delta,
            delta_rows,
            current: None,
            base_done: false,
            error_sink,
        }
    }
}

impl<S: RowStream> RowStream for BatchMergeUnordered<S> {
    type Row<'a>
        = &'a Row
    where
        Self: 'a;

    fn next_row(&mut self) -> Option<&Row> {
        if let Some(d) = self.delta_rows.next() {
            self.current = Some(d);
            return self.current.as_ref();
        }
        self.current = pull_base_row(
            &mut self.base,
            &mut self.base_done,
            &self.delta,
            &self.error_sink,
        );
        self.current.as_ref()
    }
}

#[cfg(test)]
impl BatchDelta {
    /// Test convenience: drain [`Self::rows_for_fetch`] (tests compare whole sets).
    fn fetch_vec(
        &self,
        req: &FetchRequest,
        sort: &Sort,
        predicate: Option<&RowPredicate>,
    ) -> Vec<Row> {
        self.rows_for_fetch(req, sort, predicate).collect()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use rindle::change::{Basis, Start};
    use rindle::value::{owned_row, OwnedValue as V};
    use std::rc::Rc;

    /// Rows are `(id, grp, ord)` with pk = [0].
    fn row(id: i64, grp: i64, ord: i64) -> Row {
        owned_row(vec![V::Int(id), V::Int(grp), V::Int(ord)])
    }

    fn delta() -> BatchDelta {
        let d = BatchDelta::new(vec![0]);
        d.begin();
        d
    }

    /// `(grp asc, ord asc, id asc)` — a connection sort including the pk.
    fn grp_ord_sort() -> Sort {
        vec![(1, true), (2, true), (0, true)]
    }

    fn ids(rows: &[Row]) -> Vec<i64> {
        rows.iter()
            .map(|r| match r.col(0) {
                rindle::value::Value::Int(i) => i,
                other => panic!("non-int id {other:?}"),
            })
            .collect()
    }

    /// Assert every tree's structural invariants — free to call, and the cheapest
    /// guard on the §3.1 stale-secondary trap.
    fn check(d: &BatchDelta) {
        let inner = d.inner.borrow();
        inner.touched.check_invariants(&inner.pk_sort).unwrap();
        inner.primary.check_invariants(&inner.pk_sort).unwrap();
        for (s, tree) in &inner.secondary {
            tree.check_invariants(s).unwrap();
            // Every secondary must hold exactly primary's rows.
            assert_eq!(
                tree.len(),
                inner.primary.len(),
                "secondary/primary size drift"
            );
        }
        // primary ⊆ touched under pk equality.
        let mut cur = inner.primary.values_from(None, true, &inner.pk_sort);
        while let Some(r) = cur.next_row() {
            assert!(
                inner.touched.has(r, &inner.pk_sort),
                "primary row not in touched"
            );
        }
    }

    fn apply(d: &BatchDelta, c: SourceChange) {
        d.apply(&c).unwrap();
        check(d);
    }

    #[test]
    fn fold_add_remove_edit() {
        let d = delta();
        assert!(d.is_active());
        assert!(d.is_empty());

        apply(&d, SourceChange::Add(row(1, 10, 100)));
        assert!(!d.is_empty());
        assert!(matches!(
            d.lookup_pk(&row(1, 0, 0)),
            DeltaLookup::Present(_)
        ));
        assert!(d.suppresses(&row(1, 99, 99)));
        assert!(!d.suppresses(&row(2, 10, 100)));

        // Remove a pure base row: pk becomes touched-and-absent.
        apply(&d, SourceChange::Remove(row(2, 10, 100)));
        assert!(matches!(d.lookup_pk(&row(2, 0, 0)), DeltaLookup::Absent));
        assert!(d.suppresses(&row(2, 10, 100)));

        // Edit a pure base row: touched, present with the new content.
        apply(
            &d,
            SourceChange::Edit {
                row: row(3, 11, 300),
                old: row(3, 10, 100),
            },
        );
        match d.lookup_pk(&row(3, 0, 0)) {
            DeltaLookup::Present(r) => assert_eq!(ids(std::slice::from_ref(&r)), vec![3]),
            other => panic!("expected Present, got {other:?}"),
        }

        assert!(matches!(d.lookup_pk(&row(4, 0, 0)), DeltaLookup::Untouched));
    }

    /// The §3.1 trap, named: after edit-then-edit the secondary must hold exactly ONE
    /// version of the row — deleting by the incoming `old` (the original base row)
    /// instead of the stored version would leave the intermediate version behind and
    /// ordered fetches would emit the pk twice.
    #[test]
    fn edit_then_edit_leaves_no_stale_secondary() {
        let sort = grp_ord_sort();
        // Two variants of the second edit's `old`: the previous derived version (what
        // a live capture stream reports) and the ORIGINAL base row (a stale `old` with
        // the right pk — the exact §3.1 divergence). Deleting by the stored row makes
        // both correct; deleting by the incoming `old` breaks the second.
        let base = row(1, 10, 100);
        let v1 = row(1, 20, 200);
        let v2 = row(1, 30, 300);
        for stale_old in [v1.clone(), base.clone()] {
            let d = delta();
            // Materialize the secondary FIRST so the fold must maintain it eagerly.
            assert!(d.fetch_vec(&FetchRequest::all(), &sort, None).is_empty());
            apply(
                &d,
                SourceChange::Edit {
                    row: v1.clone(),
                    old: base.clone(),
                },
            );
            apply(
                &d,
                SourceChange::Edit {
                    row: v2.clone(),
                    old: stale_old,
                },
            );

            let got = d.fetch_vec(&FetchRequest::all(), &sort, None);
            assert_eq!(got.len(), 1, "stale secondary entry: {}", got.len());
            assert_eq!(ids(&got), vec![1]);
            match d.lookup_pk(&row(1, 0, 0)) {
                DeltaLookup::Present(r) => {
                    assert!(matches!(r.col(1), rindle::value::Value::Int(30)))
                }
                other => panic!("expected Present, got {other:?}"),
            }
        }
    }

    #[test]
    fn edit_then_delete_and_delete_then_insert() {
        let d = delta();
        let sort = grp_ord_sort();
        assert!(d.fetch_vec(&FetchRequest::all(), &sort, None).is_empty());

        // edit-then-delete: the delete's `old` is the base version.
        let base = row(1, 10, 100);
        apply(
            &d,
            SourceChange::Edit {
                row: row(1, 20, 200),
                old: base.clone(),
            },
        );
        apply(&d, SourceChange::Remove(base));
        assert!(matches!(d.lookup_pk(&row(1, 0, 0)), DeltaLookup::Absent));
        assert!(d.fetch_vec(&FetchRequest::all(), &sort, None).is_empty());

        // delete-then-insert (same pk, new content).
        apply(&d, SourceChange::Remove(row(2, 10, 100)));
        apply(&d, SourceChange::Add(row(2, 50, 500)));
        let got = d.fetch_vec(&FetchRequest::all(), &sort, None);
        assert_eq!(ids(&got), vec![2]);
        assert!(matches!(got[0].col(1), rindle::value::Value::Int(50)));
    }

    /// A sort-key-changing edit must MOVE the row in the secondary, not duplicate it.
    #[test]
    fn sort_key_changing_edit_moves_in_secondary() {
        let d = delta();
        let sort = grp_ord_sort();
        apply(&d, SourceChange::Add(row(1, 10, 100)));
        apply(&d, SourceChange::Add(row(2, 20, 200)));
        // Build the index between changes, then move row 1 past row 2.
        assert_eq!(
            ids(&d.fetch_vec(&FetchRequest::all(), &sort, None)),
            vec![1, 2]
        );
        apply(
            &d,
            SourceChange::Edit {
                row: row(1, 30, 300),
                old: row(1, 10, 100),
            },
        );
        assert_eq!(
            ids(&d.fetch_vec(&FetchRequest::all(), &sort, None)),
            vec![2, 1]
        );
    }

    /// A lazily-built index must agree with one maintained eagerly across the same fold.
    #[test]
    fn lazy_index_agrees_with_eager() {
        let changes = |d: &BatchDelta| {
            apply(d, SourceChange::Add(row(3, 30, 1)));
            apply(d, SourceChange::Add(row(1, 10, 3)));
            apply(
                d,
                SourceChange::Edit {
                    row: row(3, 5, 2),
                    old: row(3, 30, 1),
                },
            );
            apply(d, SourceChange::Remove(row(2, 20, 2)));
            apply(d, SourceChange::Add(row(4, 10, 0)));
        };
        let sort = grp_ord_sort();

        // Eager: index exists before any change.
        let eager = delta();
        assert!(eager
            .fetch_vec(&FetchRequest::all(), &sort, None)
            .is_empty());
        changes(&eager);

        // Lazy: index built only at the end.
        let lazy = delta();
        changes(&lazy);

        let e = eager.fetch_vec(&FetchRequest::all(), &sort, None);
        let l = lazy.fetch_vec(&FetchRequest::all(), &sort, None);
        assert_eq!(ids(&e), ids(&l));
        // grp asc, ord asc: (5,2)=3, (10,0)=4, (10,3)=1.
        assert_eq!(ids(&e), vec![3, 4, 1]);
    }

    #[test]
    fn rows_for_fetch_constraint_and_reverse() {
        let d = delta();
        let sort = grp_ord_sort();
        apply(&d, SourceChange::Add(row(1, 10, 3)));
        apply(&d, SourceChange::Add(row(2, 10, 1)));
        apply(&d, SourceChange::Add(row(3, 20, 2)));

        // Constraint on grp: only the group, in connection-sort order.
        let req = FetchRequest::with_constraint(vec![(1, V::Int(10))]);
        assert_eq!(ids(&d.fetch_vec(&req, &sort, None)), vec![2, 1]);

        // Reverse.
        let req_rev = FetchRequest {
            constraint: Some(vec![(1, V::Int(10))]),
            reverse: true,
            ..Default::default()
        };
        assert_eq!(ids(&d.fetch_vec(&req_rev, &sort, None)), vec![1, 2]);

        // Non-matching constraint: empty.
        let req_none = FetchRequest::with_constraint(vec![(1, V::Int(99))]);
        assert!(d.fetch_vec(&req_none, &sort, None).is_empty());

        // Fully-constrained single-column pk: the index skips the connection sort.
        let req_pk = FetchRequest::with_constraint(vec![(0, V::Int(2))]);
        assert_eq!(ids(&d.fetch_vec(&req_pk, &sort, None)), vec![2]);
    }

    #[test]
    fn rows_for_fetch_multi_constraint_and_predicate() {
        let d = delta();
        let sort = grp_ord_sort();
        apply(&d, SourceChange::Add(row(1, 10, 1)));
        apply(&d, SourceChange::Add(row(2, 20, 2)));
        apply(&d, SourceChange::Add(row(3, 30, 3)));

        // multiConstraint: grp IN (10, 30) — bare-sort index, linear filter.
        let req = FetchRequest {
            multi_constraints: vec![vec![vec![(1, V::Int(10))], vec![(1, V::Int(30))]]],
            ..Default::default()
        };
        assert_eq!(ids(&d.fetch_vec(&req, &sort, None)), vec![1, 3]);

        // An empty multiConstraint entry constrains nothing.
        let req_empty = FetchRequest {
            multi_constraints: vec![vec![]],
            ..Default::default()
        };
        assert_eq!(ids(&d.fetch_vec(&req_empty, &sort, None)), vec![1, 2, 3]);

        // Predicate narrows delta rows (§2.1 — the overlay rule).
        let p: RowPredicate = Rc::new(|r: &Row| !matches!(r.col(1), rindle::value::Value::Int(20)));
        assert_eq!(
            ids(&d.fetch_vec(&FetchRequest::all(), &sort, Some(&p))),
            vec![1, 3]
        );
    }

    #[test]
    fn rows_for_fetch_start_seek_is_inclusive_and_relaxed() {
        let d = delta();
        let sort = grp_ord_sort();
        apply(&d, SourceChange::Add(row(1, 10, 1)));
        apply(&d, SourceChange::Add(row(2, 20, 2)));
        apply(&d, SourceChange::Add(row(3, 30, 3)));

        // Start at row 2 (Basis::After): the seek is inclusive — row 2 is still
        // returned; the exact After cut is generate_with_start's job downstream.
        let req = FetchRequest {
            start: Some(Start {
                row: row(2, 20, 2),
                basis: Basis::After,
            }),
            ..Default::default()
        };
        assert_eq!(ids(&d.fetch_vec(&req, &sort, None)), vec![2, 3]);
    }

    #[test]
    fn budget_overflow_sheds_typed_sticky_and_dropped() {
        // Start unbounded and let the delta tell us what two rows cost, so the test
        // pins the *mechanism* and not the current value of TREE_ENTRY_BYTES.
        let d = BatchDelta::with_max_bytes(vec![0], usize::MAX);
        d.begin();
        d.apply(&SourceChange::Add(row(1, 1, 1))).unwrap();
        d.apply(&SourceChange::Add(row(2, 2, 2))).unwrap();
        let two_rows = d.used_bytes();
        assert!(two_rows > 0, "a live delta charges the budget");

        // Re-touching an already-counted pk is net-zero — the charge tracks what is
        // HELD, not how many changes flowed through.
        d.apply(&SourceChange::Edit {
            row: row(2, 9, 9),
            old: row(2, 2, 2),
        })
        .unwrap();
        assert_eq!(d.used_bytes(), two_rows, "a same-size re-touch is net-zero");

        // Pinned at exactly what it is holding, a third distinct pk trips the typed D4
        // shed signal.
        d.budget().set_max_bytes(two_rows);
        let err = d.apply(&SourceChange::Add(row(3, 3, 3))).unwrap_err();
        assert!(matches!(err, RindleError::DeltaOverflow { max_bytes, used }
                     if max_bytes == two_rows && used > two_rows));
        // Overflow DROPS the accumulated rows AND their charge — a delta that shed but
        // kept owing would keep every peer sharing the budget shed forever.
        assert!(d.is_empty());
        assert_eq!(d.used_bytes(), 0);
        assert_eq!(d.budget().used(), 0);
        // …and is sticky: further applies re-raise without accumulating.
        let err = d.apply(&SourceChange::Add(row(4, 4, 4))).unwrap_err();
        assert!(matches!(err, RindleError::DeltaOverflow { .. }));
        assert!(d.is_empty());
        assert_eq!(d.budget().used(), 0);
        // The next derivation starts clean.
        d.end();
        d.budget().set_max_bytes(usize::MAX);
        d.begin();
        d.apply(&SourceChange::Add(row(5, 5, 5))).unwrap();
        assert!(!d.is_empty());
        // The ceiling is re-tunable on the shared handle, mid-derivation.
        d.budget().set_max_bytes(1);
        let err = d.apply(&SourceChange::Add(row(6, 6, 6))).unwrap_err();
        assert!(matches!(
            err,
            RindleError::DeltaOverflow { max_bytes: 1, .. }
        ));
    }

    /// The budget is the TRANSACTION's, not each table's: deltas sharing one
    /// `DeltaBudget` charge one pool, and each hands back only its own share.
    #[test]
    fn a_shared_budget_bounds_every_table_together() {
        let budget = Rc::new(DeltaBudget::new(usize::MAX));
        let a = BatchDelta::with_budget(vec![0], Rc::clone(&budget));
        let b = BatchDelta::with_budget(vec![0], Rc::clone(&budget));
        a.begin();
        b.begin();

        a.apply(&SourceChange::Add(row(1, 1, 1))).unwrap();
        let one_row = budget.used();
        b.apply(&SourceChange::Add(row(1, 1, 1))).unwrap();
        assert_eq!(
            budget.used(),
            2 * one_row,
            "both tables charge the same pool"
        );
        assert_eq!(a.used_bytes(), one_row, "…and each knows its own share");
        assert_eq!(b.used_bytes(), one_row);

        // Pinned at the pool's current total, the NEXT row on either table sheds —
        // which is the point: 20 tables cannot hold 20x the ceiling.
        budget.set_max_bytes(budget.used());
        assert!(b.apply(&SourceChange::Add(row(2, 2, 2))).is_err());
        // b shed its own rows; a is untouched and still owes exactly its share.
        assert!(b.is_empty());
        assert!(!a.is_empty());
        assert_eq!(budget.used(), one_row);
        assert_eq!(a.used_bytes(), one_row);

        // Ending a derivation returns that delta's share and nothing else.
        a.end();
        assert_eq!(budget.used(), 0);
    }

    /// A wide row costs more of the budget than a narrow one — the whole reason this
    /// is bytes and not a row count.
    #[test]
    fn wide_rows_cost_more_than_narrow_ones() {
        let narrow = BatchDelta::with_max_bytes(vec![0], usize::MAX);
        narrow.begin();
        narrow.apply(&SourceChange::Add(row(1, 1, 1))).unwrap();

        let wide = BatchDelta::with_max_bytes(vec![0], usize::MAX);
        wide.begin();
        wide.apply(&SourceChange::Add(owned_row(vec![
            V::Int(1),
            V::Int(1),
            V::Str("x".repeat(4096).into()),
        ])))
        .unwrap();

        assert!(
            wide.used_bytes() > narrow.used_bytes() + 4000,
            "a 4 KiB row must cost ~4 KiB more than a 24-byte one: \
             wide={} narrow={}",
            wide.used_bytes(),
            narrow.used_bytes()
        );
    }

    /// A duplicate ADD is a loud consistency error in every build — the parity
    /// replacement for the UNIQUE-constraint fault the write-then-abort path raised —
    /// never a silent overwrite that double-counts the row.
    #[test]
    fn duplicate_add_is_a_consistency_error() {
        let d = delta();
        apply(&d, SourceChange::Add(row(1, 10, 100)));
        let err = d.apply(&SourceChange::Add(row(1, 20, 200))).unwrap_err();
        assert!(matches!(err, RindleError::ConsistencyViolation { .. }));
        // delete-then-insert of the same pk stays legal (the pk has no stored content).
        apply(&d, SourceChange::Remove(row(2, 10, 100)));
        apply(&d, SourceChange::Add(row(2, 50, 500)));
    }

    /// A pk-changing Edit folds as Remove(old) + Add(row) — the same normalization
    /// `Engine::push` applies upstream — so BOTH pks are touched: the old base row is
    /// suppressed from merged scans and the new content is present. (This used to be
    /// a debug_assert whose release fallthrough left the old base row visible.)
    #[test]
    fn pk_changing_edit_folds_as_remove_plus_add() {
        let d = delta();
        let sort = grp_ord_sort();
        // Materialize the secondary FIRST so the fold must maintain it eagerly.
        assert!(d.fetch_vec(&FetchRequest::all(), &sort, None).is_empty());
        apply(
            &d,
            SourceChange::Edit {
                row: row(2, 20, 200),
                old: row(1, 10, 100),
            },
        );
        // The OLD pk is touched-and-absent (its base row must be suppressed)…
        assert!(d.suppresses(&row(1, 10, 100)));
        assert!(matches!(d.lookup_pk(&row(1, 0, 0)), DeltaLookup::Absent));
        // …and the new pk carries the content.
        assert!(matches!(
            d.lookup_pk(&row(2, 0, 0)),
            DeltaLookup::Present(_)
        ));
        let got = d.fetch_vec(&FetchRequest::all(), &sort, None);
        assert_eq!(ids(&got), vec![2]);
    }

    #[test]
    fn apply_outside_derivation_is_an_error() {
        let d = BatchDelta::new(vec![0]);
        assert!(!d.is_active());
        let err = d.apply(&SourceChange::Add(row(1, 1, 1))).unwrap_err();
        assert!(matches!(err, RindleError::Storage(_)));
    }

    /// The inactive/empty short-circuits: every read answers without building state.
    #[test]
    fn inactive_and_empty_short_circuits() {
        let d = BatchDelta::new(vec![0]);
        let sort = grp_ord_sort();
        assert!(matches!(d.lookup_pk(&row(1, 0, 0)), DeltaLookup::Untouched));
        assert!(!d.suppresses(&row(1, 0, 0)));
        assert!(d.fetch_vec(&FetchRequest::all(), &sort, None).is_empty());
        assert!(d.find_by_key(&[(0, V::Int(1))]).is_none());
        // No secondary index was materialized by any of the above.
        assert!(d.inner.borrow().secondary.is_empty());

        // begin → end clears state and deactivates.
        d.begin();
        d.apply(&SourceChange::Add(row(1, 1, 1))).unwrap();
        assert!(!d.is_empty());
        d.end();
        assert!(!d.is_active());
        assert!(d.is_empty());
        assert!(matches!(d.lookup_pk(&row(1, 0, 0)), DeltaLookup::Untouched));
    }

    /// `begin` drops the secondary MAP (plan §3.3), so M is bounded by the sorts
    /// fetched during the current transaction, not all history.
    #[test]
    fn begin_drops_secondary_indexes() {
        let d = delta();
        let sort = grp_ord_sort();
        apply(&d, SourceChange::Add(row(1, 1, 1)));
        assert_eq!(
            ids(&d.fetch_vec(&FetchRequest::all(), &sort, None)),
            vec![1]
        );
        assert_eq!(d.inner.borrow().secondary.len(), 1);
        d.begin();
        assert!(d.inner.borrow().secondary.is_empty());
    }

    #[test]
    fn find_by_key_prefers_delta_content() {
        let d = delta();
        apply(&d, SourceChange::Add(row(1, 10, 100)));
        // pk key.
        assert!(d.find_by_key(&[(0, V::Int(1))]).is_some());
        // non-pk (unique-ish) key over delta content.
        let hit = d.find_by_key(&[(1, V::Int(10)), (2, V::Int(100))]).unwrap();
        assert_eq!(ids(&[hit]), vec![1]);
        assert!(d.find_by_key(&[(1, V::Int(99))]).is_none());
    }
}