rindle/

push_index.rs

//! Guarded push fan-out: a reverse predicate index over source connections
//! (`designs/205-GUARDED-PUSH-FANOUT-DESIGN.md`).
//!
//! Every source write today fans out to **every** connection on the table and runs
//! its full compiled predicate against the changed row(s) — O(N) predicate
//! evaluations per write, N = connection slots, almost all concluding "drop" for
//! the `where col = ?` shapes that dominate `rindled`. This module is the
//! *conservative index* that prunes that fan-out to the connections that could
//! actually match.
//!
//! ## Why skipping is exact, not heuristic
//!
//! The index is only ever a **superset** filter — it may over-approximate (visit a
//! connection whose predicate then rejects the row), never under-approximate. Two
//! facts (design §"safety argument") make that sound:
//!
//! 1. `filter_push` remains the exact gate: a
//!    connection contributes nothing iff its predicate rejects **both** the old and
//!    the new row. So the index need only never *miss* a possibly-matching
//!    connection; a false positive re-runs the exact predicate, unchanged.
//! 2. Skipping is observationally equivalent for the reentrancy machinery — a
//!    skipped connection's stale `last_pushed_epoch` hides the overlay exactly as
//!    the connection's own predicate would have narrowed that overlay to nothing.
//!
//! Corollary: **every approximation here errs only toward visiting.** A guard-less
//! or un-guardable connection lands in the always-visited `scan` list; a
//! coarser-than-identity `GuardKey` bucket merely re-checks the exact predicate.
//! The one thing that would be a bug is a *finer* key.
//!
//! This module is intentionally standalone and free of the push orchestration: it
//! is pure data (`PushGuard`), a comparator newtype (`GuardKey`), and the index
//! (`PushIndex`). The extraction of a `PushGuard` from a query's `where` tree
//! lives in `crate::builder`; the wiring of a `PushIndex` into `ConnTable`'s
//! push path lives in [`crate::source_common`].

use std::cmp::Ordering;
use std::collections::{BTreeMap, BTreeSet};

use crate::change::SourceChange;
use crate::predicate::compare_predicate_values;
use crate::value::{ColId, OwnedValue, Value};

/// `predicate(row) ⇒ row[col] ∈ values` — a finite single-column implication of a
/// connection's pushed-down **equality-shaped** filter, extracted (in
/// `crate::builder`) from the same stripped condition the connection's
/// [`RowPredicate`](crate::source_common::RowPredicate) was compiled from.
/// Backend-neutral pure data (like [`SqlCondition`](crate::source_common::SqlCondition)),
/// populated for both the memory and SQLite leaves.
///
/// `values` is deduplicated by extraction and is **non-empty in practice** — an
/// empty set is the legitimate encoding of a *never-matching* predicate (an empty
/// `IN`, or `col = NULL` which SQL folds to false); such a guard indexes its slot
/// **nowhere** (not even the `scan` list), which is exact because the
/// connection can receive no pushes.
/// (No derived `PartialEq`: [`OwnedValue`] deliberately has none — the engine's
/// dual-comparator convention forces callers to pick `values_identical` vs
/// `values_equal` explicitly. Compare guards field-wise in tests.)
#[derive(Clone, Debug)]
pub struct PushGuard {
    pub col: ColId,
    pub values: Vec<OwnedValue>,
}

/// A [`BTreeMap`] key over an [`OwnedValue`](crate::value::OwnedValue) whose ordering is
/// [`compare_predicate_values`] — the SAME total order the `=` / `IN` predicate
/// identity ([`values_identical`](crate::value::values_identical)) collapses under
/// (`Null == Null`; `Int`/`Float` widen through `total_cmp`-over-`f64`).
///
/// **Coarsening direction is load-bearing.** The map's equality must never
/// *separate* two values that `values_identical` equates — else a row cell that the
/// predicate treats as identical to a guard literal could miss that literal's
/// bucket and wrongly skip a matching connection (a dropped delta). Coarser is
/// safe: two distinct large `Int`s that widen to one `f64` share a bucket, a false
/// positive the exact predicate re-rejects. `compare_predicate_values` is exactly
/// as coarse as `values_identical` on every equal pair and strictly finer nowhere
/// identity holds (checked pairwise by the property test below), so it is the right
/// key.
#[derive(Clone, Debug)]
struct GuardKey(OwnedValue);

impl Ord for GuardKey {
    #[inline]
    fn cmp(&self, other: &Self) -> Ordering {
        compare_predicate_values(self.0.as_ref(), other.0.as_ref())
    }
}
impl PartialOrd for GuardKey {
    #[inline]
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}
impl Eq for GuardKey {}
impl PartialEq for GuardKey {
    #[inline]
    fn eq(&self, other: &Self) -> bool {
        self.cmp(other) == Ordering::Equal
    }
}

/// The reverse index: guard value → connection slots, plus the always-visited scan
/// set and a refcounted union of split-edit keys.
///
/// A connection is registered in **exactly one place**: under its guard's column
/// (once per guard value) in the `eq` map, or in the `scan` set when it
/// has no extractable guard — *unless* its guard has empty `values` (never-matching),
/// in which case it is registered nowhere. [`insert`](Self::insert) /
/// [`remove`](Self::remove) are symmetric so the invariant survives slot recycling.
#[derive(Default, Debug)]
#[cfg_attr(test, derive(Clone))]
pub struct PushIndex {
    /// col → guard value → slots guarded on `(col, value)`.
    eq: BTreeMap<ColId, BTreeMap<GuardKey, Vec<u32>>>,
    /// Slots with no extractable guard (no filter, presence-only, un-guardable
    /// shape). Always a push candidate — these keep today's behavior verbatim.
    scan: BTreeSet<u32>,
    /// Refcounted union of every registered connection's `split_edit_keys`: a key
    /// is present iff some live slot lists it. Replaces the per-Edit O(N·keys) scan
    /// with an O(distinct keys) check.
    split_keys: BTreeMap<ColId, u32>,
    /// Slots whose guard values change at **runtime** — a parameterized query family's
    /// root connection (design 310 §4.1, the 205 extension): slot → (col, the values
    /// currently added, one entry per [`add_guard_value`](Self::add_guard_value) call,
    /// so two bindings sharing a first-column value refcount naturally). The static
    /// [`PushGuard`] of such a slot has EMPTY values (indexed nowhere at `insert`), so
    /// [`remove`](Self::remove) — driven by that static guard — drains this map too and
    /// leaves no stale bucket behind at `destroy`.
    dynamic: BTreeMap<u32, (ColId, Vec<OwnedValue>)>,
}

impl PushIndex {
    pub fn new() -> PushIndex {
        PushIndex::default()
    }

    /// Register `slot` under `guard` (or the scan set when `None`), and refcount its
    /// `split_edit_keys` into the union. Idempotent per (slot, guard) pair only in
    /// the sense the caller enforces: `ConnTable` inserts once per `connect`.
    pub fn insert(&mut self, slot: u32, guard: Option<&PushGuard>, split_edit_keys: &[ColId]) {
        match guard {
            // Never-matching guard (empty `IN`, `col = NULL`): index nowhere. The
            // connection can receive no pushes, so skipping it always is exact.
            Some(g) if g.values.is_empty() => {}
            Some(g) => {
                let col_map = self.eq.entry(g.col).or_default();
                for v in &g.values {
                    col_map.entry(GuardKey(v.clone())).or_default().push(slot);
                }
            }
            None => {
                self.scan.insert(slot);
            }
        }
        for &k in split_edit_keys {
            *self.split_keys.entry(k).or_insert(0) += 1;
        }
    }

    /// Un-register `slot` — the exact inverse of [`insert`](Self::insert), driven by
    /// the connection's own still-present `guard` (so it names precisely which
    /// buckets to remove it from). Empty buckets and column maps are pruned so the
    /// per-push column iteration stays proportional to *live* guarded columns. Any
    /// **dynamic** guard values the slot accumulated are drained as well.
    pub fn remove(&mut self, slot: u32, guard: Option<&PushGuard>, split_edit_keys: &[ColId]) {
        match guard {
            Some(g) if g.values.is_empty() => {}
            Some(g) => {
                for v in &g.values {
                    self.unbucket(slot, g.col, v);
                }
            }
            None => {
                self.scan.remove(&slot);
            }
        }
        if let Some((col, values)) = self.dynamic.remove(&slot) {
            for v in &values {
                self.unbucket(slot, col, v);
            }
        }
        for &k in split_edit_keys {
            if let Some(rc) = self.split_keys.get_mut(&k) {
                *rc -= 1;
                if *rc == 0 {
                    self.split_keys.remove(&k);
                }
            }
        }
    }

    /// Add one **dynamic** guard value for `slot` on `col` (design 310 §4.1, the 205
    /// extension for a family root's growing binding set): the slot becomes a push
    /// candidate for writes whose `col` cell is identical to `v`. Repeated values
    /// stack (one bucket entry each) and un-stack one at a time, so two bindings that
    /// share a first-column value keep the slot indexed until both are gone.
    ///
    /// The 205 safety argument is unchanged: the index stays a superset filter and the
    /// exact predicate re-checks every visit — provided the caller adds the guard value
    /// **before** the predicate starts accepting the value, and removes it **after** the
    /// predicate stops (the ordering discipline `Graph::bind_family_partition` follows),
    /// and only ever between pushes.
    pub fn add_guard_value(&mut self, slot: u32, col: ColId, v: OwnedValue) {
        self.eq
            .entry(col)
            .or_default()
            .entry(GuardKey(v.clone()))
            .or_default()
            .push(slot);
        self.dynamic
            .entry(slot)
            .or_insert_with(|| (col, Vec::new()))
            .1
            .push(v);
    }

    /// Remove one dynamic guard value previously added with
    /// [`add_guard_value`](Self::add_guard_value) (one stacked entry). A value that was
    /// never added is a no-op.
    pub fn remove_guard_value(&mut self, slot: u32, col: ColId, v: &OwnedValue) {
        let Some((dcol, values)) = self.dynamic.get_mut(&slot) else {
            return;
        };
        debug_assert_eq!(
            *dcol, col,
            "a slot's dynamic guard values live on one column"
        );
        let Some(pos) = values
            .iter()
            .position(|x| GuardKey(x.clone()) == GuardKey(v.clone()))
        else {
            return;
        };
        values.swap_remove(pos);
        if values.is_empty() {
            self.dynamic.remove(&slot);
        }
        self.unbucket(slot, col, v);
    }

    /// Remove ONE appearance of `slot` from the `(col, v)` bucket, pruning the bucket
    /// and the column map when they empty.
    fn unbucket(&mut self, slot: u32, col: ColId, v: &OwnedValue) {
        if let Some(col_map) = self.eq.get_mut(&col) {
            let key = GuardKey(v.clone());
            if let Some(bucket) = col_map.get_mut(&key) {
                if let Some(pos) = bucket.iter().position(|&s| s == slot) {
                    // Order within a bucket is irrelevant (candidates are sorted
                    // before the fan-out); swap_remove is O(1).
                    bucket.swap_remove(pos);
                }
                if bucket.is_empty() {
                    col_map.remove(&key);
                }
            }
            if col_map.is_empty() {
                self.eq.remove(&col);
            }
        }
    }

    /// The number of entries the index holds — bucket memberships, scan slots, and
    /// dynamic values (a dynamic value counts twice: its bucket membership and its
    /// dynamic-map entry) — a churn/leak probe's size signal.
    pub fn size(&self) -> usize {
        let buckets: usize = self
            .eq
            .values()
            .flat_map(|m| m.values())
            .map(|b| b.len())
            .sum();
        let dynamic: usize = self.dynamic.values().map(|(_, v)| v.len()).sum();
        buckets + self.scan.len() + dynamic
    }

    /// Slots that may satisfy `predicate(old) || predicate(new)` for `change` — a
    /// **superset** by construction (module docs). Sorted ascending and deduped, so
    /// the fan-out visits in slot order exactly as the full-scan loop does today.
    ///
    /// Starts from every `scan` slot, then for each guarded column looks up the
    /// change's relevant cell(s): `Add`/`Remove` contribute the one row's cell;
    /// `Edit` contributes both `row[col]` and `old[col]` (an edit can leave one
    /// guard set and enter another).
    pub fn push_candidates(&self, change: &SourceChange) -> Vec<u32> {
        let mut out: Vec<u32> = self.scan.iter().copied().collect();
        for (&col, buckets) in &self.eq {
            match change {
                SourceChange::Add(r) | SourceChange::Remove(r) => {
                    Self::collect_bucket(buckets, r.col(col), &mut out);
                }
                SourceChange::Edit { row, old } => {
                    Self::collect_bucket(buckets, row.col(col), &mut out);
                    Self::collect_bucket(buckets, old.col(col), &mut out);
                }
            }
        }
        out.sort_unstable();
        out.dedup();
        out
    }

    /// Look up `cell`'s bucket in one column's map and append its slots.
    ///
    /// `to_owned` materializes the lookup key: a no-op copy for scalar cells; one
    /// `Arc` allocation for `Str`/`Json`. (An `Absent` cell — partial union row —
    /// finds no bucket, which is correct: guard literals are never `Absent`, and
    /// every guard-eligible leaf rejects an absent cell.)
    #[inline]
    fn collect_bucket(buckets: &BTreeMap<GuardKey, Vec<u32>>, cell: Value<'_>, out: &mut Vec<u32>) {
        if let Some(slots) = buckets.get(&GuardKey(cell.to_owned())) {
            out.extend_from_slice(slots);
        }
    }

    /// The distinct split-edit keys across all registered connections. `should_split`
    /// is `split_keys().any(|k| !values_equal(row[k], old[k]))` — exactly equivalent
    /// to today's `any`-of-`any` over per-connection key lists, at O(distinct keys).
    pub fn split_keys(&self) -> impl Iterator<Item = ColId> + '_ {
        self.split_keys.keys().copied()
    }

    /// `true` iff no connection lists any split-edit key — lets the caller skip the
    /// edit-split check entirely.
    pub fn has_split_keys(&self) -> bool {
        !self.split_keys.is_empty()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::value::{owned_row, values_identical, OwnedRow};

    fn iv(n: i64) -> OwnedValue {
        OwnedValue::Int(n)
    }
    fn add(row: OwnedRow) -> SourceChange {
        SourceChange::Add(row)
    }
    fn guard(col: ColId, values: Vec<OwnedValue>) -> PushGuard {
        PushGuard { col, values }
    }

    // -- GuardKey coarseness: the never-finer property (design §Testing.3) --------

    /// The corpus of "identical-under-predicate" pairs and "distinct" values that
    /// exercise every arm of `values_identical` / `compare_predicate_values`.
    fn value_corpus() -> Vec<OwnedValue> {
        vec![
            OwnedValue::Null,
            OwnedValue::Bool(false),
            OwnedValue::Bool(true),
            OwnedValue::Int(0),
            OwnedValue::Int(7),
            OwnedValue::Float(7.0), // identical to Int(7) under the predicate domain
            OwnedValue::Float(7.5),
            OwnedValue::Float(f64::NAN),
            OwnedValue::str("7"),
            OwnedValue::str("apple"),
            OwnedValue::Json("{}".into()),
        ]
    }

    #[test]
    fn guardkey_never_finer_than_values_identical() {
        // The one direction that would be a correctness bug: two values the predicate
        // treats as identical MUST share a GuardKey bucket. (Coarser is fine.)
        let corpus = value_corpus();
        for a in &corpus {
            for b in &corpus {
                if values_identical(a.as_ref(), b.as_ref()) {
                    assert_eq!(
                        GuardKey(a.clone()),
                        GuardKey(b.clone()),
                        "values_identical({a:?}, {b:?}) but GuardKeys differ — would drop a delta",
                    );
                }
            }
        }
    }

    #[test]
    fn guardkey_int_float_widening_shares_bucket() {
        // The exact false-positive-tolerant coarsening the design relies on.
        let ki = GuardKey(OwnedValue::Int(7));
        let kf = GuardKey(OwnedValue::Float(7.0));
        assert_eq!(ki, kf);
        let mut m: BTreeMap<GuardKey, u32> = BTreeMap::new();
        m.insert(GuardKey(OwnedValue::Int(7)), 1);
        assert_eq!(m.get(&GuardKey(OwnedValue::Float(7.0))), Some(&1));
    }

    #[test]
    fn guardkey_null_is_its_own_bucket() {
        // Predicate identity is null==null (unlike join equality), so a `col IS NULL`
        // guard and a null row cell must meet in the same bucket.
        assert_eq!(GuardKey(OwnedValue::Null), GuardKey(OwnedValue::Null));
        assert_ne!(GuardKey(OwnedValue::Null), GuardKey(OwnedValue::Int(0)));
    }

    // -- PushIndex maintenance & the exactly-once invariant (design §Testing.2) ----

    /// Count every appearance of `slot` across `eq` buckets + `scan`. The invariant
    /// is: a live slot appears in exactly one *place* (its guard column, once per
    /// guard value) or in `scan`; a never-matching guard appears nowhere.
    fn places(idx: &PushIndex, slot: u32) -> (usize, bool) {
        let eq_hits: usize = idx
            .eq
            .values()
            .flat_map(|m| m.values())
            .flat_map(|v| v.iter())
            .filter(|&&s| s == slot)
            .count();
        (eq_hits, idx.scan.contains(&slot))
    }

    #[test]
    fn insert_guarded_then_remove_is_empty() {
        let mut idx = PushIndex::new();
        let g = guard(0, vec![iv(42)]);
        idx.insert(3, Some(&g), &[]);
        assert_eq!(places(&idx, 3), (1, false));
        idx.remove(3, Some(&g), &[]);
        assert_eq!(places(&idx, 3), (0, false));
        // Column map pruned to empty.
        assert!(idx.eq.is_empty(), "empty column map should be pruned");
    }

    #[test]
    fn insert_scan_then_remove() {
        let mut idx = PushIndex::new();
        idx.insert(5, None, &[]);
        assert_eq!(places(&idx, 5), (0, true));
        idx.remove(5, None, &[]);
        assert_eq!(places(&idx, 5), (0, false));
    }

    #[test]
    fn multi_value_guard_registers_once_per_value() {
        let mut idx = PushIndex::new();
        let g = guard(1, vec![iv(1), iv(2), iv(3)]); // col IN (1,2,3)
        idx.insert(9, Some(&g), &[]);
        let hits = places(&idx, 9).0;
        assert_eq!(hits, 3, "one appearance per guard value");
        idx.remove(9, Some(&g), &[]);
        assert_eq!(places(&idx, 9), (0, false));
        assert!(idx.eq.is_empty());
    }

    #[test]
    fn empty_values_guard_indexes_nowhere() {
        // `col = NULL` / `col IN ()` fold to never-matching: registered nowhere.
        let mut idx = PushIndex::new();
        let g = guard(0, vec![]);
        idx.insert(2, Some(&g), &[]);
        assert_eq!(places(&idx, 2), (0, false));
        assert!(idx.eq.is_empty() && idx.scan.is_empty());
        // Remove is a clean no-op.
        idx.remove(2, Some(&g), &[]);
        assert_eq!(places(&idx, 2), (0, false));
    }

    #[test]
    fn recycle_slot_reassigns_registration() {
        // destroy un-indexes -> slot recycled by a later connect under a new guard.
        let mut idx = PushIndex::new();
        let g_a = guard(0, vec![iv(10)]);
        idx.insert(4, Some(&g_a), &[]);
        idx.remove(4, Some(&g_a), &[]); // teardown
        let g_b = guard(1, vec![iv(20)]); // new tenant, different column
        idx.insert(4, Some(&g_b), &[]);
        assert_eq!(places(&idx, 4).0, 1);
        // It lives under the NEW column/value, not the old.
        assert!(idx.eq.get(&1).unwrap().contains_key(&GuardKey(iv(20))));
        assert!(!idx.eq.contains_key(&0), "old bucket fully pruned");
    }

    #[test]
    fn split_keys_refcount_drains_to_empty() {
        let mut idx = PushIndex::new();
        idx.insert(0, None, &[2, 5]);
        idx.insert(1, None, &[5]); // 5 now refcount 2
        let mut keys: Vec<_> = idx.split_keys().collect();
        keys.sort_unstable();
        assert_eq!(keys, vec![2, 5]);
        assert!(idx.has_split_keys());
        idx.remove(0, None, &[2, 5]); // 2 -> 0 (drop), 5 -> 1
        let keys: Vec<_> = idx.split_keys().collect();
        assert_eq!(keys, vec![5]);
        idx.remove(1, None, &[5]); // 5 -> 0 (drop)
        assert!(!idx.has_split_keys());
        assert_eq!(idx.split_keys().count(), 0);
    }

    // -- push_candidates: superset, sorted, deduped (design §Testing.2) -----------

    #[test]
    fn candidates_are_sorted_and_deduped() {
        let mut idx = PushIndex::new();
        // Two connections guarded on col 0 = 42, plus a scan connection.
        idx.insert(7, Some(&guard(0, vec![iv(42)])), &[]);
        idx.insert(3, Some(&guard(0, vec![iv(42)])), &[]);
        idx.insert(1, None, &[]);
        let cand = idx.push_candidates(&add(owned_row(vec![iv(42)])));
        assert_eq!(cand, vec![1, 3, 7], "scan + both guarded, sorted ascending");
    }

    #[test]
    fn candidates_exclude_non_matching_guard() {
        let mut idx = PushIndex::new();
        idx.insert(7, Some(&guard(0, vec![iv(42)])), &[]);
        idx.insert(1, None, &[]);
        // Write to a different value: only the scan connection is a candidate.
        let cand = idx.push_candidates(&add(owned_row(vec![iv(99)])));
        assert_eq!(cand, vec![1]);
    }

    #[test]
    fn candidates_edit_is_two_sided() {
        // An edit that moves col 0 from 42 -> 99 must visit BOTH the 42-guarded and
        // the 99-guarded connection (leaves one guard set, enters another).
        let mut idx = PushIndex::new();
        idx.insert(2, Some(&guard(0, vec![iv(42)])), &[]);
        idx.insert(5, Some(&guard(0, vec![iv(99)])), &[]);
        let change = SourceChange::Edit {
            row: owned_row(vec![iv(99)]),
            old: owned_row(vec![iv(42)]),
        };
        assert_eq!(idx.push_candidates(&change), vec![2, 5]);
    }

    #[test]
    fn candidates_edit_same_bucket_not_duplicated() {
        // Both sides of the edit hit the same bucket -> the slot appears once.
        let mut idx = PushIndex::new();
        idx.insert(4, Some(&guard(0, vec![iv(42)])), &[]);
        let change = SourceChange::Edit {
            row: owned_row(vec![iv(42), iv(1)]),
            old: owned_row(vec![iv(42), iv(2)]),
        };
        assert_eq!(idx.push_candidates(&change), vec![4]);
    }

    #[test]
    fn candidates_multi_column_guards() {
        // Guards on two different columns; a write hits only the matching column.
        let mut idx = PushIndex::new();
        idx.insert(1, Some(&guard(0, vec![iv(42)])), &[]); // projectId = 42
        idx.insert(2, Some(&guard(1, vec![iv(7)])), &[]); // ownerId = 7

        // Row: col0 = 42, col1 = 999 -> matches conn 1's guard only.
        let cand = idx.push_candidates(&add(owned_row(vec![iv(42), iv(999)])));
        assert_eq!(cand, vec![1]);
    }

    #[test]
    fn candidates_int_float_coarsening_is_a_superset() {
        // A guard registered under Int(7); a row cell of Float(7.0) must still be a
        // candidate (the predicate treats them identical).
        let mut idx = PushIndex::new();
        idx.insert(6, Some(&guard(0, vec![iv(7)])), &[]);
        let cand = idx.push_candidates(&add(owned_row(vec![OwnedValue::Float(7.0)])));
        assert_eq!(cand, vec![6]);
    }

    #[test]
    fn candidates_null_guard_matches_null_cell() {
        let mut idx = PushIndex::new();
        idx.insert(8, Some(&guard(0, vec![OwnedValue::Null])), &[]); // col IS NULL
        let cand = idx.push_candidates(&add(owned_row(vec![OwnedValue::Null])));
        assert_eq!(cand, vec![8]);
        // A non-null cell does not hit the null bucket.
        let miss = idx.push_candidates(&add(owned_row(vec![iv(1)])));
        assert!(miss.is_empty());
    }

    // -- dynamic guard values (design 310 §4.1 / S2) -------------------------------

    #[test]
    fn dynamic_values_index_and_unindex_one_at_a_time() {
        let mut idx = PushIndex::new();
        // A family root registers with an EMPTY static guard: indexed nowhere.
        let g = guard(0, vec![]);
        idx.insert(3, Some(&g), &[]);
        assert_eq!(places(&idx, 3), (0, false));
        assert!(idx.push_candidates(&add(owned_row(vec![iv(7)]))).is_empty());

        idx.add_guard_value(3, 0, iv(7));
        assert_eq!(places(&idx, 3), (1, false));
        assert_eq!(idx.push_candidates(&add(owned_row(vec![iv(7)]))), vec![3]);
        assert!(idx.push_candidates(&add(owned_row(vec![iv(8)]))).is_empty());
        // Int/Float coarsening holds for dynamic values too.
        assert_eq!(
            idx.push_candidates(&add(owned_row(vec![OwnedValue::Float(7.0)]))),
            vec![3]
        );

        idx.add_guard_value(3, 0, iv(9));
        assert_eq!(places(&idx, 3), (2, false));
        idx.remove_guard_value(3, 0, &iv(7));
        assert!(idx.push_candidates(&add(owned_row(vec![iv(7)]))).is_empty());
        assert_eq!(idx.push_candidates(&add(owned_row(vec![iv(9)]))), vec![3]);
        idx.remove_guard_value(3, 0, &iv(9));
        assert_eq!(places(&idx, 3), (0, false));
        assert!(
            idx.eq.is_empty(),
            "empty buckets and column maps are pruned"
        );
        assert!(idx.dynamic.is_empty());
        assert_eq!(idx.size(), 0);
        // Removing a value that was never added is a no-op.
        idx.remove_guard_value(3, 0, &iv(42));
    }

    #[test]
    fn repeated_dynamic_values_stack_and_unstack() {
        // Two bindings sharing a first-column value: the slot stays indexed until
        // BOTH are removed (a wrong "gone" here would drop a delta).
        let mut idx = PushIndex::new();
        idx.insert(5, Some(&guard(1, vec![])), &[]);
        idx.add_guard_value(5, 1, iv(1));
        idx.add_guard_value(5, 1, iv(1));
        assert_eq!(places(&idx, 5).0, 2);
        idx.remove_guard_value(5, 1, &iv(1));
        assert_eq!(
            idx.push_candidates(&add(owned_row(vec![iv(0), iv(1)]))),
            vec![5],
            "one binding still shares the value"
        );
        idx.remove_guard_value(5, 1, &iv(1));
        assert!(idx
            .push_candidates(&add(owned_row(vec![iv(0), iv(1)])))
            .is_empty());
    }

    #[test]
    fn remove_drains_dynamic_values_and_recycles_cleanly() {
        // `destroy` un-indexes from the STATIC (empty) guard; the dynamic map must be
        // drained too, or a recycled slot inherits stale buckets.
        let mut idx = PushIndex::new();
        let g = guard(0, vec![]);
        idx.insert(4, Some(&g), &[]);
        idx.add_guard_value(4, 0, iv(10));
        idx.add_guard_value(4, 0, iv(11));
        idx.remove(4, Some(&g), &[]);
        assert_eq!(places(&idx, 4), (0, false));
        assert!(idx.eq.is_empty() && idx.dynamic.is_empty());
        assert_eq!(idx.size(), 0);
        // The recycled slot starts clean under a new tenant.
        idx.insert(4, Some(&guard(1, vec![iv(20)])), &[]);
        assert!(idx
            .push_candidates(&add(owned_row(vec![iv(10), iv(0)])))
            .is_empty());
        assert_eq!(
            idx.push_candidates(&add(owned_row(vec![iv(0), iv(20)]))),
            vec![4]
        );
    }

    /// Every add/remove sequence up to four deep over three family-root slots and two
    /// guard values, exhaustively: after each step, for every cell value the push
    /// candidates of an `Add` (and a `Remove`) carrying it are exactly the slots holding
    /// at least one live copy of that value — the refcount, not the last add or the first
    /// remove — an `Edit` between two values is the union, the size probe counts every
    /// live copy twice, and an index with no live copies is fully pruned. A slot whose last
    /// copy of a value is removed while a sibling binding still shares it stays a candidate:
    /// the wrong "gone" there is a dropped delta (design 310 §4.1).
    ///
    /// The dynamic guard's bounded model check as a walk of every sequence rather than a
    /// Kani harness: the domain is finite, the code is safe Rust over B-trees, and a
    /// harness of a smaller bound could not finish inside the Kani lane's budget.
    #[test]
    fn dynamic_guard_candidates_equal_live_values_after_every_bounded_sequence() {
        const SLOTS: u32 = 3;
        const VALS: usize = 2;
        const DEPTH: usize = 4;
        type Counts = [[u8; VALS]; SLOTS as usize];

        fn live(counts: &Counts, v: usize) -> Vec<u32> {
            (0..SLOTS).filter(|s| counts[*s as usize][v] > 0).collect()
        }

        fn check(idx: &PushIndex, counts: &Counts, rows: &[OwnedRow]) {
            for (v, row) in rows.iter().enumerate() {
                let expected = live(counts, v);
                assert_eq!(
                    idx.push_candidates(&add(row.clone())),
                    expected,
                    "{counts:?}"
                );
                assert_eq!(
                    idx.push_candidates(&SourceChange::Remove(row.clone())),
                    expected,
                    "{counts:?}"
                );
            }
            for (a, new) in rows.iter().enumerate() {
                for (b, old) in rows.iter().enumerate() {
                    let mut expected = live(counts, a);
                    expected.extend(live(counts, b));
                    expected.sort_unstable();
                    expected.dedup();
                    let edit = SourceChange::Edit {
                        row: new.clone(),
                        old: old.clone(),
                    };
                    assert_eq!(idx.push_candidates(&edit), expected, "{counts:?}");
                }
            }
            let copies: usize = counts.iter().flatten().map(|c| *c as usize).sum();
            assert_eq!(idx.size(), 2 * copies, "{counts:?}");
            if copies == 0 {
                assert!(idx.eq.is_empty() && idx.dynamic.is_empty(), "{counts:?}");
            }
        }

        fn explore(idx: &PushIndex, counts: &mut Counts, depth: usize, rows: &[OwnedRow]) -> usize {
            check(idx, counts, rows);
            if depth == 0 {
                return 1;
            }
            let mut visited = 1;
            for slot in 0..SLOTS {
                for v in 0..VALS {
                    let mut next = idx.clone();
                    next.add_guard_value(slot, 0, iv(v as i64));
                    counts[slot as usize][v] += 1;
                    visited += explore(&next, counts, depth - 1, rows);
                    counts[slot as usize][v] -= 1;

                    let mut next = idx.clone();
                    next.remove_guard_value(slot, 0, &iv(v as i64));
                    let had = counts[slot as usize][v];
                    counts[slot as usize][v] = had.saturating_sub(1);
                    visited += explore(&next, counts, depth - 1, rows);
                    counts[slot as usize][v] = had;
                }
            }
            visited
        }

        let mut idx = PushIndex::new();
        let root = guard(0, vec![]);
        for slot in 0..SLOTS {
            idx.insert(slot, Some(&root), &[]);
        }
        let rows: Vec<OwnedRow> = (0..VALS).map(|v| owned_row(vec![iv(v as i64)])).collect();
        let visited = explore(&idx, &mut [[0; VALS]; SLOTS as usize], DEPTH, &rows);
        let branching = 2 * SLOTS as usize * VALS;
        let expected: usize = (0..=DEPTH as u32).map(|d| branching.pow(d)).sum();
        assert_eq!(visited, expected);
    }

    #[test]
    fn candidates_empty_index_is_empty() {
        let idx = PushIndex::new();
        assert!(idx.push_candidates(&add(owned_row(vec![iv(1)]))).is_empty());
    }
}