armdb 0.12.0

//! Engine-maintained counters (T9).
//!
//! A source collection's write projects `0..N` contributions (`u64` weights
//! keyed by `(target, key)`); [`diff_counters`]/[`diff_counters_bytes`] diff
//! old against new contribution snapshots, reduce them to a signed delta per
//! `(group, key)`, and the write path applies each delta via
//! [`ErasedCounterTarget::apply_delta`], which calls
//! [`TypedCounterTarget::resolve_typed`] — the single arithmetic point every
//! caller (write path, transaction commit, open-time recount, `repair`) goes
//! through to turn a delta into a write.
//!
//! All four paths that can move a counter go through it: the non-transactional
//! write path (the facade and the `atomic()`/TTL replay teardown, the same
//! channel the index projection already uses), `db.tx()` commit
//! (`txn::projection_commit`, which resolves the delta to an absolute value
//! for the log record), the open-time [`recount`], and `schema::repair`.

use std::collections::HashMap;
use std::sync::Arc;

use crate::Key;
use crate::error::{DbError, DbResult};
use crate::hex::hex;
use crate::hook::{TypedWriteHook, WriteHook};
use crate::projection::{IndexKeyed, ProjectableIndexTx};
use crate::upsert::Upsert;

/// What to do when the target row a contribution names does not exist.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OnMissingTarget {
    /// Default: the contribution has nowhere to go and the write fails.
    Error,
    /// Deliberately drop the contribution.
    Skip,
    /// Call `apply` with `None` and let it materialise the row.
    Create,
}

/// Erased result of [`TypedCounterTarget::resolve`].
#[derive(Debug)]
pub(crate) enum CounterApply {
    Put(Vec<u8>),
    Delete,
    Skip,
}

/// A collection that can hold a counter field. Implemented for the typed
/// managed families in Task 3; sealed through `IndexKeyed`'s own seal.
pub trait CounterCollection: IndexKeyed {
    fn decode_value(&self, bytes: &[u8]) -> DbResult<Self::V>;
    fn contains_bytes(&self, key: &[u8]) -> DbResult<bool>;
    /// Read path — must NOT take the shard write lock. `read_stored` and
    /// `contains` on the erased target go through here, and both run per row
    /// on full scans.
    fn read_bytes(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>>;
    /// RMW under this collection's own shard lock.
    #[allow(clippy::type_complexity)]
    fn upsert_counter(
        &self,
        key: &[u8],
        f: &mut dyn FnMut(Option<&Self::V>) -> Upsert<Self::V>,
    ) -> DbResult<()>;
}

type ReadFn<V> = Box<dyn Fn(&V) -> u64 + Send + Sync>;
type ApplyFn<V> = Box<dyn Fn(Option<&V>, u64) -> Upsert<V> + Send + Sync>;

/// One declared counter: the target collection plus the application's
/// `read` (stored value -> counter) and `apply` (new count -> write decision)
/// closures.
pub struct TypedCounterTarget<C: CounterCollection> {
    coll: Arc<C>,
    field: &'static str,
    on_missing: OnMissingTarget,
    read: ReadFn<C::V>,
    apply: ApplyFn<C::V>,
}

impl<C: CounterCollection> TypedCounterTarget<C> {
    pub fn new(
        coll: Arc<C>,
        field: &'static str,
        on_missing: OnMissingTarget,
        read: impl Fn(&C::V) -> u64 + Send + Sync + 'static,
        apply: impl Fn(Option<&C::V>, u64) -> Upsert<C::V> + Send + Sync + 'static,
    ) -> Self {
        Self {
            coll,
            field,
            on_missing,
            read: Box::new(read),
            apply: Box::new(apply),
        }
    }

    /// The arithmetic: current typed value + signed delta -> write decision.
    /// Byte-level [`Self::resolve`] is a thin decode/encode wrapper over this
    /// — this is the one place that does checked `i128` arithmetic, applies
    /// the [`OnMissingTarget`] policy, and enforces "`Delete` only at zero."
    ///
    /// `cur` is `None` exactly when the target row does not exist; `resolve`
    /// never sees the row key (only the write path that calls it does), so
    /// the two errors below carry a `"<resolve>"` placeholder in `key` — the
    /// caller that has the key rewrites it with [`DbError::with_key`].
    pub(crate) fn resolve_typed(&self, cur: Option<&C::V>, delta: i128) -> DbResult<Upsert<C::V>> {
        if cur.is_none() {
            match self.on_missing {
                OnMissingTarget::Error => {
                    return Err(DbError::CounterTargetMissing {
                        targets: self.coll.collection_name().to_string(),
                        key: String::from("<resolve>"),
                    });
                }
                OnMissingTarget::Skip => return Ok(Upsert::Keep),
                OnMissingTarget::Create => {}
            }
        }
        let base = i128::from(cur.map_or(0, |v| (self.read)(v)));
        let next = base
            .checked_add(delta)
            .and_then(|n| u64::try_from(n).ok())
            .ok_or_else(|| DbError::CounterOverflow {
                collection: self.coll.collection_name(),
                field: self.field,
                key: String::from("<resolve>"),
            })?;
        match (self.apply)(cur, next) {
            Upsert::Put(v) => Ok(Upsert::Put(v)),
            Upsert::Keep => Ok(Upsert::Keep),
            Upsert::Delete if next == 0 => Ok(Upsert::Delete),
            Upsert::Delete => Err(DbError::Internal(
                "counter apply returned Delete at a non-zero count",
            )),
        }
    }

    /// Byte-level wrapper over [`Self::resolve_typed`]: decode `cur`, run the
    /// arithmetic, erase the decision to [`CounterApply`] (encoding only the
    /// `Put` case). The transaction commit path works in bytes, so this
    /// stays; `apply_delta` (already holding a typed value from
    /// `upsert_counter`) calls `resolve_typed` directly instead.
    pub(crate) fn resolve(&self, cur: Option<&[u8]>, delta: i128) -> DbResult<CounterApply> {
        let decoded = cur.map(|bytes| self.coll.decode_value(bytes)).transpose()?;
        match self.resolve_typed(decoded.as_ref(), delta)? {
            Upsert::Put(v) => Ok(CounterApply::Put(self.coll.encode_value(&v)?)),
            Upsert::Keep => Ok(CounterApply::Skip),
            Upsert::Delete => Ok(CounterApply::Delete),
        }
    }
}

impl DbError {
    /// Rewrite the `"<resolve>"` placeholder [`TypedCounterTarget::resolve`]
    /// leaves in [`DbError::CounterTargetMissing`] / [`DbError::CounterOverflow`]
    /// — `resolve` sees only bytes and a delta, never the row key. Any other
    /// variant is returned unchanged.
    pub(crate) fn with_key(self, key: &[u8]) -> DbError {
        match self {
            DbError::CounterTargetMissing { targets, .. } => DbError::CounterTargetMissing {
                targets,
                key: hex(key),
            },
            DbError::CounterOverflow {
                collection, field, ..
            } => DbError::CounterOverflow {
                collection,
                field,
                key: hex(key),
            },
            other => other,
        }
    }
}

/// Type-erased view of a [`TypedCounterTarget`] the engine holds.
pub(crate) trait ErasedCounterTarget: Send + Sync {
    fn collection_name(&self) -> &'static str;
    fn field(&self) -> &'static str;
    /// The declared policy for a contribution whose target row is absent.
    /// Read by `schema::expected::build_counters`, which must not put a
    /// contribution the write path would have **dropped** into the expected
    /// map — see its own rustdoc.
    fn on_missing(&self) -> OnMissingTarget;
    fn resolve(&self, cur: Option<&[u8]>, delta: i128) -> DbResult<CounterApply>;
    /// Apply `delta` to `key` under the target's own lock. Returns whether a
    /// write actually happened: `false` when the resolved decision was
    /// [`Upsert::Keep`] — an `OnMissingTarget::Skip` on an absent row, or an
    /// application `apply` that declined. Callers that count repairs
    /// (`recount`, `schema::repair`) must not report those as fixed, or they
    /// claim a repair that leaves the same mismatch behind on the next run.
    fn apply_delta(&self, key: &[u8], delta: i128) -> DbResult<bool>;
    fn contains(&self, key: &[u8]) -> DbResult<bool>;
    fn for_each_counter_row(&self, f: &mut dyn FnMut(&[u8], u64) -> DbResult<()>) -> DbResult<()>;
    fn durable_flush(&self) -> DbResult<()>;
    fn as_tx_participant(&self) -> Arc<dyn crate::txn::participant::TxParticipant>;
}

/// Manual `Debug` for the trait object: `resolve_group` returns
/// `Option<&Arc<dyn ErasedCounterTarget>>`, and `Result::unwrap_err` requires
/// the `Ok` side to be `Debug` even when only the `Err` side is inspected.
/// `TypedCounterTarget` itself can't derive `Debug` (its `read`/`apply`
/// closures don't implement it), so this formats the trait object by its
/// stable identity — collection and field — instead.
impl std::fmt::Debug for dyn ErasedCounterTarget {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ErasedCounterTarget")
            .field("collection", &self.collection_name())
            .field("field", &self.field())
            .finish()
    }
}

impl<C: CounterCollection + ProjectableIndexTx> ErasedCounterTarget for TypedCounterTarget<C> {
    fn collection_name(&self) -> &'static str {
        self.coll.collection_name()
    }

    fn field(&self) -> &'static str {
        self.field
    }

    fn on_missing(&self) -> OnMissingTarget {
        self.on_missing
    }

    fn resolve(&self, cur: Option<&[u8]>, delta: i128) -> DbResult<CounterApply> {
        TypedCounterTarget::resolve(self, cur, delta)
    }

    /// Runs the RMW under this target's own shard lock: `upsert_counter`
    /// hands `resolve_typed` the already-typed current value and applies
    /// whatever `Upsert` it returns — no encoding or decoding happens here at
    /// all (the codec only runs on the byte-level `resolve`/transaction
    /// path). `outcome` smuggles a `DbResult` out of the `FnMut`, whose own
    /// return type is `Upsert<Self::V>`, not `DbResult`; `wrote` smuggles out
    /// whether the decision was a real write, which the trait's callers need
    /// and `Ok(())` cannot express.
    fn apply_delta(&self, key: &[u8], delta: i128) -> DbResult<bool> {
        counter_delta_fault()?;
        let mut outcome: DbResult<()> = Ok(());
        let mut wrote = false;
        self.coll
            .upsert_counter(key, &mut |cur| match self.resolve_typed(cur, delta) {
                Ok(decision) => {
                    wrote = !matches!(decision, Upsert::Keep);
                    decision
                }
                Err(e) => {
                    outcome = Err(e.with_key(key));
                    Upsert::Keep
                }
            })?;
        outcome.map(|()| wrote)
    }

    fn contains(&self, key: &[u8]) -> DbResult<bool> {
        self.coll.contains_bytes(key)
    }

    fn for_each_counter_row(&self, f: &mut dyn FnMut(&[u8], u64) -> DbResult<()>) -> DbResult<()> {
        self.coll.for_each_bytes(&mut |k, v| {
            let val = self.coll.decode_value(v)?;
            f(k, (self.read)(&val))
        })
    }

    fn durable_flush(&self) -> DbResult<()> {
        self.coll.durable_flush()
    }

    /// `CounterCollection` (`: IndexKeyed`) does not itself carry the
    /// transaction-participant seam — this impl adds `ProjectableIndexTx` on
    /// top (mirrors [`crate::projection::IndexTargets::with_index`], the
    /// analogous seam for secondary-index targets).
    fn as_tx_participant(&self) -> Arc<dyn crate::txn::participant::TxParticipant> {
        Arc::clone(&self.coll).as_tx_participant()
    }
}

/// Members of a `one_of` group, in declaration order. A group of length 1 is
/// an ordinary single target.
#[derive(Clone)]
pub(crate) struct ErasedGroup {
    pub(crate) members: Vec<Arc<dyn ErasedCounterTarget>>,
    pub(crate) one_of: bool,
}

/// A declared set of counter targets, built with [`CounterTargets::with_counter`]
/// and [`CounterTargets::one_of`].
///
/// **v1 limitation:** at most one target may be declared per collection name,
/// and at most one `one_of` group may be declared, within a single
/// `CounterTargets`. `add`/`add_one_of` route purely by collection name and
/// "the `one_of` group" respectively, with no field qualifier — a second
/// target on the same collection, or a second `one_of` group, would be
/// unreachable through that routing and would silently absorb its
/// contributions into the first match instead. Both are rejected at
/// declaration time (`with_counter`/`OneOfBuilder::build` assert). A single
/// source cannot yet maintain two counter fields side by side in the same
/// target row (e.g. a count and a sum) — lifting this later means adding a
/// field-qualified `add`, not changing the current one.
#[derive(Clone, Default)]
pub struct CounterTargets {
    groups: Vec<ErasedGroup>,
}

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

    /// Declare a single counter target.
    ///
    /// Asserts that no target for `coll`'s collection is already declared in
    /// this `CounterTargets` — see the type's rustdoc for why a second target
    /// on the same collection is a declaration-time error, not a supported
    /// configuration.
    #[allow(private_bounds)]
    pub fn with_counter<C: CounterCollection + ProjectableIndexTx>(
        mut self,
        coll: &Arc<C>,
        field: &'static str,
        on_missing: OnMissingTarget,
        read: impl Fn(&C::V) -> u64 + Send + Sync + 'static,
        apply: impl Fn(Option<&C::V>, u64) -> Upsert<C::V> + Send + Sync + 'static,
    ) -> Self {
        let name = coll.collection_name();
        assert!(
            self.groups
                .iter()
                .flat_map(|g| g.members.iter())
                .all(|m| m.collection_name() != name),
            "with_counter: collection `{name}` already has a declared counter target in this CounterTargets"
        );
        let t = TypedCounterTarget::new(Arc::clone(coll), field, on_missing, read, apply);
        self.groups.push(ErasedGroup {
            members: vec![Arc::new(t)],
            one_of: false,
        });
        self
    }

    /// Start a `one_of` group: alternative targets, only one of which should
    /// hold a given contribution's row. Every [`OneOfBuilder::with`] until
    /// [`OneOfBuilder::build`] joins the group.
    pub fn one_of(self) -> OneOfBuilder {
        OneOfBuilder {
            targets: self,
            members: Vec::new(),
        }
    }

    pub(crate) fn groups(&self) -> &[ErasedGroup] {
        &self.groups
    }

    pub(crate) fn all(&self) -> impl Iterator<Item = &Arc<dyn ErasedCounterTarget>> {
        self.groups.iter().flat_map(|g| g.members.iter())
    }

    pub fn is_empty(&self) -> bool {
        self.groups.is_empty()
    }

    /// The number of declared **groups**, not of targets: a `one_of` group of
    /// three alternative collections counts as one.
    pub fn len(&self) -> usize {
        self.groups.len()
    }

    /// Pick the target of `gi` that owns `key`. A non-`one_of` group returns
    /// its only member unconditionally — that member's own `OnMissingTarget`
    /// decides what happens on a missing row later, in `resolve`, so there is
    /// no `contains` probe here. A `one_of` group requires exactly one
    /// alternative to hold the row: zero is [`DbError::CounterTargetMissing`],
    /// more than one is [`DbError::AmbiguousCounterTarget`].
    pub(crate) fn resolve_group(
        &self,
        gi: usize,
        key: &[u8],
    ) -> DbResult<Option<&Arc<dyn ErasedCounterTarget>>> {
        let g = self
            .groups
            .get(gi)
            .ok_or(DbError::Internal("counter group index"))?;
        if !g.one_of {
            return Ok(Some(&g.members[0]));
        }
        let mut holder = None;
        for m in &g.members {
            if m.contains(key)? {
                if holder.is_some() {
                    return Err(DbError::AmbiguousCounterTarget {
                        targets: group_names(g),
                        key: hex(key),
                    });
                }
                holder = Some(m);
            }
        }
        holder
            .map(Some)
            .ok_or_else(|| DbError::CounterTargetMissing {
                targets: group_names(g),
                key: hex(key),
            })
    }
}

/// Comma-joined collection names of a group's members, for the
/// `CounterTargetMissing` / `AmbiguousCounterTarget` finding text.
fn group_names(g: &ErasedGroup) -> String {
    g.members
        .iter()
        .map(|m| m.collection_name())
        .collect::<Vec<_>>()
        .join(", ")
}

/// Builder for a `one_of` group — started by [`CounterTargets::one_of`].
pub struct OneOfBuilder {
    targets: CounterTargets,
    members: Vec<Arc<dyn ErasedCounterTarget>>,
}

impl OneOfBuilder {
    /// Add one alternative target to the group. Mirrors
    /// [`CounterTargets::with_counter`], including its "collection already
    /// declared" assert — checked against both the already-closed groups and
    /// the members accumulated so far in this in-progress group.
    #[allow(private_bounds)]
    pub fn with<C: CounterCollection + ProjectableIndexTx>(
        mut self,
        coll: &Arc<C>,
        field: &'static str,
        on_missing: OnMissingTarget,
        read: impl Fn(&C::V) -> u64 + Send + Sync + 'static,
        apply: impl Fn(Option<&C::V>, u64) -> Upsert<C::V> + Send + Sync + 'static,
    ) -> Self {
        let name = coll.collection_name();
        assert!(
            self.targets
                .groups
                .iter()
                .flat_map(|g| g.members.iter())
                .chain(self.members.iter())
                .all(|m| m.collection_name() != name),
            "one_of: collection `{name}` already has a declared counter target in this CounterTargets"
        );
        let t = TypedCounterTarget::new(Arc::clone(coll), field, on_missing, read, apply);
        self.members.push(Arc::new(t));
        self
    }

    /// Close the group and return the updated [`CounterTargets`].
    ///
    /// A `one_of` group only means something with at least two alternatives —
    /// with fewer, there is nothing to disambiguate. That is a usage error:
    /// debug builds assert it; release builds degrade the group to an
    /// ordinary single target (`one_of: false`) instead of panicking in
    /// production. With **zero** members there is no target at all — the
    /// group is dropped and `self.targets` is returned unchanged, so it
    /// contributes nothing to `len()`/`is_empty()`.
    ///
    /// A `CounterTargets` may declare at most one `one_of` group — `add_one_of`
    /// routes to "the" `one_of` group with no further disambiguation, so a
    /// second one would be unreachable and silently absorb its contributions
    /// into the first. Asserted here, once it's known this group actually
    /// becomes `one_of` (i.e. has at least two members).
    pub fn build(mut self) -> CounterTargets {
        if self.members.is_empty() {
            return self.targets;
        }
        debug_assert!(
            self.members.len() >= 2,
            "one_of() group must have at least two members"
        );
        let one_of = self.members.len() >= 2;
        if one_of {
            assert!(
                !self.targets.groups.iter().any(|g| g.one_of),
                "one_of: a CounterTargets may declare at most one one_of group"
            );
        }
        self.targets.groups.push(ErasedGroup {
            members: self.members,
            one_of,
        });
        self.targets
    }
}

/// Buffer + budgets + declared target table for one source's projection
/// evaluation (mirrors [`crate::projection::ProjectionSink`]).
///
/// Contributions to the same `(group, key)` are summed with checked `u128`
/// arithmetic, never last-write-wins. The budget is charged **per emission,
/// before coalescing**: two `add` calls for the same key cost two rows
/// against `max_rows`, even though they collapse into one accumulated entry
/// — otherwise a projection could emit unboundedly many contributions to one
/// key and stay under the limit.
pub struct CounterSink<'a> {
    targets: &'a CounterTargets,
    max_rows: usize,
    max_bytes: usize,
    emissions: usize,
    bytes: usize,
    /// (group index, key bytes) -> summed weight.
    acc: HashMap<(usize, Vec<u8>), u128>,
}

impl CounterSink<'_> {
    pub(crate) fn new(
        targets: &CounterTargets,
        max_rows: usize,
        max_bytes: usize,
    ) -> CounterSink<'_> {
        CounterSink {
            targets,
            max_rows,
            max_bytes,
            emissions: 0,
            bytes: 0,
            acc: HashMap::new(),
        }
    }

    /// Borrow the accumulated contributions without consuming the sink.
    /// Test-only: the engine always finishes a projection with
    /// [`Self::into_acc`], so an inspecting borrow has no production caller —
    /// `#[cfg(test)]` rather than a module-wide `#[allow(dead_code)]`, which
    /// would hide the next genuinely dead member.
    #[cfg(test)]
    pub(crate) fn contributions(&self) -> &HashMap<(usize, Vec<u8>), u128> {
        &self.acc
    }

    pub(crate) fn into_acc(self) -> HashMap<(usize, Vec<u8>), u128> {
        self.acc
    }

    /// Contribute `weight` to the single, non-`one_of` group declared for
    /// `target`'s collection. Matching is by declared collection name, not by
    /// `Arc` identity — `target` need not be the exact instance passed to
    /// `with_counter`, only declared under the same name.
    pub fn add<C: CounterCollection>(
        &mut self,
        target: &Arc<C>,
        key: C::K,
        weight: u64,
    ) -> DbResult<()> {
        let name = target.collection_name();
        let gi = self
            .targets
            .groups()
            .iter()
            .position(|g| !g.one_of && g.members[0].collection_name() == name)
            .ok_or(DbError::UndeclaredCounterTarget { target: name })?;
        self.record(gi, C::encode_key(&key), weight)
    }

    /// Route to the single `one_of` group; the holder is chosen at apply time
    /// by [`CounterTargets::resolve_group`], not here.
    pub fn add_one_of(&mut self, key: impl Key, weight: u64) -> DbResult<()> {
        let gi = self
            .targets
            .groups()
            .iter()
            .position(|g| g.one_of)
            .ok_or(DbError::UndeclaredCounterTarget { target: "<one_of>" })?;
        self.record(gi, key.as_bytes().to_vec(), weight)
    }

    fn record(&mut self, gi: usize, key: Vec<u8>, weight: u64) -> DbResult<()> {
        // Charged per emission, before coalescing: otherwise an unbounded
        // number of contributions to one key stays under the row limit.
        self.emissions += 1;
        self.bytes += key.len() + 8;
        if self.emissions > self.max_rows || self.bytes > self.max_bytes {
            return Err(DbError::ProjectionBudgetExceeded {
                rows: self.emissions,
                bytes: self.bytes,
            });
        }
        let slot = self.acc.entry((gi, key)).or_insert(0u128);
        *slot = slot
            .checked_add(u128::from(weight))
            .ok_or(DbError::Internal("counter contribution sum overflow"))?;
        Ok(())
    }
}

/// One computed counter delta, deferred for application until after the
/// source's shard lock is released — the counter-half twin of
/// `crate::projection::PendingRow`. `group` indexes into the
/// [`CounterTargets`] the delta was diffed against; `key` is the encoded
/// target row key; `delta` is the signed change to apply via
/// [`ErasedCounterTarget::apply_delta`].
#[derive(Debug, Clone)]
pub(crate) struct PendingDelta {
    pub group: usize,
    pub key: Vec<u8>,
    pub delta: i128,
}

/// Attach each diffed [`PendingDelta`] to the concrete target that owns its
/// row, for the transaction commit path (T9 Task 9) — the counter twin of
/// [`crate::projection::PendingProjection::into_tx_derived`].
///
/// Only the *routing* happens here. The arithmetic does not: a delta is
/// resolved to an absolute value later, under the target's own shard lock, so
/// that the log record carries the value rather than the change. Group
/// resolution is safe at this point because a counter target is always a
/// rank-1 participant (`Db::tx_register_counter_targets`) and therefore not yet
/// locked by this commit — a `one_of` group's `contains` probe cannot deadlock
/// against a lock this thread already holds.
#[cfg(feature = "armour")]
pub(crate) fn into_tx_counter_deltas(
    targets: &CounterTargets,
    deltas: Vec<PendingDelta>,
) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
    let mut out = Vec::with_capacity(deltas.len());
    for d in deltas {
        // `None` is reserved (see `PendingProjection::apply_or_incomplete`)
        // and unreachable today: `resolve_group` errors on both the
        // zero-holder and the ambiguous case, and a non-`one_of` group always
        // resolves to its one member. Unlike the deferred plan, silently
        // skipping here would **drop a delta from a transaction that then
        // commits** — the counter would be permanently wrong with nothing to
        // repair it, since this path deliberately sets no dirty marker. Fail
        // loudly instead if the invariant ever changes.
        let Some(target) = targets.resolve_group(d.group, &d.key)? else {
            return Err(DbError::Internal(
                "counter group resolved to no target inside a transaction",
            ));
        };
        out.push(crate::txn::participant::TxCounterDelta {
            participant: target.as_tx_participant(),
            target: Arc::clone(target),
            key: d.key,
            delta: d.delta,
        });
    }
    Ok(out)
}

/// Snapshot one side (`old` or `new`) of a typed mutation into a fresh
/// [`CounterSink`]. An absent value (insert's `old`, delete's `new`)
/// contributes nothing, so its snapshot is the empty map — matches
/// `crate::projection::project_snapshot`'s "empty for `None`" shape.
fn counter_snapshot<K: Key, T, H: TypedWriteHook<K, T>>(
    hook: &H,
    targets: &CounterTargets,
    max_rows: usize,
    max_bytes: usize,
    key: &K,
    snapshot: Option<&T>,
) -> DbResult<HashMap<(usize, Vec<u8>), u128>> {
    let mut sink = CounterSink::new(targets, max_rows, max_bytes);
    if let Some(value) = snapshot {
        hook.counter_project(key, value, &mut sink)?;
    }
    Ok(sink.into_acc())
}

/// Byte-snapshot twin of [`counter_snapshot`] for `WriteHook` families. Mirrors
/// `crate::projection::project_snapshot_bytes`: the snapshotted side is always
/// passed as `counter_project`'s `new` argument — that method's shape mirrors
/// [`WriteHook::project`], which likewise only reads its `new` parameter.
fn counter_snapshot_bytes<K: Key, H: WriteHook<K>>(
    hook: &H,
    targets: &CounterTargets,
    max_rows: usize,
    max_bytes: usize,
    key: &K,
    snapshot: Option<&[u8]>,
) -> DbResult<HashMap<(usize, Vec<u8>), u128>> {
    let mut sink = CounterSink::new(targets, max_rows, max_bytes);
    if let Some(value) = snapshot {
        hook.counter_project(key, None, Some(value), &mut sink)?;
    }
    Ok(sink.into_acc())
}

/// Reduce two contribution snapshots (`old`, `new`) to signed per-`(group,
/// key)` deltas: a weight unchanged between the two drops out, a weight
/// present only in `new` is a positive delta, only in `old` a negative one.
/// Shared by [`diff_counters`] and [`diff_counters_bytes`] — the two differ
/// only in how the snapshots are built.
fn diff_counter_snapshots(
    targets: &CounterTargets,
    a: HashMap<(usize, Vec<u8>), u128>,
    b: HashMap<(usize, Vec<u8>), u128>,
) -> DbResult<Vec<PendingDelta>> {
    // The diff never sees a target's *current* stored value, only summed
    // contribution weights, so an overflow here can't reuse `resolve`'s
    // `"<resolve>"` key placeholder — name the target from the group itself.
    let overflow = |slot: &(usize, Vec<u8>)| -> DbError {
        let (collection, field) = targets
            .groups()
            .get(slot.0)
            .and_then(|g| g.members.first())
            .map(|m| (m.collection_name(), m.field()))
            .unwrap_or(("<unknown>", "<unknown>"));
        DbError::CounterOverflow {
            collection,
            field,
            key: hex(&slot.1),
        }
    };

    let mut out = Vec::new();
    for (slot, new_w) in &b {
        let old_w = a.get(slot).copied().unwrap_or(0);
        let delta = i128::try_from(*new_w).map_err(|_| overflow(slot))?
            - i128::try_from(old_w).map_err(|_| overflow(slot))?;
        if delta != 0 {
            out.push(PendingDelta {
                group: slot.0,
                key: slot.1.clone(),
                delta,
            });
        }
    }
    for (slot, old_w) in &a {
        if !b.contains_key(slot) {
            out.push(PendingDelta {
                group: slot.0,
                key: slot.1.clone(),
                delta: -i128::try_from(*old_w).map_err(|_| overflow(slot))?,
            });
        }
    }
    // Deterministic apply order — the plan is replayed and compared in tests.
    out.sort_by(|x, y| (x.group, &x.key).cmp(&(y.group, &y.key)));
    Ok(out)
}

/// Diff a typed mutation's old/new contribution snapshots into deltas —
/// counter half of [`crate::projection::diff_projection`]. `H::COUNTS` is
/// compile-time: an uncounted collection's write path pays nothing beyond
/// this one check.
pub(crate) fn diff_counters<K: Key, T, H: TypedWriteHook<K, T>>(
    hook: &H,
    targets: &CounterTargets,
    max_rows: usize,
    max_bytes: usize,
    key: &K,
    old: Option<&T>,
    new: Option<&T>,
) -> DbResult<Vec<PendingDelta>> {
    if !H::COUNTS {
        return Ok(Vec::new());
    }
    let a = counter_snapshot(hook, targets, max_rows, max_bytes, key, old)?;
    let b = counter_snapshot(hook, targets, max_rows, max_bytes, key, new)?;
    diff_counter_snapshots(targets, a, b)
}

/// Byte-snapshot twin of [`diff_counters`] for the `WriteHook` (Const/Zero/Var)
/// families — mirrors [`crate::projection::diff_projection_bytes`].
pub(crate) fn diff_counters_bytes<K: Key, H: WriteHook<K>>(
    hook: &H,
    targets: &CounterTargets,
    max_rows: usize,
    max_bytes: usize,
    key: &K,
    old: Option<&[u8]>,
    new: Option<&[u8]>,
) -> DbResult<Vec<PendingDelta>> {
    if !H::COUNTS {
        return Ok(Vec::new());
    }
    let a = counter_snapshot_bytes(hook, targets, max_rows, max_bytes, key, old)?;
    let b = counter_snapshot_bytes(hook, targets, max_rows, max_bytes, key, new)?;
    diff_counter_snapshots(targets, a, b)
}

/// Fault injection for [`ErasedCounterTarget::apply_delta`] (T9 Task 8) — lets
/// a test simulate a hard exit between a source write landing and its counter
/// delta being applied, without an actual crash. Compiles away entirely
/// without `test-faults` (a non-default feature).
#[cfg(any(test, feature = "test-faults"))]
pub(crate) static FAIL_COUNTER_DELTA: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

#[inline(always)]
fn counter_delta_fault() -> DbResult<()> {
    #[cfg(any(test, feature = "test-faults"))]
    if FAIL_COUNTER_DELTA.load(std::sync::atomic::Ordering::SeqCst) {
        return Err(DbError::Internal("injected counter delta failure"));
    }
    Ok(())
}

/// Open-time / on-demand recount (T9 Task 8): a full two-pass walk that
/// recomputes every declared target row's aggregate from the source's
/// current live data and repairs any row whose stored value drifted from it.
/// Returns the number of target rows repaired.
///
/// Shared by all six write-path families (`TypedTree`, `TypedMap`,
/// `ConstTree`, `ConstMap`, `VarTree`, `VarMap` — the `Zero*`/`VarTyped*`
/// wrappers delegate to whichever of these backs them) — each supplies its
/// own `project` (how to run the hook's counter projection over one row) and
/// `scan` (how to walk every live row) closures, so the walk logic itself
/// exists exactly once. `V` is the row shape the family scans with: a typed
/// `&T` for the two typed families, `&[u8]` for the four byte-level ones.
///
/// # Two passes, and why both are needed
///
/// Pass 1 sums every source row's contributions into `expected`, exactly like
/// a write's projection. Pass 2 does **not** stop at `expected`'s keys — it
/// also walks every declared target member's *currently stored* rows, and
/// unions the two key sets. A row with no contribution at all (everyone left
/// the group) has no entry in `expected` and would never surface from pass 1
/// alone; only walking the target catches it, and drives it to zero.
///
/// Each unioned key is then routed through [`CounterTargets::resolve_group`]
/// exactly like the write path resolves a live delta: a `one_of` group with
/// zero or several current holders for that key reports the same
/// [`DbError::CounterTargetMissing`]/[`DbError::AmbiguousCounterTarget`] here
/// as it would applying a write.
#[allow(clippy::type_complexity)]
pub(crate) fn recount<K: Key, V: ?Sized>(
    project: impl Fn(&K, &V, &mut CounterSink<'_>) -> DbResult<()>,
    targets: &CounterTargets,
    limits: (usize, usize),
    scan: &mut dyn FnMut(&mut dyn FnMut(&K, &V) -> DbResult<()>) -> DbResult<()>,
) -> DbResult<u64> {
    // Pass 1: expected sums per (group, key) from a full source scan.
    let mut expected: HashMap<(usize, Vec<u8>), u128> = HashMap::new();
    scan(&mut |k, v| {
        let mut sink = CounterSink::new(targets, limits.0, limits.1);
        project(k, v, &mut sink)?;
        for (slot, w) in sink.into_acc() {
            let e = expected.entry(slot).or_insert(0u128);
            *e = e
                .checked_add(w)
                .ok_or(DbError::Internal("recount sum overflow"))?;
        }
        Ok(())
    })?;

    // Pass 2: per group, union the keys currently stored in ANY member with
    // the keys `expected` has a contribution for, then resolve and repair
    // each.
    //
    // A dangling contribution — a live source row whose target row does not
    // exist — degrades to a `warn` and is skipped, rather than aborting the
    // whole recount: on the write path the same state costs one write, but
    // at open it would cost the whole database (every subsequent open would
    // hit the identical error and never clear the flag that triggers it).
    // `resolve_group` reports it via `CounterTargetMissing` from two sites: a
    // `one_of` group with zero current holders, and — via `apply_delta` ->
    // `resolve_typed` — a non-`one_of` group whose sole target row is simply
    // missing. Both are caught here. `AmbiguousCounterTarget` (several
    // current holders) is a different, worse condition — there is no target
    // to safely skip *to* — and is left a hard failure.
    let mut repaired = 0u64;
    for (gi, group) in targets.groups().iter().enumerate() {
        let mut existing: HashMap<Vec<u8>, u64> = HashMap::new();
        for member in &group.members {
            member.for_each_counter_row(&mut |key, stored| {
                existing.insert(key.to_vec(), stored);
                Ok(())
            })?;
        }
        let mut keys: std::collections::HashSet<Vec<u8>> = existing.keys().cloned().collect();
        keys.extend(
            expected
                .keys()
                .filter(|(g, _)| *g == gi)
                .map(|(_, key)| key.clone()),
        );
        for key in keys {
            // `None` is reserved (see `PendingProjection::apply_or_incomplete`):
            // a non-`one_of` group always resolves to its one member, so this
            // arm is unreachable today — kept a no-op rather than an `unwrap`.
            let member = match targets.resolve_group(gi, &key) {
                Ok(Some(m)) => m,
                Ok(None) => continue,
                Err(DbError::CounterTargetMissing { targets, key }) => {
                    tracing::warn!(
                        group = gi,
                        targets = %targets,
                        key = %key,
                        "recount: dangling contribution — no live target row holds this key, skipping (see schema::validate)"
                    );
                    continue;
                }
                Err(e) => return Err(e),
            };
            let want_raw = expected.get(&(gi, key.clone())).copied().unwrap_or(0);
            let want = u64::try_from(want_raw).map_err(|_| DbError::CounterOverflow {
                collection: member.collection_name(),
                field: member.field(),
                key: hex(&key),
            })?;
            let stored = existing.get(&key).copied().unwrap_or(0);
            if want != stored {
                let delta = i128::from(want) - i128::from(stored);
                // `Ok(false)` — the resolved decision was `Upsert::Keep`, i.e.
                // nothing was written (an `OnMissingTarget::Skip` on an absent
                // row, or an application `apply` that declined). Counting it
                // would report a repair that did not happen.
                match member.apply_delta(&key, delta) {
                    Ok(true) => repaired += 1,
                    Ok(false) => {}
                    Err(DbError::CounterTargetMissing { targets, key }) => {
                        tracing::warn!(
                            collection = member.collection_name(),
                            field = member.field(),
                            targets = %targets,
                            key = %key,
                            "recount: dangling contribution — target row missing, skipping (see schema::validate)"
                        );
                    }
                    Err(e) => return Err(e),
                }
            }
        }
    }

    // Targets may buffer Bitcask writes; flush before either caller
    // (`Db::open_managed`, `Db::recount_counters`) clears `counters_dirty` —
    // mirrors `reconcile_carrier_projections`'s identical durability barrier
    // for secondary indexes. `Persist::update` writes `db.info` synchronously,
    // so without this a crash between the recount returning and the target's
    // next periodic/close flush would lose the repaired rows *and* the flag
    // that would trigger another recount: permanent silent drift.
    for group in targets.groups() {
        for member in &group.members {
            member.durable_flush()?;
        }
    }
    Ok(repaired)
}

/// `pub(crate)` (not private): `single_stub_targets` and `StubCounterColl` are
/// reused by `projection::tests::counter_carrier_reports_both_projections` to
/// build a real `and_counters` half without duplicating this stub — both
/// modules compile together under `cfg(test)`.
#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    use crate::projection::{ProjectableIndex, sealed};
    use crate::txn::participant::{TxBackend, TxParticipant};

    /// Test-only `CounterCollection`: the counter value lives directly in
    /// `state`, little-endian encoded — matching `resolve`'s `u64` arithmetic.
    pub(crate) struct StubCounterColl {
        name: &'static str,
        state: Arc<std::sync::Mutex<Option<u64>>>,
    }

    impl StubCounterColl {
        fn with(initial: Option<u64>) -> Self {
            Self::named("stub_counter", initial)
        }

        /// A stub carrying an explicit `collection_name`, so two stubs can be
        /// told apart by declared name — `CounterSink::add` matches by name,
        /// not `Arc` identity, so an "undeclared target" test needs a stub
        /// whose name genuinely differs from the declared one.
        ///
        /// `pub(crate)` (not private): reused by
        /// `projection::tests::counter_delta_incomplete_plan_carries_only_the_remainder`
        /// to build two distinctly-named stub targets — one always-succeeding,
        /// one always-failing — without duplicating this type.
        pub(crate) fn named(name: &'static str, initial: Option<u64>) -> Self {
            Self {
                name,
                state: Arc::new(std::sync::Mutex::new(initial)),
            }
        }

        /// Test helper for `one_of` resolution: mark this stub as holding a
        /// row, regardless of `key` (mirrors `contains_bytes`, which also
        /// ignores the key and reports presence only).
        fn insert(&self, _key: &[u8; 8]) {
            *self.state.lock().unwrap() = Some(0);
        }
    }

    impl sealed::Sealed for StubCounterColl {}

    impl ProjectableIndex for StubCounterColl {
        fn collection_name(&self) -> &'static str {
            self.name
        }

        fn put_bytes(&self, _key: &[u8], value: &[u8]) -> DbResult<()> {
            let bytes: [u8; 8] = value
                .try_into()
                .map_err(|_| DbError::Internal("bad counter width"))?;
            *self.state.lock().unwrap() = Some(u64::from_le_bytes(bytes));
            Ok(())
        }

        fn delete_bytes(&self, _key: &[u8]) -> DbResult<()> {
            *self.state.lock().unwrap() = None;
            Ok(())
        }

        fn values_equal(&self, a: &[u8], b: &[u8]) -> DbResult<bool> {
            Ok(a == b)
        }

        fn for_each_bytes(&self, f: &mut dyn FnMut(&[u8], &[u8]) -> DbResult<()>) -> DbResult<()> {
            if let Some(n) = *self.state.lock().unwrap() {
                f(&[0u8; 8], &n.to_le_bytes())?;
            }
            Ok(())
        }

        fn durable_flush(&self) -> DbResult<()> {
            Ok(())
        }
    }

    impl IndexKeyed for StubCounterColl {
        type K = [u8; 8];
        type V = u64;

        fn encode_value(&self, v: &u64) -> DbResult<Vec<u8>> {
            Ok(v.to_le_bytes().to_vec())
        }
    }

    impl CounterCollection for StubCounterColl {
        fn decode_value(&self, bytes: &[u8]) -> DbResult<u64> {
            let arr: [u8; 8] = bytes
                .try_into()
                .map_err(|_| DbError::Internal("bad counter width"))?;
            Ok(u64::from_le_bytes(arr))
        }

        fn contains_bytes(&self, _key: &[u8]) -> DbResult<bool> {
            Ok(self.state.lock().unwrap().is_some())
        }

        fn read_bytes(&self, _key: &[u8]) -> DbResult<Option<Vec<u8>>> {
            Ok(self.state.lock().unwrap().map(|n| n.to_le_bytes().to_vec()))
        }

        fn upsert_counter(
            &self,
            _key: &[u8],
            f: &mut dyn FnMut(Option<&u64>) -> Upsert<u64>,
        ) -> DbResult<()> {
            let mut guard = self.state.lock().unwrap();
            match f(guard.as_ref()) {
                Upsert::Put(v) => *guard = Some(v),
                Upsert::Keep => {}
                Upsert::Delete => *guard = None,
            }
            Ok(())
        }
    }

    /// Stub transaction participant: `with_counter`/`OneOfBuilder::with`
    /// require `ProjectableIndexTx`, but none of the tests in this module
    /// exercise a transaction, so `lock_shards` is never actually called.
    struct StubTxParticipant;

    impl TxParticipant for StubTxParticipant {
        fn name(&self) -> &'static str {
            "stub_counter"
        }
        fn backend(&self) -> TxBackend {
            TxBackend::Bitcask
        }
        fn schema_fingerprint(&self) -> u64 {
            0
        }
        fn encrypted(&self) -> bool {
            false
        }
        fn max_entry_bytes(&self) -> Option<usize> {
            None
        }
        fn shard_of(&self, _key: &[u8]) -> DbResult<usize> {
            Ok(0)
        }
        fn lock_shards<'a>(
            &'a self,
            _shards: &[usize],
        ) -> DbResult<Box<dyn crate::txn::participant::TxLocked + 'a>> {
            Err(DbError::Internal("stub"))
        }
    }

    impl ProjectableIndexTx for StubCounterColl {
        fn as_tx_participant(self: Arc<Self>) -> Arc<dyn TxParticipant> {
            Arc::new(StubTxParticipant)
        }
    }

    /// Minimal target: value is the counter itself, so `read` is identity.
    fn stub_target(initial: Option<u64>) -> TypedCounterTarget<StubCounterColl> {
        TypedCounterTarget::new(
            Arc::new(StubCounterColl::with(initial)),
            "count",
            OnMissingTarget::Create,
            |v: &u64| *v,
            |_cur: Option<&u64>, n: u64| {
                if n == 0 {
                    Upsert::Delete
                } else {
                    Upsert::Put(n)
                }
            },
        )
    }

    #[test]
    fn counter_resolve_adds_and_subtracts() {
        let t = stub_target(Some(5));
        let CounterApply::Put(bytes) = t.resolve(Some(&5u64.to_le_bytes()), 3).unwrap() else {
            panic!("expected Put");
        };
        assert_eq!(u64::from_le_bytes(bytes.try_into().unwrap()), 8);
    }

    #[test]
    fn counter_resolve_zero_deletes() {
        let t = stub_target(Some(1));
        assert!(matches!(
            t.resolve(Some(&1u64.to_le_bytes()), -1).unwrap(),
            CounterApply::Delete
        ));
    }

    #[test]
    fn counter_resolve_underflow_is_an_error_not_saturation() {
        let t = stub_target(Some(1));
        let err = t.resolve(Some(&1u64.to_le_bytes()), -2).unwrap_err();
        assert!(matches!(err, DbError::CounterOverflow { .. }));
    }

    #[test]
    fn counter_resolve_overflow_is_an_error() {
        let t = stub_target(Some(u64::MAX));
        let err = t.resolve(Some(&u64::MAX.to_le_bytes()), 1).unwrap_err();
        assert!(matches!(err, DbError::CounterOverflow { .. }));
    }

    #[test]
    fn counter_resolve_missing_row_follows_the_declared_policy() {
        // Error
        let t = TypedCounterTarget::new(
            Arc::new(StubCounterColl::with(None)),
            "count",
            OnMissingTarget::Error,
            |v: &u64| *v,
            |_c, n| Upsert::Put(n),
        );
        assert!(matches!(
            t.resolve(None, 1).unwrap_err(),
            DbError::CounterTargetMissing { .. }
        ));
        // Skip
        let t = TypedCounterTarget::new(
            Arc::new(StubCounterColl::with(None)),
            "count",
            OnMissingTarget::Skip,
            |v: &u64| *v,
            |_c, n| Upsert::Put(n),
        );
        assert!(matches!(t.resolve(None, 1).unwrap(), CounterApply::Skip));
        // Create
        let t = stub_target(None);
        assert!(matches!(t.resolve(None, 2).unwrap(), CounterApply::Put(_)));
    }

    #[test]
    fn counter_resolve_rejects_delete_at_a_non_zero_count() {
        let t = TypedCounterTarget::new(
            Arc::new(StubCounterColl::with(Some(4))),
            "count",
            OnMissingTarget::Create,
            |v: &u64| *v,
            |_c, _n| Upsert::Delete, // always deletes — a usage bug
        );
        assert!(matches!(
            t.resolve(Some(&4u64.to_le_bytes()), 1).unwrap_err(),
            DbError::Internal(_)
        ));
    }

    /// `Upsert::Keep` (the application chooses not to write) erases to
    /// `CounterApply::Skip` — the mapping is `Put -> Put`, `Delete -> Delete`,
    /// `Keep -> Skip`. No other test makes `apply` return `Keep`.
    #[test]
    fn counter_resolve_keep_erases_to_skip() {
        let t = TypedCounterTarget::new(
            Arc::new(StubCounterColl::with(Some(5))),
            "count",
            OnMissingTarget::Create,
            |v: &u64| *v,
            |_cur, _n| Upsert::Keep, // application declines to write
        );
        assert!(matches!(
            t.resolve(Some(&5u64.to_le_bytes()), 3).unwrap(),
            CounterApply::Skip
        ));
    }

    /// A `one_of()` builder closed with no members contributes nothing — the
    /// documented "degrade to a plain single target" presumes at least one
    /// member; with zero, there is no target to degrade to.
    #[test]
    fn counter_one_of_with_no_members_yields_empty_targets() {
        let targets = CounterTargets::new().one_of().build();
        assert!(targets.is_empty());
        assert_eq!(targets.len(), 0);
    }

    /// One declared single target ("declared_counter") plus the handle
    /// `CounterSink::add` matches it by name.
    pub(crate) fn single_stub_targets() -> (CounterTargets, Arc<StubCounterColl>) {
        let coll = Arc::new(StubCounterColl::named("declared_counter", None));
        let targets = CounterTargets::new().with_counter(
            &coll,
            "count",
            OnMissingTarget::Create,
            |v: &u64| *v,
            |_cur: Option<&u64>, n: u64| Upsert::Put(n),
        );
        (targets, coll)
    }

    /// A single `one_of` group with two alternative stubs, `a` and `b`,
    /// neither holding the key yet — the caller toggles presence with
    /// `insert` to drive "no holder -> one holder -> two holders".
    fn one_of_stub_targets() -> (CounterTargets, Arc<StubCounterColl>, Arc<StubCounterColl>) {
        let a = Arc::new(StubCounterColl::named("stub_a", None));
        let b = Arc::new(StubCounterColl::named("stub_b", None));
        let targets = CounterTargets::new()
            .one_of()
            .with(
                &a,
                "count",
                OnMissingTarget::Create,
                |v: &u64| *v,
                |_cur: Option<&u64>, n: u64| Upsert::Put(n),
            )
            .with(
                &b,
                "count",
                OnMissingTarget::Create,
                |v: &u64| *v,
                |_cur: Option<&u64>, n: u64| Upsert::Put(n),
            )
            .build();
        (targets, a, b)
    }

    #[test]
    fn counter_sink_sums_duplicate_contributions() {
        let (targets, coll) = single_stub_targets();
        let mut sink = CounterSink::new(&targets, 16, 1024);
        sink.add(&coll, [1u8; 8], 2).unwrap();
        sink.add(&coll, [1u8; 8], 3).unwrap();
        assert_eq!(*sink.contributions().values().next().unwrap(), 5u128);
    }

    #[test]
    fn counter_sink_charges_budget_before_coalescing() {
        let (targets, coll) = single_stub_targets();
        let mut sink = CounterSink::new(&targets, 2, 1024); // 2 emissions allowed
        sink.add(&coll, [1u8; 8], 1).unwrap();
        sink.add(&coll, [1u8; 8], 1).unwrap();
        let err = sink.add(&coll, [1u8; 8], 1).unwrap_err(); // same key, still charged
        assert!(matches!(err, DbError::ProjectionBudgetExceeded { .. }));
    }

    #[test]
    fn counter_sink_rejects_an_undeclared_target() {
        let (targets, _declared) = single_stub_targets();
        let other = Arc::new(StubCounterColl::with(Some(0))); // never declared
        let mut sink = CounterSink::new(&targets, 16, 1024);
        assert!(matches!(
            sink.add(&other, [1u8; 8], 1).unwrap_err(),
            DbError::UndeclaredCounterTarget { .. }
        ));
    }

    #[test]
    fn counter_one_of_group_requires_exactly_one_holder() {
        let (targets, a, b) = one_of_stub_targets(); // neither holds the key yet
        let key = [9u8; 8];
        assert!(matches!(
            targets.resolve_group(0, &key).unwrap_err(),
            DbError::CounterTargetMissing { .. }
        ));
        a.insert(&key);
        assert!(targets.resolve_group(0, &key).unwrap().is_some());
        b.insert(&key);
        assert!(matches!(
            targets.resolve_group(0, &key).unwrap_err(),
            DbError::AmbiguousCounterTarget { .. }
        ));
    }

    /// `add` dispatches purely by collection name, taking the *first*
    /// matching group — a second `with_counter` on the same collection would
    /// be unreachable and silently absorb its contributions into the first
    /// one instead of erroring. Rejected at declaration time.
    #[test]
    #[should_panic(expected = "already has a declared counter target")]
    fn counter_targets_rejects_a_second_target_on_the_same_collection() {
        let coll = Arc::new(StubCounterColl::named("dup_counter", None));
        let _ = CounterTargets::new()
            .with_counter(
                &coll,
                "count",
                OnMissingTarget::Create,
                |v: &u64| *v,
                |_cur: Option<&u64>, n: u64| Upsert::Put(n),
            )
            .with_counter(
                &coll,
                "other_field",
                OnMissingTarget::Create,
                |v: &u64| *v,
                |_cur: Option<&u64>, n: u64| Upsert::Put(n),
            );
    }

    /// `add_one_of` dispatches to "the" `one_of` group, taking the *first*
    /// one — a second `one_of` group would be unreachable and silently
    /// absorb its contributions into the first. Rejected at declaration time.
    #[test]
    #[should_panic(expected = "at most one one_of group")]
    fn counter_targets_rejects_a_second_one_of_group() {
        let a = Arc::new(StubCounterColl::named("one_of_a", None));
        let b = Arc::new(StubCounterColl::named("one_of_b", None));
        let c = Arc::new(StubCounterColl::named("one_of_c", None));
        let d = Arc::new(StubCounterColl::named("one_of_d", None));
        let _ = CounterTargets::new()
            .one_of()
            .with(
                &a,
                "count",
                OnMissingTarget::Create,
                |v: &u64| *v,
                |_cur: Option<&u64>, n: u64| Upsert::Put(n),
            )
            .with(
                &b,
                "count",
                OnMissingTarget::Create,
                |v: &u64| *v,
                |_cur: Option<&u64>, n: u64| Upsert::Put(n),
            )
            .build()
            .one_of()
            .with(
                &c,
                "count",
                OnMissingTarget::Create,
                |v: &u64| *v,
                |_cur: Option<&u64>, n: u64| Upsert::Put(n),
            )
            .with(
                &d,
                "count",
                OnMissingTarget::Create,
                |v: &u64| *v,
                |_cur: Option<&u64>, n: u64| Upsert::Put(n),
            )
            .build();
    }
}