armdb 0.14.1

//! Two-round transaction commit for projected secondary indexes (T8 Task 5).

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

use zerocopy::IntoBytes;

use crate::counter::{CounterApply, ErasedCounterTarget};
use crate::error::{DbError, DbResult};
use crate::txn::participant::{ClaimKind, TxLocked, TxParticipant};
use crate::txn::{Action, Op};
use crate::unique::UniqueRow;
use crate::unique::resolve::{ClaimDecision, decide};
use crate::upsert::Upsert;

pub(crate) struct Entry {
    pub id: u32,
    pub rank: u8,
    pub participant: Arc<dyn TxParticipant>,
    pub shards: Vec<usize>,
    pub lock_all: bool,
}

pub(crate) struct Write {
    pub entry: usize,
    pub key: Vec<u8>,
    pub value: Option<Vec<u8>>,
    pub derived: bool,
}

type ProjectionInput = (usize, Vec<u8>, Option<Vec<u8>>, Option<Vec<u8>>);
type EvalRank0Outcome = (Vec<Write>, Vec<ProjectionInput>);
/// Borrowed twin of [`ProjectionInput`] — the claim collapse pass (review
/// round I2) reads `inputs` by reference instead of cloning every key and
/// both value snapshots of every write, so its per-row slots hold references
/// into `inputs` rather than owned copies.
type ClaimRowRef<'a> = (usize, &'a [u8], Option<&'a [u8]>, Option<&'a [u8]>);
type StreamRow = (usize, Vec<u8>, Option<Vec<u8>>, Option<Vec<u8>>);
/// Post-success callbacks produced by lowering a commit's streams.
///
/// Publication permits are *not* part of this: they are taken by
/// `tx_prepare_streams` before the commit guard is entered, precisely so the
/// blocking wait for one never happens under a shard lock.
type StreamLowered = Vec<Box<dyn FnOnce() + Send>>;

fn idx_for(entries: &[Entry], participant: &Arc<dyn TxParticipant>) -> Option<usize> {
    entries
        .iter()
        .position(|e| Arc::ptr_eq(&e.participant, participant))
}

fn ensure_entry(
    entries: &mut Vec<Entry>,
    db: &crate::txn::TxContext<'_>,
    participant: Arc<dyn TxParticipant>,
    rank: u8,
    key: &[u8],
) -> DbResult<usize> {
    let sid = participant.shard_of(key)?;
    if let Some(pos) = idx_for(entries, &participant) {
        if !entries[pos].shards.contains(&sid) {
            entries[pos].shards.push(sid);
        }
        return Ok(pos);
    }
    let id = db.collection_id(participant.as_ref())?;
    entries.push(Entry {
        id,
        rank,
        participant,
        shards: vec![sid],
        lock_all: false,
    });
    Ok(entries.len() - 1)
}

fn ensure_participant(
    entries: &mut Vec<Entry>,
    db: &crate::txn::TxContext<'_>,
    participant: Arc<dyn TxParticipant>,
    rank: u8,
) -> DbResult<usize> {
    if let Some(pos) = idx_for(entries, &participant) {
        return Ok(pos);
    }
    let id = db.collection_id(participant.as_ref())?;
    entries.push(Entry {
        id,
        rank,
        participant,
        shards: Vec::new(),
        lock_all: false,
    });
    Ok(entries.len() - 1)
}

fn sort_entries(entries: &mut [Entry]) {
    entries.sort_by_key(|a| (a.rank, a.id));
    for e in entries.iter_mut() {
        e.shards.sort_unstable();
    }
}

fn set_write(
    writes: &mut Vec<Write>,
    entry: usize,
    key: Vec<u8>,
    value: Option<Vec<u8>>,
    derived: bool,
) {
    match writes.iter_mut().find(|w| w.entry == entry && w.key == key) {
        Some(existing) => {
            // Value collapses, `derived` does not: the flag records who first
            // put this row in the plan, and it is only read to word an error
            // ([`admit`], [`max_bytes_err`]). A counter write collapsing onto
            // the caller's own write to the same row must not relabel that
            // row "derived" — the caller would then be told their oversized
            // value was an engine-generated one.
            existing.value = value;
        }
        None => writes.push(Write {
            entry,
            key,
            value,
            derived,
        }),
    }
}

pub(crate) fn abort_locked<'a>(locked: Vec<Box<dyn TxLocked + 'a>>) {
    for l in locked.into_iter().rev() {
        l.abort();
    }
}

pub(crate) struct PlanOutcome {
    pub entries: Vec<Entry>,
    pub rank0_len: usize,
    pub rank0_ops: Vec<Op>,
    pub rank1_ops: Vec<Op>,
}

pub(crate) fn plan(
    db: &crate::txn::TxContext<'_>,
    ops: Vec<Op>,
    claim_ops: &[crate::unique::ExplicitClaim],
) -> DbResult<PlanOutcome> {
    let mut rank0_ops = Vec::new();
    let mut rank1_ops = Vec::new();
    for op in ops {
        if db.rank_of(op.participant.as_ref()) == 0 {
            rank0_ops.push(op);
        } else {
            rank1_ops.push(op);
        }
    }
    // A rank-1 participant is a declared projection/counter **target**: it is
    // locked in the commit's second round, and [`eval_rank1`] decides its
    // writes without projecting them. A target that is *itself* a projecting
    // or counting source would therefore have its own indexes/counters
    // silently stop being maintained for any write made inside a transaction.
    //
    // Reject that topology up front, before any lock is taken, rather than let
    // the drift accumulate until someone runs `schema::validate`. A plain
    // target — the ordinary case, no projections of its own — reports no
    // preplan targets and is unaffected.
    let mut checked: Vec<&'static str> = Vec::new();
    for op in &rank1_ops {
        let name = op.participant.name();
        if checked.contains(&name) {
            continue;
        }
        checked.push(name);
        if !op.participant.tx_preplan_projection_targets()?.is_empty() {
            return Err(DbError::TxTargetIsAlsoASource { collection: name });
        }
    }
    let mut entries: Vec<Entry> = Vec::new();
    for op in &rank0_ops {
        ensure_entry(&mut entries, db, op.participant.clone(), 0, &op.key)?;
    }
    let sources: Vec<_> = entries.iter().map(|e| e.participant.clone()).collect();
    for src in sources {
        for target in src.tx_preplan_projection_targets()? {
            ensure_participant(&mut entries, db, target, 1)?;
        }
        for target in src.tx_preplan_stream_targets()? {
            let pos = ensure_participant(&mut entries, db, target, 1)?;
            entries[pos].lock_all = true;
        }
    }
    for op in &rank1_ops {
        ensure_entry(&mut entries, db, op.participant.clone(), 1, &op.key)?;
    }
    // T10 Task 8: an explicit `reserve`/`release` names its namespace
    // directly — there is no source write for a preplan step to discover it
    // through. Add it exactly like an ordinary rank-1 op (its key's shard
    // locked immediately), so `resolve_claims` finds it already present and
    // already locked.
    for ec in claim_ops {
        ensure_entry(&mut entries, db, Arc::clone(&ec.participant), 1, &ec.key)?;
    }
    sort_entries(&mut entries);
    let rank0_len = entries.iter().filter(|e| e.rank == 0).count();
    Ok(PlanOutcome {
        entries,
        rank0_len,
        rank0_ops,
        rank1_ops,
    })
}

pub(crate) fn eval_rank0<'a>(
    entries: &'a [Entry],
    locked: &mut [Box<dyn TxLocked + 'a>],
    ops: Vec<Op>,
) -> DbResult<EvalRank0Outcome> {
    let mut writes: Vec<Write> = Vec::new();
    let mut projection_inputs = Vec::new();
    let mut eval_err = None;
    for op in ops {
        let Some(entry_pos) = idx_for(entries, &op.participant) else {
            eval_err = Some(DbError::Internal("operation lost its participant"));
            break;
        };
        let pending = writes
            .iter()
            .rev()
            .find(|w| w.entry == entry_pos && w.key == op.key)
            .map(|w| w.value.clone());
        if let Action::Insert(bytes) = &op.action {
            let occupied = match pending.as_ref() {
                Some(value) => value.is_some(),
                None => match locked[entry_pos].contains(&op.key) {
                    Ok(value) => value,
                    Err(e) => {
                        eval_err = Some(e);
                        break;
                    }
                },
            };
            if occupied {
                eval_err = Some(DbError::KeyExists);
                break;
            }
            // The presence-only path deliberately does not read the stored
            // payload. `current` is known to be absent here, so projections
            // see the same absent -> present transition as `put`.
            projection_inputs.push((entry_pos, op.key.clone(), None, Some(bytes.clone())));
            set_write(&mut writes, entry_pos, op.key, Some(bytes.clone()), false);
            continue;
        }
        let current: Option<Vec<u8>> = match pending {
            Some(value) => value,
            None => match locked[entry_pos].read(&op.key) {
                Ok(v) => v,
                Err(e) => {
                    eval_err = Some(e);
                    break;
                }
            },
        };
        match op.action {
            Action::Put(bytes) => {
                projection_inputs.push((
                    entry_pos,
                    op.key.clone(),
                    current.clone(),
                    Some(bytes.clone()),
                ));
                set_write(&mut writes, entry_pos, op.key, Some(bytes), false);
            }
            Action::Insert(_) => unreachable!("insert handled by presence-only path"),
            Action::Delete => {
                projection_inputs.push((entry_pos, op.key.clone(), current.clone(), None));
                set_write(&mut writes, entry_pos, op.key, None, false);
            }
            Action::Eval(f) => match f(current.as_deref()) {
                Ok(Upsert::Put(bytes)) => {
                    projection_inputs.push((
                        entry_pos,
                        op.key.clone(),
                        current.clone(),
                        Some(bytes.clone()),
                    ));
                    set_write(&mut writes, entry_pos, op.key, Some(bytes), false);
                }
                Ok(Upsert::Delete) => {
                    projection_inputs.push((entry_pos, op.key.clone(), current.clone(), None));
                    set_write(&mut writes, entry_pos, op.key, None, false);
                }
                Ok(Upsert::Keep) => {}
                Err(e) => {
                    eval_err = Some(e);
                    break;
                }
            },
            Action::Guard(f) => {
                if let Err(e) = f(current.as_deref()) {
                    eval_err = Some(e);
                    break;
                }
            }
        }
    }
    eval_err.map_or(Ok((writes, projection_inputs)), Err)
}

/// One `(target, field, key)` slot of a transaction's counter work: every delta
/// that slot collected, already summed, waiting to be resolved to an absolute
/// value under the target's lock. Two counted sources maintaining two different
/// fields of the **same** target row get one slot each — see `project_rank0`.
///
/// `entry` indexes [`Entry`] (and therefore the parallel `locked` vector), so
/// the resolution round can read the target's current bytes without another
/// lookup.
pub(crate) struct PendingCounter {
    pub entry: usize,
    pub key: Vec<u8>,
    pub target: Arc<dyn ErasedCounterTarget>,
    pub delta: i128,
}

/// One claim slot from the rank-0 round: a claim or release a decided source
/// write emitted into a unique namespace, deferred until the namespace's own
/// lock is held (T10 Task 6) — the claim twin of [`PendingCounter`].
///
/// Unlike [`PendingCounter`], claim slots are **not** coalesced by
/// `(entry, key)` — two claims on one key from the same transaction are
/// exactly the conflict this feature exists to detect, and summing or
/// replacing them the way counters do would erase it. See `project_rank0`.
///
/// **Ordering contract for the resolver.** `PendingClaim::claims` is in
/// emission order, and that order is a *within-row* guarantee only:
/// `project_rank0` diffs each source row once (releases before claims for
/// *that* row — see `into_tx_claim_ops`) and appends the row's ops whole, so
/// across different rows the list interleaves however the caller ordered its
/// writes. `[Claim("rail", id1), Release("rail", id2)]` is a legal emission
/// for a legal hand-off of "rail" from `id2` to `id1` in one transaction.
/// **[`resolve_claims`] partitions the list by [`ClaimKind`] and applies
/// every release before any claim — it does not rely on list order** to get
/// that partition for free.
///
/// `entry` indexes [`Entry`] (and therefore the parallel `locked` vector), so
/// [`resolve_claims`] can read/write the namespace row without another
/// lookup.
///
/// **T10 Task 8:** a `PendingClaim` also arrives from an explicit
/// `reserve`/`release` call ([`pending_claims_from_explicit`], converting
/// [`crate::unique::ExplicitClaim`]) — not only from a projected source's
/// decided write. On a `Reserve` built either way, `kind`'s `deadline_ms`
/// field is the *requested TTL* until `resolve_claims` stamps it into an
/// absolute deadline; see [`crate::txn::participant::ClaimKind::Reserve`]'s
/// own doc.
pub(crate) struct PendingClaim {
    pub entry: usize,
    pub key: Vec<u8>,
    pub owner: crate::unique::UniqueOwner,
    pub kind: crate::txn::participant::ClaimKind,
    pub namespace: &'static str,
}

/// What the rank-0 round derives from the source writes it decided: index rows
/// (already merged into `writes`) plus the shards those rows live on, the
/// counter deltas that still need the targets' locks to become writes, and the
/// claim ops that still need the namespaces' locks to be decided (T10 Task 7).
pub(crate) struct Rank0Derived {
    /// Target entry -> extra shards its derived writes touch, for the rank-1
    /// lock set.
    pub extra_shards: HashMap<usize, Vec<usize>>,
    /// Summed counter deltas, one per `(target, field, key)`, in first-seen
    /// order.
    pub counters: Vec<PendingCounter>,
    /// Claim ops, in emission order — **not** coalesced (see [`PendingClaim`]).
    pub claims: Vec<PendingClaim>,
    /// Collapsed stream transitions, evaluated only after rank-1 queue shards
    /// have been locked. Their post-success callbacks come from
    /// [`lower_streams`], which is the only place a stream is evaluated.
    pub stream_rows: Vec<StreamRow>,
}

/// Reject a derived write into an encrypted collection when the log itself is
/// not encrypted — the record would otherwise carry that collection's values
/// in the clear. Applies to index rows and counter writes alike.
///
/// `pub(crate)`, not private: `Tx::commit` (`txn/mod.rs`, T10 Task 8 review
/// round M2) reuses this for `claim_ops`'s namespace participants instead of
/// a fourth copy of the same message.
pub(crate) fn check_derived_encryption(
    participant: &Arc<dyn TxParticipant>,
    log_encrypted: bool,
) -> DbResult<()> {
    if participant.encrypted() && !log_encrypted {
        return Err(DbError::Config(
            "an encrypted collection needs TxConfig::log_encryption_key: \
             the transaction log would otherwise hold its values in the clear",
        ));
    }
    Ok(())
}

pub(crate) fn project_rank0(
    entries: &[Entry],
    writes: &mut Vec<Write>,
    inputs: Vec<ProjectionInput>,
    log_encrypted: bool,
) -> DbResult<Rank0Derived> {
    let mut extra_shards: HashMap<usize, Vec<usize>> = HashMap::new();
    // `(entry, field, key) -> position in `counters``. Deltas for one target
    // row are summed across the **whole** transaction and resolved exactly
    // once: two source writes contributing to the same counter must produce one
    // absolute value, not two reads of a value that only one of them has
    // updated.
    //
    // The **field** is part of the key, not just the participant and the row:
    // two counted sources may maintain two different fields of the same target
    // row (`SchemaRegistry::finish`'s duplicate rule is keyed
    // `(collection, field)` precisely to allow it). Coalescing on
    // `(entry, key)` alone summed their unrelated deltas into one slot that
    // kept only the first source's `TypedCounterTarget`, so the sum landed on
    // the first field and the second field was never written. Each field's
    // deltas therefore get their own slot; `resolve_counters` composes them,
    // because it reads the row's current bytes from this transaction's own
    // pending write when one already exists.
    let mut counter_slots: HashMap<(usize, &'static str, Vec<u8>), usize> = HashMap::new();
    let mut counters: Vec<PendingCounter> = Vec::new();

    // Claims: collapse this transaction's per-write inputs to **one row per
    // (source entry, source row key)** before diffing — unlike a counter's
    // additive delta, a claim/release pair does not telescope away in a flat
    // ops list. `eval_rank0` chains each write's `old` to the previous
    // write's own pending value, so a row rewritten several times in one
    // transaction (`put(a).put(b).put(a)`) appears in `inputs` as three
    // *consecutive* diffs — (∅,a), (a,b), (b,a) — not one. Diffing each of
    // those independently would emit Claim(a), then Release(a)+Claim(b), then
    // Release(b)+Claim(a): four spurious ops around the row's actual net
    // effect, which is simply Claim(a). So the claim pass diffs the row's
    // *first* old against its *last* new — exactly once per row — while the
    // index/counter passes below still see every write (their per-write
    // model is correct: an index row is a pure function of the current value,
    // and a counter's per-write deltas telescope to the same net effect a
    // single first-to-last diff would give).
    // Review round I2: neither pass below needs to *own* `inputs` — both read
    // `key`/`old`/`new` through a borrow (`tx_project`/`tx_counter_deltas`/
    // `tx_claims` all take `&[u8]`/`Option<&[u8]>`), so both iterate `&inputs`
    // and `claim_rows` holds references into it instead of cloning every key
    // and both value snapshots of every write in every `db.tx()` commit —
    // including ones with no claim half at all. `inputs` outlives both loops
    // (it is this function's own parameter), so the borrows are sound.
    let mut claim_row_slots: HashMap<(usize, &[u8]), usize> = HashMap::new();
    let mut claim_rows: Vec<ClaimRowRef<'_>> = Vec::new();
    for (src, key, old, new) in &inputs {
        let key: &[u8] = key.as_slice();
        match claim_row_slots.entry((*src, key)) {
            std::collections::hash_map::Entry::Occupied(slot) => {
                claim_rows[*slot.get()].3 = new.as_deref();
            }
            std::collections::hash_map::Entry::Vacant(slot) => {
                slot.insert(claim_rows.len());
                claim_rows.push((*src, key, old.as_deref(), new.as_deref()));
            }
        }
    }

    for (src, key, old, new) in &inputs {
        let src = *src;
        let derived = entries[src]
            .participant
            .tx_project(key, old.as_deref(), new.as_deref())?;
        for row in derived {
            check_derived_encryption(&row.participant, log_encrypted)?;
            let Some(target_pos) = idx_for(entries, &row.participant) else {
                return Err(DbError::UndeclaredIndexTarget {
                    index: row.participant.name(),
                });
            };
            let sid = row.participant.shard_of(&row.key)?;
            extra_shards.entry(target_pos).or_default().push(sid);
            set_write(writes, target_pos, row.key, row.value, true);
        }
        let deltas =
            entries[src]
                .participant
                .tx_counter_deltas(key, old.as_deref(), new.as_deref())?;
        for d in deltas {
            check_derived_encryption(&d.participant, log_encrypted)?;
            let Some(target_pos) = idx_for(entries, &d.participant) else {
                return Err(DbError::UndeclaredCounterTarget {
                    target: d.participant.name(),
                });
            };
            let sid = d.participant.shard_of(&d.key)?;
            extra_shards.entry(target_pos).or_default().push(sid);
            match counter_slots.entry((target_pos, d.target.field(), d.key.clone())) {
                std::collections::hash_map::Entry::Occupied(slot) => {
                    let pos = *slot.get();
                    counters[pos].delta = counters[pos]
                        .delta
                        .checked_add(d.delta)
                        .ok_or(DbError::Internal("counter delta sum overflow"))?;
                }
                std::collections::hash_map::Entry::Vacant(slot) => {
                    slot.insert(counters.len());
                    counters.push(PendingCounter {
                        entry: target_pos,
                        key: d.key,
                        target: d.target,
                        delta: d.delta,
                    });
                }
            }
        }
    }

    // Claims, one `tx_claims` call per collapsed row (see the comment on
    // `claim_rows` above) — the ops themselves are **not** coalesced, unlike
    // the counter loop above: two claims on one key from this transaction
    // are the conflict the resolver (T10 Task 7) exists to catch, and
    // summing or replacing them here would erase it before the resolver
    // ever sees it.
    let mut claims: Vec<PendingClaim> = Vec::new();
    // Only a source that actually declares a stream needs its rows materialized
    // for `lower_streams`: this is the one place the commit path has to *own*
    // key and value snapshots rather than borrow them, so building it for every
    // transaction would reintroduce exactly the per-write clone the claim pass
    // above was rewritten to avoid (review round I2).
    let mut emitting_sources: Vec<bool> = Vec::with_capacity(entries.len());
    for entry in entries.iter() {
        emitting_sources.push(entry.participant.tx_emits()?);
    }
    let stream_rows = claim_rows
        .iter()
        .filter(|(src, ..)| emitting_sources.get(*src).copied().unwrap_or(false))
        .map(|(src, key, old, new)| {
            (
                *src,
                key.to_vec(),
                old.map(<[u8]>::to_vec),
                new.map(<[u8]>::to_vec),
            )
        })
        .collect();
    for (src, key, old, new) in claim_rows {
        let claim_ops = entries[src].participant.tx_claims(key, old, new)?;
        for op in claim_ops {
            check_derived_encryption(&op.participant, log_encrypted)?;
            let Some(target_pos) = idx_for(entries, &op.participant) else {
                return Err(DbError::UndeclaredUniqueTarget {
                    namespace: op.namespace,
                });
            };
            let sid = op.participant.shard_of(&op.key)?;
            extra_shards.entry(target_pos).or_default().push(sid);
            claims.push(PendingClaim {
                entry: target_pos,
                key: op.key,
                owner: op.owner,
                kind: op.kind,
                namespace: op.namespace,
            });
        }
    }

    Ok(Rank0Derived {
        extra_shards,
        counters,
        claims,
        stream_rows,
    })
}

/// Lower streams while every queue destination is locked. Their keys are
/// allocated by lowering, hence planning marked these participants `lock_all`.
pub(crate) fn lower_streams(
    entries: &[Entry],
    writes: &mut Vec<Write>,
    rows: Vec<StreamRow>,
    log_encrypted: bool,
    admissions: &mut crate::queue::QueueDepthAdmissions,
) -> DbResult<StreamLowered> {
    let mut after_success = Vec::new();
    for (src, key, old, new) in rows {
        let mut stream =
            entries[src]
                .participant
                .tx_stream_writes(&key, old.as_deref(), new.as_deref())?;
        for admission in stream.queue_depth_admissions.drain(..) {
            admission.record(admissions);
        }
        for row in stream.writes.drain(..) {
            check_derived_encryption(&row.participant, log_encrypted)?;
            let target_pos =
                idx_for(entries, &row.participant).ok_or(DbError::UndeclaredIndexTarget {
                    index: row.participant.name(),
                })?;
            set_write(writes, target_pos, row.key, row.value, true);
        }
        after_success.append(&mut stream.after_success);
    }
    Ok(after_success)
}

/// Render a [`crate::unique::UniqueOwner`] for [`DbError::UniqueViolation`].
/// This layer only knows the claimant's opaque 64-bit collection tag and its
/// own row key (`unique::row::collection_tag`) — not the source collection's
/// name or type — so that is what gets rendered; `key` is hex via
/// `crate::hex::hex`, the same renderer the counter errors use.
fn render_owner(owner: &crate::unique::UniqueOwner) -> String {
    let reservation = if owner.is_reservation() {
        " (reservation)"
    } else {
        ""
    };
    format!(
        "collection_tag={:#x} key={}{reservation}",
        owner.collection_id(),
        crate::hex::hex(owner.key_bytes())
    )
}

/// Decode a namespace row from its stored bytes. The namespace collection is
/// a fixed-size `ZeroMap<_, { size_of::<UniqueRow>() }, _, _, Fixed>`
/// (`unique::index::Inner`), so a length mismatch here means the row is
/// corrupt, not merely "a different type" — reported the same way the Zero
/// hook adapter reports its own decode failures (`hook.rs`).
///
/// Also rejects a row that decodes but is not [well-formed]
/// (`UniqueRow::is_well_formed`) — `key_len` wider than
/// `UNIQUE_OWNER_KEY_CAP` — the same check the public read path
/// (`UniqueIndex::owner_of`) already applies. Skipping it here would let
/// `owner.key_bytes()` slice past the end of its fixed buffer later in this
/// module (`render_owner`, on the `Conflict` path), panicking with every
/// namespace shard's lock still held — the review round's I4.
///
/// [well-formed]: crate::unique::UniqueRow::is_well_formed
fn decode_unique_row(bytes: &[u8]) -> DbResult<UniqueRow> {
    let row: UniqueRow = zerocopy::FromBytes::read_from_bytes(bytes)
        .map_err(|_| DbError::FormatMismatch("unique namespace row decode failed".into()))?;
    if !row.is_well_formed() {
        return Err(DbError::FormatMismatch(
            "unique namespace row is corrupt: key_len exceeds its own key buffer".into(),
        ));
    }
    Ok(row)
}

/// Turn this transaction's explicit `reserve`/`release` ops
/// ([`crate::unique::ExplicitClaim`], T10 Task 8) into [`PendingClaim`]s —
/// the explicit-op twin of how the claims loop inside [`project_rank0`]
/// turns a projected source's [`crate::txn::participant::ClaimOp`]s into the
/// same shape. Each explicit op's namespace participant was already added to
/// `entries` by [`plan`] (with its key's shard already in that entry's lock
/// set), so this only resolves the `entry` index [`resolve_claims`] needs to
/// read/write it — the same [`idx_for`] lookup `project_rank0` uses for a
/// projected claim's target.
pub(crate) fn pending_claims_from_explicit(
    entries: &[Entry],
    claim_ops: Vec<crate::unique::ExplicitClaim>,
) -> DbResult<Vec<PendingClaim>> {
    let mut out = Vec::with_capacity(claim_ops.len());
    for ec in claim_ops {
        let Some(entry) = idx_for(entries, &ec.participant) else {
            return Err(DbError::Internal(
                "explicit reserve/release lost its namespace participant",
            ));
        };
        out.push(PendingClaim {
            entry,
            key: ec.key,
            owner: ec.owner,
            kind: ec.kind,
            namespace: ec.namespace,
        });
    }
    Ok(out)
}

/// Decide every pending claim/release under the namespace's own lock (T10
/// Task 7), turning [`project_rank0`]'s collected [`PendingClaim`]s — merged
/// with [`pending_claims_from_explicit`]'s (T10 Task 8) — into namespace
/// writes, or refusing the whole commit with [`DbError::UniqueViolation`].
///
/// Runs **after** [`eval_rank1`] and **before** [`resolve_counters`]: a
/// conflict is cheaper to discover before counter arithmetic runs, and a
/// refused transaction must not have run any.
///
/// **Two-level resolution.** [`project_rank0`] already collapsed each source
/// row's writes to its net claim/release, once per row (Task 6's job). This
/// pass does the second level: partition the remaining list **by kind** and
/// resolve every release before any claim. [`PendingClaim`]'s own doc spells
/// out why list order alone does not give that split for free. Resolving
/// releases first is what lets a same-transaction hand-off succeed (release
/// one row's claim, then claim it for another): the claim's overlay read
/// then sees the release's own write to the same row, rather than rejecting
/// itself against a row this same transaction is about to remove.
///
/// **T10 Task 8: explicit acquisitions resolve before projected ones.** The
/// caller ([`crate::txn::Tx::commit`]) places every [`PendingClaim`] built
/// from an explicit `reserve` ahead of every one `project_rank0` derived from
/// a source write, so within the "acquisitions" half above an explicit
/// reservation for `(namespace, key, owner)` always resolves *before* that
/// same owner's own row claiming the same key in the same transaction. That
/// ordering — not the order the caller happened to chain `.reserve()`/`.put()`
/// on the builder — is what makes the takeover land correctly regardless of
/// call order: the reservation writes first, then the claim's "owner is me,
/// has a deadline" branch (`unique::resolve::decide`) rewrites it permanent.
/// The opposite order is unsound: `decide`'s `Reserve` arm on an
/// already-owned row unconditionally refreshes the deadline (unlike its
/// `Claim` arm, which checks `is_reservation()` first), so a reservation
/// resolving *after* its own takeover claim would silently turn a permanent
/// claim back into an expiring one.
///
/// A `Reserve`'s `deadline_ms` is **not yet absolute** when it reaches this
/// function (see [`PendingClaim`]'s doc) — it is stamped here, from **one**
/// clock read shared by every reservation this call resolves (taken lazily,
/// so a transaction with none never reads the clock at all), immediately
/// before the kind is handed to [`decide`].
pub(crate) fn resolve_claims<'a>(
    locked: &mut [Box<dyn TxLocked + 'a>],
    claims: Vec<PendingClaim>,
    writes: &mut Vec<Write>,
) -> DbResult<()> {
    let (releases, acquisitions): (Vec<_>, Vec<_>) = claims
        .into_iter()
        .partition(|c| matches!(c.kind, ClaimKind::Release));

    let mut now_ms: Option<u64> = None;

    for c in releases.into_iter().chain(acquisitions) {
        // Overlay first, storage only otherwise — the same rule
        // `resolve_counters` uses (see its own doc below); it is what lets
        // an earlier op in *this* loop compose with a later one on the same
        // row, which the hand-off case above depends on.
        let current: Option<Vec<u8>> = match writes
            .iter()
            .rev()
            .find(|w| w.entry == c.entry && w.key == c.key)
        {
            Some(w) => w.value.clone(),
            None => locked
                .get_mut(c.entry)
                .ok_or(DbError::Internal("unique namespace lost its lock"))?
                .read(&c.key)?,
        };
        let row = current.as_deref().map(decode_unique_row).transpose()?;
        let kind = match c.kind {
            ClaimKind::Reserve {
                deadline_ms: ttl_ms,
            } => {
                let now = *now_ms.get_or_insert_with(crate::unique::reservation_now_ms);
                let deadline_ms =
                    now.checked_add(ttl_ms)
                        .ok_or(DbError::UniqueReservationInvalid {
                            namespace: c.namespace,
                            reason: "deadline overflow",
                        })?;
                ClaimKind::Reserve { deadline_ms }
            }
            other => other,
        };
        match decide(row, &c.owner, kind) {
            ClaimDecision::Write(row) => {
                set_write(writes, c.entry, c.key, Some(row.as_bytes().to_vec()), true)
            }
            ClaimDecision::Delete => set_write(writes, c.entry, c.key, None, true),
            ClaimDecision::NoOp => {}
            ClaimDecision::Conflict(owner) => {
                return Err(DbError::UniqueViolation {
                    namespace: c.namespace,
                    key: crate::hex::hex(&c.key),
                    owner: render_owner(&owner),
                });
            }
        }
    }
    Ok(())
}

/// Turn each summed counter delta into an absolute write, under the target's
/// own lock (T9 Task 9).
///
/// Runs **after** [`eval_rank1`], which is what makes a counter see the result
/// of a user write to the same row in the same transaction rather than the
/// value before it: the current value is taken from this transaction's own
/// pending write when there is one, and only otherwise from storage.
///
/// Reading this transaction's own pending write first is also what lets two
/// counter *fields* of the same target row compose: the second field resolves
/// against the bytes the first one just produced, so both land in the single
/// logged value for that row.
///
/// The result is an ordinary put/delete in the same log record as the source
/// write — a delta never reaches the log, so the record format is unchanged and
/// replay stays exactly-once (no dirty marker, no recount, no window).
pub(crate) fn resolve_counters<'a>(
    locked: &mut [Box<dyn TxLocked + 'a>],
    counters: Vec<PendingCounter>,
    writes: &mut Vec<Write>,
) -> DbResult<()> {
    for c in counters {
        let current: Option<Vec<u8>> = match writes
            .iter()
            .rev()
            .find(|w| w.entry == c.entry && w.key == c.key)
        {
            Some(w) => w.value.clone(),
            None => locked
                .get_mut(c.entry)
                .ok_or(DbError::Internal("counter target lost its lock"))?
                .read(&c.key)?,
        };
        // `resolve` never sees the row key, so its two errors carry a
        // `"<resolve>"` placeholder — rewrite it here, where the key is known.
        match c
            .target
            .resolve(current.as_deref(), c.delta)
            .map_err(|e| e.with_key(&c.key))?
        {
            CounterApply::Put(bytes) => set_write(writes, c.entry, c.key, Some(bytes), true),
            CounterApply::Delete => set_write(writes, c.entry, c.key, None, true),
            CounterApply::Skip => {}
        }
    }
    Ok(())
}

/// Decide the writes of the rank-1 (target) round.
///
/// **Rank-1 participants do not run their own projections.** Unlike
/// [`eval_rank0`], this collects no `projection_inputs`: a target's own
/// secondary-index rows and counter deltas would need a third lock round to
/// reach *their* targets, which the two-round protocol does not have. That is
/// why [`plan`] refuses a transaction that writes directly to a target which is
/// itself a projecting or counting source ([`DbError::TxTargetIsAlsoASource`])
/// — the alternative is silent drift.
pub(crate) fn eval_rank1<'a>(
    entries: &'a [Entry],
    locked: &mut [Box<dyn TxLocked + 'a>],
    rank0_len: usize,
    ops: Vec<Op>,
    writes: &mut Vec<Write>,
) -> DbResult<()> {
    for op in ops {
        let Some(entry_pos) = idx_for(entries, &op.participant) else {
            return Err(DbError::Internal("operation lost its participant"));
        };
        debug_assert!(entry_pos >= rank0_len);
        let locked_pos = entry_pos;
        let pending = writes
            .iter()
            .rev()
            .find(|w| w.entry == entry_pos && w.key == op.key)
            .map(|w| w.value.clone());
        if let Action::Insert(bytes) = &op.action {
            let occupied = match pending.as_ref() {
                Some(value) => value.is_some(),
                None => locked[locked_pos].contains(&op.key)?,
            };
            if occupied {
                return Err(DbError::KeyExists);
            }
            set_write(writes, entry_pos, op.key, Some(bytes.clone()), false);
            continue;
        }
        let current: Option<Vec<u8>> = match pending {
            Some(value) => value,
            None => locked[locked_pos].read(&op.key)?,
        };
        match op.action {
            Action::Put(bytes) => set_write(writes, entry_pos, op.key, Some(bytes), false),
            Action::Insert(_) => unreachable!("insert handled by presence-only path"),
            Action::Delete => set_write(writes, entry_pos, op.key, None, false),
            Action::Eval(f) => match f(current.as_deref()) {
                Ok(Upsert::Put(bytes)) => set_write(writes, entry_pos, op.key, Some(bytes), false),
                Ok(Upsert::Delete) => set_write(writes, entry_pos, op.key, None, false),
                Ok(Upsert::Keep) => {}
                Err(e) => return Err(e),
            },
            Action::Guard(f) => f(current.as_deref())?,
        }
    }
    Ok(())
}

pub(crate) fn admit(
    db: &crate::txn::TxContext<'_>,
    limits: crate::txn::TxLimits,
    entries: &[Entry],
    writes: &[Write],
) -> DbResult<()> {
    for entry in entries {
        db.collection_id(entry.participant.as_ref())?;
    }
    if writes.len() > limits.max_ops {
        let has_derived = writes.iter().any(|w| w.derived);
        return Err(DbError::Client(if has_derived {
            "transaction exceeds max_ops (including derived secondary-index, counter and queue rows)"
        } else {
            "transaction exceeds max_ops"
        }));
    }
    for w in writes {
        let value_len = w.value.as_deref().map_or(0, <[u8]>::len);
        if let Some(cap) = entries[w.entry].participant.max_entry_bytes()
            && crate::entry::entry_size(w.key.len(), value_len as u32) as usize > cap
        {
            return Err(DbError::Client(if w.derived {
                "derived secondary-index or counter row exceeds write_buffer_size"
            } else {
                "transaction entry exceeds write_buffer_size"
            }));
        }
    }
    Ok(())
}

pub(crate) fn max_bytes_err(writes: &[Write]) -> DbError {
    let has_derived = writes.iter().any(|w| w.derived);
    DbError::Client(if has_derived {
        "transaction exceeds max_bytes (including derived secondary-index and counter rows)"
    } else {
        "transaction exceeds max_bytes"
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::txlog::record::{OpKind, VersionTag};
    use crate::txn::participant::{ClaimOp, DeferredReplay, PlannedOp, TxBackend};
    use crate::unique::{UniqueOwner, UniqueRow};

    /// Stub namespace participant — good enough to be `idx_for`-matched and
    /// have its shard set recorded; never locked or written to by these
    /// tests. Mirrors `counter::tests::StubTxParticipant`/
    /// `projection::tests::StubParticipant`.
    struct StubNamespace;
    impl TxParticipant for StubNamespace {
        fn name(&self) -> &'static str {
            "stub_namespace"
        }
        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 TxLocked + 'a>> {
            Err(DbError::Internal("stub"))
        }
    }

    /// Stub source: diffs the `old`/`new` byte snapshots `project_rank0` hands
    /// it into claim/release ops on `StubNamespace`, using the bytes
    /// themselves as the claimed namespace key. Real enough to exercise
    /// `project_rank0`'s row-collapsing (F2) without pulling in
    /// `claim_delta`/`Projected`/a real `UniqueIndex`.
    struct StubClaimSource {
        namespace: Arc<dyn TxParticipant>,
    }
    impl TxParticipant for StubClaimSource {
        fn name(&self) -> &'static str {
            "stub_claim_source"
        }
        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 TxLocked + 'a>> {
            Err(DbError::Internal("stub"))
        }
        fn tx_claims(
            &self,
            _key: &[u8],
            old: Option<&[u8]>,
            new: Option<&[u8]>,
        ) -> DbResult<Vec<ClaimOp>> {
            let mut ops = Vec::new();
            if old != new {
                let owner = UniqueOwner::new(1, b"src", 0).unwrap();
                if let Some(o) = old {
                    ops.push(ClaimOp {
                        participant: Arc::clone(&self.namespace),
                        namespace: "stub_namespace",
                        key: o.to_vec(),
                        owner,
                        kind: ClaimKind::Release,
                    });
                }
                if let Some(n) = new {
                    ops.push(ClaimOp {
                        participant: Arc::clone(&self.namespace),
                        namespace: "stub_namespace",
                        key: n.to_vec(),
                        owner,
                        kind: ClaimKind::Claim,
                    });
                }
            }
            Ok(ops)
        }
    }

    /// `entries[0]` is the claiming source, `entries[1]` is its namespace —
    /// the shape `project_rank0` expects (`idx_for` matches by `Arc::ptr_eq`).
    fn stub_entries() -> Vec<Entry> {
        let namespace: Arc<dyn TxParticipant> = Arc::new(StubNamespace);
        let source: Arc<dyn TxParticipant> = Arc::new(StubClaimSource {
            namespace: Arc::clone(&namespace),
        });
        vec![
            Entry {
                id: 0,
                rank: 0,
                participant: source,
                shards: vec![0],
                lock_all: false,
            },
            Entry {
                id: 1,
                rank: 1,
                participant: namespace,
                shards: Vec::new(),
                lock_all: false,
            },
        ]
    }

    /// F1 (review): two *different* source rows independently claiming the
    /// same namespace key must both reach the plan, uncoalesced — a resolver
    /// that only ever sees one of them cannot detect the conflict this
    /// feature exists to catch. This is the one invariant that makes the
    /// claims loop different from the counter loop right above it in this
    /// file; a future "harmonisation" that summed/replaced by `(entry, key)`
    /// the way counters do would pass every other test in the crate while
    /// silently breaking exactly this.
    #[test]
    fn project_rank0_does_not_coalesce_two_rows_claiming_the_same_key() {
        let entries = stub_entries();
        let inputs: Vec<ProjectionInput> = vec![
            (0, vec![1], None, Some(b"rail".to_vec())),
            (0, vec![2], None, Some(b"rail".to_vec())),
        ];
        let mut writes = Vec::new();
        let derived = project_rank0(&entries, &mut writes, inputs, false).unwrap();

        assert_eq!(
            derived.claims.len(),
            2,
            "two source rows claiming \"rail\" must produce two claim ops, not one"
        );
        assert!(
            derived
                .claims
                .iter()
                .all(|c| c.kind == ClaimKind::Claim && c.key == b"rail".to_vec()),
            "both ops must be claims on \"rail\": {:?}",
            derived
                .claims
                .iter()
                .map(|c| (c.kind, c.key.clone()))
                .collect::<Vec<_>>()
        );
        assert!(
            derived.extra_shards.contains_key(&1),
            "the namespace entry (index 1) must be in extra_shards for the rank-1 lock set"
        );
    }

    /// F2 (review, real bug fix): a row rewritten several times in one
    /// transaction must diff as **one** net change, not once per write.
    /// `eval_rank0` chains each write's `old` to the previous write's own
    /// pending value, so `put(a).put(b).put(a)` reaches `project_rank0` as
    /// three consecutive `ProjectionInput`s — (∅,a), (a,b), (b,a)`) — whose
    /// *union*, if diffed independently, is Release(a)+Claim(b)+Release(b)+
    /// Claim(a) around the actual net effect, Claim(a). `derived.claims` must
    /// be exactly that one op, and must not mention `"b"` at all — it was
    /// only ever an intermediate value inside this one transaction.
    #[test]
    fn project_rank0_collapses_one_rows_chained_writes_before_diffing() {
        let entries = stub_entries();
        let inputs: Vec<ProjectionInput> = vec![
            (0, vec![9], None, Some(b"a".to_vec())),
            (0, vec![9], Some(b"a".to_vec()), Some(b"b".to_vec())),
            (0, vec![9], Some(b"b".to_vec()), Some(b"a".to_vec())),
        ];
        let mut writes = Vec::new();
        let derived = project_rank0(&entries, &mut writes, inputs, false).unwrap();

        assert_eq!(
            derived.claims.len(),
            1,
            "the row's net effect is one claim, not one op per write: {:?}",
            derived
                .claims
                .iter()
                .map(|c| (c.kind, c.key.clone()))
                .collect::<Vec<_>>()
        );
        assert_eq!(derived.claims[0].kind, ClaimKind::Claim);
        assert_eq!(derived.claims[0].key, b"a".to_vec());
        assert!(
            derived.claims.iter().all(|c| c.key != b"b".to_vec()),
            "no op may mention \"b\" — it was only ever an intermediate value"
        );
    }

    // -- resolve_claims (T10 Task 7) ---------------------------------------

    /// Minimal [`TxLocked`] stub for `resolve_claims` unit tests: one
    /// namespace's storage as a plain `HashMap`. `resolve_claims` only ever
    /// calls `read` on a locked participant — every other method is
    /// unreachable from these tests and panics if that ever changes.
    struct StubClaimLocked {
        store: std::collections::HashMap<Vec<u8>, Vec<u8>>,
    }
    impl TxLocked for StubClaimLocked {
        fn read(&mut self, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
            Ok(self.store.get(key).cloned())
        }
        fn contains(&mut self, _key: &[u8]) -> DbResult<bool> {
            unreachable!("resolve_claims does not call contains")
        }
        fn reserve(&mut self, _ops: &[PlannedOp<'_>]) -> DbResult<Vec<VersionTag>> {
            unreachable!("resolve_claims does not reserve")
        }
        fn apply_put(&mut self, _key: &[u8], _value: &[u8], _tag: VersionTag) -> DbResult<usize> {
            unreachable!("resolve_claims does not apply")
        }
        fn apply_delete(&mut self, _key: &[u8], _tag: VersionTag) -> DbResult<usize> {
            unreachable!("resolve_claims does not apply")
        }
        fn needs_replay(&mut self, _key: &[u8], _kind: OpKind, _tag: VersionTag) -> DbResult<bool> {
            unreachable!("resolve_claims does not replay")
        }
        fn mark_logged(&mut self) {}
        fn abort(self: Box<Self>) {}
        #[cfg(test)]
        fn unapplied_reservations(&self) -> usize {
            0
        }
        #[cfg(test)]
        fn tx_debug_arm(&mut self, _key: &[u8], _tag: VersionTag) -> DbResult<()> {
            Ok(())
        }
        fn finish(
            self: Box<Self>,
            _fenced: &mut Vec<(usize, u64)>,
            _wakes: &mut crate::watch::WatchWakeBatch,
        ) -> DbResult<Box<dyn DeferredReplay>> {
            unreachable!("resolve_claims does not finish")
        }
    }

    fn claim_locked(rows: Vec<(&[u8], UniqueOwner)>) -> Vec<Box<dyn TxLocked>> {
        let mut store = std::collections::HashMap::new();
        for (key, owner) in rows {
            store.insert(key.to_vec(), UniqueRow::new(owner).as_bytes().to_vec());
        }
        vec![Box::new(StubClaimLocked { store })]
    }

    fn pending_claim(
        entry: usize,
        key: &[u8],
        owner: UniqueOwner,
        kind: ClaimKind,
    ) -> PendingClaim {
        PendingClaim {
            entry,
            key: key.to_vec(),
            owner,
            kind,
            namespace: "stub_namespace",
        }
    }

    /// The load-bearing regression this whole task exists to pin: a
    /// **list-order** `[Claim, Release]` on the same key — exactly the shape
    /// `PendingClaim`'s own doc calls out as a legal hand-off emission —
    /// must still resolve as "release, then claim", not in list order. If
    /// `resolve_claims` ever went back to resolving in list order, this is
    /// the test that would catch it: the claim would see the row still held
    /// by its old owner and reject itself with `UniqueViolation`.
    #[test]
    fn resolve_claims_applies_releases_before_claims_regardless_of_list_order() {
        let old_owner = UniqueOwner::new(1, b"users:1", 0).unwrap();
        let new_owner = UniqueOwner::new(1, b"users:2", 0).unwrap();
        let mut locked = claim_locked(vec![(b"rail".as_slice(), old_owner)]);

        // Claim before Release in list order — the opposite of what
        // resolution must do.
        let claims = vec![
            pending_claim(0, b"rail", new_owner, ClaimKind::Claim),
            pending_claim(0, b"rail", old_owner, ClaimKind::Release),
        ];
        let mut writes = Vec::new();
        resolve_claims(&mut locked, claims, &mut writes).unwrap();

        assert_eq!(
            writes.len(),
            1,
            "the release and the claim collapse to one write on \"rail\""
        );
        let value = writes[0]
            .value
            .as_ref()
            .expect("the claim must leave the row written, not deleted");
        let row: UniqueRow = zerocopy::FromBytes::read_from_bytes(value.as_slice()).unwrap();
        assert_eq!(
            row.owner(),
            new_owner,
            "the claim must win once the release is resolved first"
        );
    }

    /// The rule's other load-bearing case, pinned again at the resolver
    /// level (not just `decide`'s own unit tests in `unique::resolve`): a
    /// release must never delete a row a different owner already holds.
    #[test]
    fn resolve_claims_refuses_a_release_of_a_foreign_row_and_leaves_it_untouched() {
        let foreign = UniqueOwner::new(1, b"users:99", 0).unwrap();
        let mine = UniqueOwner::new(1, b"users:1", 0).unwrap();
        let mut locked = claim_locked(vec![(b"rail".as_slice(), foreign)]);

        let claims = vec![pending_claim(0, b"rail", mine, ClaimKind::Release)];
        let mut writes = Vec::new();
        let err = resolve_claims(&mut locked, claims, &mut writes).unwrap_err();
        assert!(
            matches!(
                err,
                DbError::UniqueViolation {
                    namespace: "stub_namespace",
                    ..
                }
            ),
            "{err}"
        );
        assert!(writes.is_empty(), "a foreign row must not be touched");
    }

    /// Two different source rows claiming the same key in one transaction:
    /// exactly one wins, the other conflicts against the write the first
    /// one just made (overlay-first, per `resolve_claims`'s own doc).
    #[test]
    fn resolve_claims_lets_the_first_of_two_claims_on_one_key_win() {
        let first = UniqueOwner::new(1, b"users:1", 0).unwrap();
        let second = UniqueOwner::new(2, b"channels:1", 0).unwrap();
        let mut locked = claim_locked(vec![]);

        let claims = vec![
            pending_claim(0, b"dup", first, ClaimKind::Claim),
            pending_claim(0, b"dup", second, ClaimKind::Claim),
        ];
        let mut writes = Vec::new();
        let err = resolve_claims(&mut locked, claims, &mut writes).unwrap_err();
        assert!(matches!(err, DbError::UniqueViolation { .. }), "{err}");
        assert_eq!(
            writes.len(),
            1,
            "the first claim's write must still stand even though the transaction as a whole \
             is refused by `resolve_claims`'s caller — `commit()` is what discards it"
        );
    }

    /// Review round I4: a corrupt namespace row (`key_len` past its own
    /// buffer) must fail cleanly, not slice out of bounds and panic while
    /// every namespace shard's lock is still held. `decode_unique_row` must
    /// reject it before `decide` ever sees it — if it did not, a `Conflict`
    /// decision's `render_owner` would index `owner.key_bytes()` with the
    /// corrupt `key_len` against a fixed 56-byte buffer.
    #[test]
    fn resolve_claims_rejects_a_corrupt_row_instead_of_panicking() {
        let mut corrupt = vec![0u8; size_of::<UniqueRow>()];
        // `UniqueRow` is `#[repr(C)]`: `expires_at: u64` (8) + `collection_id:
        // u64` (8) puts `key_len: u8` at offset 16. `UNIQUE_OWNER_KEY_CAP` is
        // 56, so 200 is well past it.
        corrupt[16] = 200;
        let mut store = std::collections::HashMap::new();
        store.insert(b"rail".to_vec(), corrupt);
        let mut locked: Vec<Box<dyn TxLocked>> = vec![Box::new(StubClaimLocked { store })];

        let claimant = UniqueOwner::new(1, b"users:1", 0).unwrap();
        let claims = vec![pending_claim(0, b"rail", claimant, ClaimKind::Claim)];
        let mut writes = Vec::new();
        let err = resolve_claims(&mut locked, claims, &mut writes).unwrap_err();
        assert!(matches!(err, DbError::FormatMismatch(_)), "{err}");
        assert!(writes.is_empty());
    }
}