armdb 0.8.0

//! Crash-atomic multi-collection transaction (T5, part 2).
//!
//! Spec: `docs/superpowers/specs/26-08-05-crash-atomic-multi-collection-tx.md`.
//! The commit carrier lives in [`crate::txlog`]; this module is the seam that
//! lets a heterogeneous set of collections be committed through it.

use std::sync::Arc;

use crate::error::{DbError, DbResult};
use crate::key::Key;
use crate::txlog::record::{OpKind, TxOp, TxRecord, VersionTag};
use crate::txn::participant::{DeferredReplay, PlannedOp, TxLocked, TxParticipant};
use crate::txn_depth::ATOMIC_DEPTH;
use crate::upsert::{IntoUpsert, Upsert};

pub(crate) mod checkpoint;
pub(crate) mod families;
pub(crate) mod held;
pub(crate) mod participant;
pub(crate) mod replay;

/// Crash stages for tests. The child process sets one and `_exit`s when the
/// commit path reaches it, so the parent can reopen and assert.
#[cfg(feature = "test-faults")]
pub mod failpoints {
    pub const NONE: u8 = 0;
    pub const AFTER_LOG_APPEND: u8 = 1; // record durable, nothing applied
    pub const MID_APPLY: u8 = 2; // some operations applied
    pub const AFTER_APPLY: u8 = 3; // applied, no checkpoint
    pub static STAGE: std::sync::atomic::AtomicU8 = std::sync::atomic::AtomicU8::new(NONE);

    pub(crate) fn hit(stage: u8) {
        if STAGE.load(std::sync::atomic::Ordering::SeqCst) == stage {
            // No destructors, no flush, no hint file — a real crash.
            unsafe { libc::_exit(9) }
        }
    }
}

/// Configuration for crash-atomic multi-collection transactions.
#[derive(Debug, Clone, Copy)]
pub struct TxConfig {
    /// Starting value of a transaction's `durable` flag, i.e. whether
    /// `commit()` waits for the log fsync. **Off** by default: a commit is then
    /// atomic against a process crash but not against power loss. Call
    /// [`Tx::durable`] per transaction, or set this to make it the default.
    pub durable_by_default: bool,
    pub segment_bytes: u64,
    pub checkpoint_bytes: u64,
    pub max_ops: usize,
    pub max_bytes: usize,
    /// Drop pending records whose schema fingerprint no longer matches instead
    /// of failing the open. Off — losing a committed transaction silently is
    /// worse than refusing to start.
    ///
    /// The drop is not atomic either: collections open one at a time, so a
    /// sibling opened before the mismatch was noticed has already replayed its
    /// share of the record, and dropping it then freezes a **partially applied**
    /// transaction. Turn this on only to get a database to start at all.
    pub discard_pending_on_schema_mismatch: bool,
    /// Key for the transaction log's record envelope. Required as soon as an
    /// **encrypted** collection takes part in a transaction: the log would
    /// otherwise hold that collection's values in the clear. Independent of the
    /// per-collection `Config::encryption_key` — one log serves every
    /// collection, so it has its own key and its own salt.
    #[cfg(feature = "encryption")]
    pub log_encryption_key: Option<[u8; 32]>,
}

impl Default for TxConfig {
    fn default() -> Self {
        Self {
            durable_by_default: false,
            segment_bytes: 16 << 20,
            checkpoint_bytes: 64 << 20,
            max_ops: 1024,
            max_bytes: 8 << 20,
            discard_pending_on_schema_mismatch: false,
            #[cfg(feature = "encryption")]
            log_encryption_key: None,
        }
    }
}

/// A collection that a transaction can name. Implemented for the six managed
/// families; the codec methods run at the *call site*, where the value type is
/// still known — everything below the builder is bytes.
///
/// **Sealed by construction, not by convention.** The trait is `pub` because
/// `Tx::put` and friends are, but it lives in a private module and every
/// builder method also requires the crate-private `TxParticipant`, so it cannot
/// be implemented or named outside this crate — you cannot write a generic
/// function bounded on it. The set of participants is fixed by design: a
/// transaction has to know each one's shard map, GSN counter and mutation seam.
pub trait TxTarget: Send + Sync + 'static {
    type Key: Key;
    type Val;
    fn tx_encode(&self, value: &Self::Val, out: &mut Vec<u8>) -> DbResult<()>;
    fn tx_decode(&self, bytes: &[u8]) -> DbResult<Self::Val>;
}

type UpsertBytes = Upsert<Vec<u8>>;
pub(crate) type TxEvalFn = Box<dyn FnOnce(Option<&[u8]>) -> DbResult<UpsertBytes>>;
pub(crate) type TxGuardFn = Box<dyn FnOnce(Option<&[u8]>) -> DbResult<()>>;

pub(crate) enum Action {
    Put(Vec<u8>),
    Delete,
    /// `upsert`: decide under the lock from the stored bytes.
    Eval(TxEvalFn),
    /// `guard`: validate under the lock; may abort the whole commit.
    Guard(TxGuardFn),
}

pub(crate) struct Op {
    pub participant: Arc<dyn TxParticipant>,
    pub key: Vec<u8>,
    pub action: Action,
}

/// Builder + commit for a crash-atomic multi-collection transaction.
///
/// Nothing is written until [`commit`](Self::commit) reaches its log append:
/// an error from any step before that — including a closure's `Err` — leaves
/// no value write, no index mutation and no hook.
/// Suppress the apply phase after a successful log append (tests only): the
/// on-disk state is then exactly what a crash between the commit marker and the
/// first write leaves behind.
#[cfg(any(test, feature = "test-faults"))]
pub(crate) static SKIP_APPLY: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

/// Fail the write at this index of the apply loop (tests only); `usize::MAX`
/// disables it. An index rather than a "after N landed" count, so a test can
/// fail the **first** write — the case where a later write on the same shard
/// would otherwise take a higher version than the one that failed.
#[cfg(test)]
pub(crate) static APPLY_FAIL_AT: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(usize::MAX);

/// Panic at this index of the apply loop (tests only); `usize::MAX` disables it.
/// Stands in for a panic anywhere inside a family's write path — the one exit
/// from apply that no `?` covers.
#[cfg(test)]
pub(crate) static APPLY_PANIC_AT: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(usize::MAX);

pub struct Tx<'d> {
    pub(crate) db: &'d crate::armour::Db,
    pub(crate) ops: Vec<Op>,
    pub(crate) durable: bool,
}

/// The builder's limit gate, in one place so every method enforces it
/// identically — and before any lock is taken.
pub(crate) fn check_op_limits(current_ops: usize, max_ops: usize, key_len: usize) -> DbResult<()> {
    if key_len > crate::txlog::record::MAX_KEY_LEN {
        return Err(DbError::Client("transaction key exceeds 64 KiB"));
    }
    if current_ops >= max_ops {
        return Err(DbError::Client("transaction exceeds max_ops"));
    }
    Ok(())
}

#[allow(private_bounds)]
impl<'d> Tx<'d> {
    pub fn put<S>(mut self, collection: &Arc<S>, key: &S::Key, value: &S::Val) -> DbResult<Self>
    where
        S: TxTarget + TxParticipant + 'static,
    {
        let mut bytes = Vec::new();
        collection.tx_encode(value, &mut bytes)?;
        let key_bytes = key.as_bytes().to_vec();
        check_op_limits(self.ops.len(), self.db.tx_config().max_ops, key_bytes.len())?;
        self.ops.push(Op {
            participant: collection.clone() as Arc<dyn TxParticipant>,
            key: key_bytes,
            action: Action::Put(bytes),
        });
        Ok(self)
    }

    pub fn delete<S>(mut self, collection: &Arc<S>, key: &S::Key) -> DbResult<Self>
    where
        S: TxTarget + TxParticipant + 'static,
    {
        let key_bytes = key.as_bytes().to_vec();
        check_op_limits(self.ops.len(), self.db.tx_config().max_ops, key_bytes.len())?;
        self.ops.push(Op {
            participant: collection.clone() as Arc<dyn TxParticipant>,
            key: key_bytes,
            action: Action::Delete,
        });
        Ok(self)
    }

    /// Decide the write from the stored value, under the transaction's locks.
    ///
    /// # The closure runs under shard locks
    ///
    /// `f` is evaluated at `commit()` while this thread holds every shard the
    /// transaction touches. It must not reach back into the database: `atomic()`
    /// and a nested `commit()` return [`DbError::ReentrantTx`] rather than
    /// deadlock, but a **facade** call (`put`, `get`, `delete`, `upsert` on a
    /// collection) is not instrumented and will park forever on a lock this
    /// thread already owns. Read what you need before building the transaction.
    pub fn upsert<S, U>(
        mut self,
        collection: &Arc<S>,
        key: &S::Key,
        f: impl FnOnce(Option<&S::Val>) -> U + 'static,
    ) -> DbResult<Self>
    where
        S: TxTarget + TxParticipant + 'static,
        U: IntoUpsert<S::Val> + 'static,
    {
        let key_bytes = key.as_bytes().to_vec();
        check_op_limits(self.ops.len(), self.db.tx_config().max_ops, key_bytes.len())?;
        let coll = collection.clone();
        let eval = move |stored: Option<&[u8]>| -> DbResult<Upsert<Vec<u8>>> {
            let decoded = match stored {
                Some(bytes) => Some(coll.tx_decode(bytes)?),
                None => None,
            };
            match f(decoded.as_ref()).into_parts().0 {
                Upsert::Put(value) => {
                    let mut out = Vec::new();
                    coll.tx_encode(&value, &mut out)?;
                    Ok(Upsert::Put(out))
                }
                Upsert::Keep => Ok(Upsert::Keep),
                Upsert::Delete => Ok(Upsert::Delete),
            }
        };
        self.ops.push(Op {
            participant: collection.clone() as Arc<dyn TxParticipant>,
            key: key_bytes,
            action: Action::Eval(Box::new(eval)),
        });
        Ok(self)
    }

    /// Fail the whole transaction unless the stored value satisfies `f`.
    ///
    /// Like [`upsert`](Self::upsert), `f` runs under the transaction's shard
    /// locks and must not reach back into the database — see that method's
    /// reentrancy note.
    pub fn guard<S>(
        mut self,
        collection: &Arc<S>,
        key: &S::Key,
        f: impl FnOnce(Option<&S::Val>) -> DbResult<()> + 'static,
    ) -> DbResult<Self>
    where
        S: TxTarget + TxParticipant + 'static,
    {
        let key_bytes = key.as_bytes().to_vec();
        check_op_limits(self.ops.len(), self.db.tx_config().max_ops, key_bytes.len())?;
        let coll = collection.clone();
        let check = move |stored: Option<&[u8]>| -> DbResult<()> {
            let decoded = match stored {
                Some(bytes) => Some(coll.tx_decode(bytes)?),
                None => None,
            };
            f(decoded.as_ref())
        };
        self.ops.push(Op {
            participant: collection.clone() as Arc<dyn TxParticipant>,
            key: key_bytes,
            action: Action::Guard(Box::new(check)),
        });
        Ok(self)
    }

    /// Wait for the log record to reach disk before the transaction is applied.
    /// Without it the transaction is atomic against a process crash but not
    /// against power loss.
    pub fn durable(mut self) -> Self {
        self.durable = true;
        self
    }

    pub fn commit(self) -> DbResult<()> {
        // Either direction of reentry deadlocks: an enclosing `atomic()` holds a
        // shard lock this commit would queue behind, and an enclosing commit
        // holds the locks its own closures run under.
        if ATOMIC_DEPTH.with(|d| d.get()) > 0 || crate::txn_depth::in_tx_commit() {
            return Err(DbError::ReentrantTx);
        }
        if self.ops.is_empty() {
            return Ok(());
        }

        // ── 1. Plan: group by participant, canonical (collection_id, shard) order
        let mut entries: Vec<Entry> = Vec::new();
        for op in self.ops.iter() {
            let id = self.db.tx_collection_id(op.participant.as_ref())?;
            if op.participant.encrypted() && !self.db.txlog().is_encrypted() {
                return Err(DbError::Config(
                    "an encrypted collection needs TxConfig::log_encryption_key: \
                     the transaction log would otherwise hold its values in the clear",
                ));
            }
            let sid = op.participant.shard_of(&op.key)?;
            match entries.iter_mut().find(|e| e.id == id) {
                Some(entry) => {
                    if !entry.shards.contains(&sid) {
                        entry.shards.push(sid);
                    }
                }
                None => entries.push(Entry {
                    id,
                    participant: op.participant.clone(),
                    shards: vec![sid],
                }),
            }
        }
        entries.sort_by_key(|e| e.id);
        for e in &mut entries {
            e.shards.sort_unstable();
        }

        // ── 2. Lock in canonical order
        //
        // From here to the last release this thread owns shard locks, and the
        // `upsert`/`guard` closures below run inside that window. The guard makes
        // any collection call from one of them return `ReentrantTx` instead of
        // parking forever on a mutex this thread already holds.
        let commit_guard = crate::txn_depth::TxCommitGuard::enter();
        let mut locked: Vec<Box<dyn TxLocked>> = Vec::with_capacity(entries.len());
        for entry in &entries {
            match entry.participant.lock_shards(&entry.shards) {
                Ok(l) => locked.push(l),
                Err(e) => {
                    for l in locked.into_iter().rev() {
                        l.abort();
                    }
                    return Err(e);
                }
            }
        }

        // ── 3. Evaluate under the locks, in insertion order, with
        //       read-your-writes over the pending write set. `self.ops` is
        //       consumed: `Eval`/`Guard` hold `FnOnce` closures.
        let mut writes: Vec<Write> = Vec::new();
        let mut eval_err: Option<DbError> = None;
        for op in self.ops {
            let Some(entry_pos) = entries
                .iter()
                .position(|e| Arc::ptr_eq(&e.participant, &op.participant))
            else {
                eval_err = Some(DbError::Internal("operation lost its participant"));
                break;
            };
            // Read-your-writes: an earlier operation on the same key wins over
            // the stored value.
            let current: Option<Vec<u8>> = match writes
                .iter()
                .rev()
                .find(|w| w.entry == entry_pos && w.key == op.key)
            {
                Some(w) => w.value.clone(),
                None => match locked[entry_pos].read(&op.key) {
                    Ok(v) => v,
                    Err(e) => {
                        eval_err = Some(e);
                        break;
                    }
                },
            };
            match op.action {
                Action::Put(bytes) => set_write(&mut writes, entry_pos, op.key, Some(bytes)),
                Action::Delete => set_write(&mut writes, entry_pos, op.key, None),
                Action::Eval(f) => match f(current.as_deref()) {
                    Ok(Upsert::Put(bytes)) => {
                        set_write(&mut writes, entry_pos, op.key, Some(bytes))
                    }
                    Ok(Upsert::Delete) => set_write(&mut writes, entry_pos, op.key, None),
                    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;
                    }
                }
            }
        }

        if let Some(e) = eval_err {
            for l in locked.into_iter().rev() {
                l.abort();
            }
            return Err(e);
        }
        if writes.is_empty() {
            // Every operation decided `Keep`: nothing to log, nothing to apply.
            for l in locked.into_iter().rev() {
                l.abort();
            }
            return Ok(());
        }

        // A write the participant's own write path cannot take must be refused
        // here, while steps 1–3 are still reversible. It cannot wait for `apply`:
        // by then the record is durable, and a deterministic failure would be
        // re-attempted by replay at every open — the database would stop opening.
        // Checked after evaluation because an `upsert` produces its bytes here.
        //
        // A delete is checked too, with an empty value: a tombstone is an entry,
        // so a long enough key makes one unappliable on its own. Reopening a
        // collection with a smaller `write_buffer_size` than its keys were
        // written under is how that happens in practice.
        if let Some(e) = writes.iter().find_map(|w| {
            let value_len = w.value.as_deref().map_or(0, <[u8]>::len);
            let cap = entries[w.entry].participant.max_entry_bytes()?;
            (crate::entry::entry_size(w.key.len(), value_len as u32) as usize > cap).then_some(
                DbError::Client("transaction entry exceeds write_buffer_size"),
            )
        }) {
            for l in locked.into_iter().rev() {
                l.abort();
            }
            return Err(e);
        }

        // ── 4–5. Reserve, encode, append — abort on any error before the log point
        let pre_append: DbResult<(u64, crate::txlog::Committed, Vec<VersionTag>)> = (|| {
            // Tags must align with `writes` (insertion order). Reserving
            // per-entry and pushing in entry order would zip the wrong GSN onto
            // an op when the caller wrote collections out of canonical id order.
            let mut tag_slots: Vec<Option<VersionTag>> = vec![None; writes.len()];
            for (pos, l) in locked.iter_mut().enumerate() {
                let indices: Vec<usize> = writes
                    .iter()
                    .enumerate()
                    .filter(|(_, w)| w.entry == pos)
                    .map(|(i, _)| i)
                    .collect();
                let planned: Vec<PlannedOp<'_>> = indices
                    .iter()
                    .map(|&i| PlannedOp {
                        key: &writes[i].key,
                        delete: writes[i].value.is_none(),
                    })
                    .collect();
                let reserved = l.reserve(&planned)?;
                if reserved.len() != indices.len() {
                    return Err(DbError::Internal("reserve returned too few tags"));
                }
                for (j, &wi) in indices.iter().enumerate() {
                    tag_slots[wi] = Some(reserved[j]);
                }
            }
            let tags: Vec<VersionTag> = tag_slots
                .into_iter()
                .map(|t| t.ok_or(DbError::Internal("missing reserved tag for a write")))
                .collect::<DbResult<_>>()?;

            let tx_id = self.db.txlog().next_tx_id()?;
            let record = build_record(tx_id, &entries, &writes, &tags);
            let mut encoded_len = Vec::new();
            crate::txlog::record::encode_record(&record, &mut encoded_len)?;
            if encoded_len.len() > self.db.tx_config().max_bytes {
                return Err(DbError::Client("transaction exceeds max_bytes"));
            }
            let committed = self.db.txlog().append(&record)?;
            Ok((tx_id, committed, tags))
        })();

        let (tx_id, committed, tags) = match pre_append {
            Ok(v) => v,
            Err(e) => {
                for l in locked.into_iter().rev() {
                    l.abort();
                }
                return Err(e);
            }
        };
        // The log point. From here every exit — `finish`, the `abort` below, or
        // a panic unwinding through the locked contexts — must leave the
        // reservations alone: the durable record names those slots and replay
        // owns them.
        for l in locked.iter_mut() {
            l.mark_logged();
        }
        // Past the log point the record's bookkeeping is owed unconditionally.
        // Every exit from here on — the suppressed-apply return below, an error
        // or a panic in `apply`, a defensive `finish()` failure, a user hook that
        // panics — must not leave `tx_id` pinning the checkpoint floor for the
        // rest of the process's life. Only a destructor covers a panic, and it
        // has to exist *before* the first thing that can panic, so it is created
        // here and told whether the apply succeeded afterwards.
        let mut settle = Settle {
            log: self.db.txlog(),
            tx_id,
            applied: false,
        };
        let mut sync_err = None;
        if self.durable {
            // A sync failure means the record may not be durable — but it is in
            // the log, so applying it is still correct and replay is still
            // possible. Surface the error after the locks are released.
            if let Err(e) = self.db.txlog().sync_upto(&committed) {
                tracing::error!(tx_id, "txlog sync failed after append: {e}");
                sync_err.get_or_insert(e);
            }
        }
        #[cfg(any(test, feature = "test-faults"))]
        if SKIP_APPLY.load(std::sync::atomic::Ordering::SeqCst) {
            // Reproduces a crash between the commit marker and apply: the
            // record is durable, the collections are untouched.
            for l in locked.into_iter().rev() {
                l.abort();
            }
            return Ok(());
        }
        #[cfg(feature = "test-faults")]
        failpoints::hit(failpoints::AFTER_LOG_APPEND);

        // ── 6. Apply in insertion order; remember which participant produced
        //       each event so hooks replay in that same order.
        let mut apply_err = None;
        let mut order: Vec<(usize, usize)> = Vec::new(); // (entry, local event index)
        let mut per_entry_events = vec![0usize; entries.len()];
        // Applied writes, which is *not* `order.len()`: a write may buffer no
        // hook event at all. The failpoints count writes, so they keep firing on
        // a `NoHook` collection.
        #[cfg(feature = "test-faults")]
        let mut applied_writes = 0usize;
        // Shards whose apply already failed, as `(entry, shard)`.
        //
        // Best-effort apply continues across *other* shards — the commit is
        // durable and replay finishes what is left — but never on this one. The
        // Bitcask replay rule is `reserved gsn > the shard's recovered
        // watermark`, so a later write landing on this shard would pull the
        // watermark above the version that failed: the next open would read the
        // failed operation as already on disk, skip it, settle the record and
        // retire it. Stopping the shard keeps every unapplied version above the
        // watermark, which is what replay needs to finish the record.
        let mut failed_shards: Vec<(usize, usize)> = Vec::new();
        // `_index` addresses the `APPLY_FAIL_AT` test hook and nothing else.
        #[cfg_attr(not(test), allow(clippy::unused_enumerate_index))]
        for (_index, (w, tag)) in writes.iter().zip(tags.iter()).enumerate() {
            let sid = match entries[w.entry].participant.shard_of(&w.key) {
                Ok(sid) => sid,
                Err(e) => {
                    apply_err.get_or_insert(e);
                    continue;
                }
            };
            #[cfg(test)]
            if _index == APPLY_FAIL_AT.load(std::sync::atomic::Ordering::SeqCst) {
                apply_err.get_or_insert(DbError::Internal("injected apply failure"));
                failed_shards.push((w.entry, sid));
                continue;
            }
            #[cfg(test)]
            assert!(
                _index != APPLY_PANIC_AT.load(std::sync::atomic::Ordering::SeqCst),
                "injected apply panic"
            );
            if failed_shards.contains(&(w.entry, sid)) {
                continue;
            }
            let outcome = match &w.value {
                Some(bytes) => locked[w.entry].apply_put(&w.key, bytes, *tag),
                None => locked[w.entry].apply_delete(&w.key, *tag),
            };
            // The count is the number of hook events the write buffered — zero
            // when it changed nothing (a delete of an absent key) or when the
            // family buffers none. Recording one per *write* instead would slide
            // every later hook onto the wrong event.
            let produced = match outcome {
                Ok(n) => n,
                Err(e) => {
                    apply_err.get_or_insert(e);
                    failed_shards.push((w.entry, sid));
                    continue;
                }
            };
            for _ in 0..produced {
                order.push((w.entry, per_entry_events[w.entry]));
                per_entry_events[w.entry] += 1;
            }
            #[cfg(feature = "test-faults")]
            {
                applied_writes += 1;
            }
            #[cfg(feature = "test-faults")]
            if applied_writes == 1 {
                failpoints::hit(failpoints::MID_APPLY);
            }
        }

        #[cfg(feature = "test-faults")]
        failpoints::hit(failpoints::AFTER_APPLY);

        settle.applied = apply_err.is_none();

        // ── 7. Release every lock, then replay hooks
        //
        // Every `finish()` runs even if an earlier one fails: it is what
        // releases that participant's shard guards through `ShardGuards`, and
        // dropping the `Box<dyn TxLocked>` instead would release a Map family's
        // guard pair in the inverted order (`ctx.rs`, the S1 regression).
        let mut deferred: Vec<Box<dyn DeferredReplay>> = Vec::with_capacity(locked.len());
        let mut finish_err = None;
        for l in locked {
            match l.finish() {
                Ok(d) => deferred.push(d),
                Err(e) => {
                    tracing::error!(tx_id, "releasing a transaction participant failed: {e}");
                    finish_err.get_or_insert(e);
                }
            }
        }
        // Every lock is now released; hooks are documented to run outside the
        // locked section, so they may touch collections again.
        drop(commit_guard);

        // Hook order is only meaningful when every participant handed back its
        // events; a gap would shift `entry` indices onto the wrong participant.
        if finish_err.is_none() {
            for (entry, local) in order {
                deferred[entry].replay_one(local);
            }
        }
        drop(deferred); // retirement

        drop(settle); // settle before the cut, so this record can retire now
        // The cut is log maintenance over *other* collections; a flush failing
        // there says nothing about this transaction, which is durable and
        // applied. Reporting it as the commit's result would have callers retry
        // a transaction that already happened. `Db::close` treats the same
        // failure the same way (`db.rs`).
        if let Err(e) = self.db.maybe_checkpoint() {
            tracing::error!(tx_id, "checkpoint after commit failed: {e}");
        }
        match apply_err.or(finish_err) {
            Some(e) => Err(e),
            None => sync_err.map_or(Ok(()), Err),
        }
    }
}

struct Entry {
    id: u32,
    participant: Arc<dyn TxParticipant>,
    shards: Vec<usize>,
}

/// Settles a logged record's bookkeeping on every exit from the tail of
/// `commit()`, including a panic.
///
/// `applied` gates checkpoint retirement; the in-flight set only pins the cut.
/// A partial apply must leave the record pending (`applied = false`) so replay
/// finishes it at the next open — but it must **not** keep pinning the floor,
/// or the log grows without bound for as long as the process runs.
struct Settle<'a> {
    log: &'a crate::txlog::TxLog,
    tx_id: u64,
    applied: bool,
}

impl Drop for Settle<'_> {
    fn drop(&mut self) {
        if self.applied {
            self.log.mark_applied(self.tx_id);
        } else {
            self.log.clear_in_flight(self.tx_id);
        }
    }
}

struct Write {
    entry: usize,
    key: Vec<u8>,
    /// `None` is a delete.
    value: Option<Vec<u8>>,
}

/// Collapse duplicate keys: the last decision wins, but the write keeps the
/// position of the key's first appearance, so apply order matches the order the
/// caller wrote the transaction in.
fn set_write(writes: &mut Vec<Write>, entry: usize, key: Vec<u8>, value: Option<Vec<u8>>) {
    match writes.iter_mut().find(|w| w.entry == entry && w.key == key) {
        Some(existing) => existing.value = value,
        None => writes.push(Write { entry, key, value }),
    }
}

fn build_record(tx_id: u64, entries: &[Entry], writes: &[Write], tags: &[VersionTag]) -> TxRecord {
    let ops = writes
        .iter()
        .zip(tags.iter())
        .map(|(w, tag)| {
            let entry = &entries[w.entry];
            TxOp {
                collection_id: entry.id,
                name: entry.participant.name().to_string(),
                schema_fingerprint: entry.participant.schema_fingerprint(),
                key: w.key.clone(),
                kind: if w.value.is_some() {
                    OpKind::Put
                } else {
                    OpKind::Delete
                },
                value: w.value.clone().unwrap_or_default(),
                version: *tag,
            }
        })
        .collect();
    TxRecord { tx_id, ops }
}

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

    #[test]
    fn tx_builder_rejects_an_oversized_key_before_any_lock() {
        // A `Key` is fixed-size, so an over-long key can only arrive through a
        // future variable-key family; the check is still the builder's job and
        // is tested directly, before anything can lock a shard.
        let err = check_op_limits(0, 8, crate::txlog::record::MAX_KEY_LEN + 1).unwrap_err();
        assert!(matches!(err, crate::DbError::Client(_)), "{err}");
    }

    #[test]
    fn tx_builder_rejects_more_ops_than_max_ops() {
        assert!(
            check_op_limits(7, 8, 8).is_ok(),
            "the 8th operation still fits"
        );
        let err = check_op_limits(8, 8, 8).unwrap_err();
        assert!(matches!(err, crate::DbError::Client(_)), "{err}");
    }
}

#[cfg(all(test, feature = "armour"))]
pub(super) mod commit_tests {
    use std::sync::Arc;

    use crate::armour::Db;
    use crate::{Config, NoHook, TtlHook, TxConfig, Upsert};

    // Two typed collections — the smallest pair that exercises two
    // participants, canonical ordering and hooks. Task 12 duplicates this
    // fixture for the integration test, which cannot see `#[cfg(test)]` items.
    use crate::{CollectionMeta, RapiraCodec, TypedMap};
    use armour_core::GetType;
    use rapira::Rapira;

    pub(super) type UserId = [u8; 8];
    pub(super) type CounterId = [u8; 8];

    #[derive(Clone, Copy, Debug, PartialEq, Eq, GetType, Rapira)]
    pub(super) struct User {
        pub active: bool,
    }
    impl CollectionMeta for User {
        type SelfId = UserId;
        const NAME: &'static str = "users";
    }

    #[derive(Clone, Copy, Debug, PartialEq, Eq, GetType, Rapira)]
    pub(super) struct Counter {
        pub n: u32,
    }
    impl CollectionMeta for Counter {
        type SelfId = CounterId;
        const NAME: &'static str = "counters";
    }

    pub(super) fn open_pair(
        db: &Arc<Db>,
    ) -> (
        Arc<TypedMap<UserId, User, RapiraCodec, TtlHook<User, NoHook>>>,
        Arc<TypedMap<CounterId, Counter, RapiraCodec, TtlHook<Counter, NoHook>>>,
    ) {
        let users = db
            .open_typed_map::<User, RapiraCodec, _>(Config::test(), NoHook, &[])
            .unwrap();
        let counters = db
            .open_typed_map::<Counter, RapiraCodec, _>(Config::test(), NoHook, &[])
            .unwrap();
        (users, counters)
    }

    /// The tests below name keys through this constructor — an 8-byte key.
    pub(super) fn user_id(n: u64) -> UserId {
        n.to_be_bytes()
    }

    /// A variable-length payload, so a test can exceed `write_buffer_size`.
    #[derive(Clone, Debug, PartialEq, Eq, GetType, Rapira)]
    pub(super) struct Blob {
        pub payload: Vec<u8>,
    }
    impl CollectionMeta for Blob {
        type SelfId = [u8; 8];
        const NAME: &'static str = "blobs";
    }

    /// One shard and a small write buffer: every key collides on the same shard,
    /// and an oversized value is reachable without a multi-megabyte allocation.
    pub(super) fn open_blobs(
        db: &Arc<Db>,
    ) -> Arc<TypedMap<[u8; 8], Blob, RapiraCodec, TtlHook<Blob, NoHook>>> {
        db.open_typed_map::<Blob, RapiraCodec, _>(
            Config::balanced()
                .shard_count(1)
                .write_buffer_size(8192)
                .hints(true)
                .build(),
            NoHook,
            &[],
        )
        .unwrap()
    }

    /// A failed apply must stop its shard, or the operation it failed on is lost
    /// for good.
    ///
    /// The Bitcask replay rule is `reserved gsn > the shard's recovered
    /// watermark`. Letting the next write on that shard land with `base + 1`
    /// pulls the watermark above the failed `base + 0`, so the next open reads
    /// the failed operation as "already on disk", skips it, settles the record
    /// and retires it — a durable, committed write silently gone.
    #[test]
    fn tx_apply_failure_stops_its_shard_so_replay_can_finish_the_record() {
        use std::sync::atomic::Ordering;

        let dir = tempfile::tempdir().unwrap();
        {
            let db = Arc::new(Db::open_test(dir.path()).unwrap());
            let blobs = open_blobs(&db);

            // One shard, so both writes share a GSN block; the *first* fails.
            crate::txn::APPLY_FAIL_AT.store(0, Ordering::SeqCst);
            let err = db
                .tx()
                .put(
                    &blobs,
                    &user_id(1),
                    &Blob {
                        payload: vec![7u8; 8],
                    },
                )
                .unwrap()
                .put(
                    &blobs,
                    &user_id(2),
                    &Blob {
                        payload: vec![1u8; 8],
                    },
                )
                .unwrap()
                .durable()
                .commit()
                .unwrap_err();
            crate::txn::APPLY_FAIL_AT.store(usize::MAX, Ordering::SeqCst);
            assert!(format!("{err}").contains("injected apply failure"), "{err}");

            assert!(
                blobs.get(&user_id(2)).is_none(),
                "a write on a shard whose apply failed must not take a higher GSN"
            );
            db.close().unwrap();
        }

        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let blobs = open_blobs(&db);
        assert!(
            blobs.get(&user_id(1)).is_some(),
            "the failed operation must still be replayable"
        );
        assert!(
            blobs.get(&user_id(2)).is_some(),
            "replay must finish the whole record"
        );
    }

    /// Stopping the shard inside the apply loop is not enough: once `commit()`
    /// releases the locks, an ordinary write on that shard lands with a GSN
    /// above the unapplied reservation, and after a restart that entry is the
    /// recovered watermark — so `needs_replay` skips the committed operation and
    /// it is lost exactly as before, just a little later.
    #[test]
    fn tx_apply_failure_fences_the_shard_against_later_writes() {
        use std::sync::atomic::Ordering;

        let dir = tempfile::tempdir().unwrap();
        {
            let db = Arc::new(Db::open_test(dir.path()).unwrap());
            let blobs = open_blobs(&db);

            crate::txn::APPLY_FAIL_AT.store(0, Ordering::SeqCst);
            let err = db
                .tx()
                .put(
                    &blobs,
                    &user_id(1),
                    &Blob {
                        payload: vec![7u8; 8],
                    },
                )
                .unwrap()
                .durable()
                .commit()
                .unwrap_err();
            crate::txn::APPLY_FAIL_AT.store(usize::MAX, Ordering::SeqCst);
            assert!(format!("{err}").contains("injected apply failure"), "{err}");

            // The locks are gone. A plain write on the same shard would take a
            // higher GSN and bury the gap replay depends on.
            let outcome = blobs.put(
                &user_id(2),
                Blob {
                    payload: vec![2u8; 8],
                },
            );
            assert!(
                matches!(outcome, Err(crate::DbError::ShardFenced { .. })),
                "a shard owing a committed write must refuse further writes"
            );
            db.close().unwrap();
        }

        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let blobs = open_blobs(&db);
        assert!(
            blobs.get(&user_id(1)).is_some(),
            "replay must still finish the committed operation"
        );
        assert!(
            blobs
                .put(
                    &user_id(3),
                    Blob {
                        payload: vec![3u8; 8]
                    }
                )
                .is_ok(),
            "the fence lasts only until the record is replayed at open"
        );
    }

    /// A tombstone is an entry too. Its size is the header plus the key, so a
    /// key long enough — a collection reopened with a smaller `write_buffer_size`
    /// is the realistic way in — makes a `delete` unappliable exactly like an
    /// oversized value, with the same consequence: the record is durable, and
    /// replay re-attempts it at every open.
    #[test]
    fn tx_rejects_an_oversized_tombstone_before_the_log_point() {
        #[derive(Clone, Copy, Debug, PartialEq, Eq, GetType, Rapira)]
        struct Marker {
            n: u32,
        }
        impl CollectionMeta for Marker {
            type SelfId = [u8; 9000];
            const NAME: &'static str = "markers";
        }

        fn open(
            db: &Arc<Db>,
            buffer: usize,
        ) -> Arc<TypedMap<[u8; 9000], Marker, RapiraCodec, TtlHook<Marker, NoHook>>> {
            db.open_typed_map::<Marker, RapiraCodec, _>(
                Config::balanced()
                    .shard_count(1)
                    .write_buffer_size(buffer)
                    .hints(true)
                    .build(),
                NoHook,
                &[],
            )
            .unwrap()
        }

        let dir = tempfile::tempdir().unwrap();
        let key = [7u8; 9000];
        {
            let db = Arc::new(Db::open_test(dir.path()).unwrap());
            let markers = open(&db, 32 * 1024);
            markers.put(&key, Marker { n: 1 }).unwrap();
            db.close().unwrap();
        }

        // Reopened with a buffer too small for this key's tombstone.
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let markers = open(&db, 8192);
        let err = db
            .tx()
            .delete(&markers, &key)
            .unwrap()
            .durable()
            .commit()
            .unwrap_err();

        assert!(
            matches!(err, crate::DbError::Client(_)),
            "an oversized tombstone is a client error: {err}"
        );
        assert!(
            db.pending_transactions().is_empty(),
            "no record may reach the log for a delete that cannot be applied"
        );
    }

    /// An entry the collection's write path cannot take must be refused while
    /// the transaction is still reversible. Past the log append the same failure
    /// is unrepairable: the record is durable, so replay would re-attempt the
    /// oversized write on every subsequent open and refuse to open the database.
    #[test]
    fn tx_rejects_an_oversized_entry_before_the_log_point() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let blobs = open_blobs(&db);

        let err = db
            .tx()
            .put(
                &blobs,
                &user_id(1),
                &Blob {
                    payload: vec![7u8; 32 * 1024],
                },
            )
            .unwrap()
            .put(
                &blobs,
                &user_id(2),
                &Blob {
                    payload: vec![1u8; 8],
                },
            )
            .unwrap()
            .durable()
            .commit()
            .unwrap_err();

        assert!(
            matches!(err, crate::DbError::Client(_)),
            "an oversized entry is a client error: {err}"
        );
        assert!(
            blobs.get(&user_id(2)).is_none(),
            "the transaction must roll back whole — no sibling write may land"
        );
        assert!(
            db.pending_transactions().is_empty(),
            "no record may reach the log for a transaction that cannot be applied"
        );
    }

    #[test]
    fn tx_db_assigns_stable_collection_ids_and_finds_participants() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, counters) = open_pair(&db);

        let id_users = db.tx_collection_id(users.as_ref()).unwrap();
        let id_counters = db.tx_collection_id(counters.as_ref()).unwrap();
        assert_ne!(id_users, id_counters);
        assert_eq!(
            db.tx_collection_id(users.as_ref()).unwrap(),
            id_users,
            "ids are stable"
        );
    }

    #[test]
    fn tx_db_config_limits_are_honoured() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(
            Db::open_with_tx_config(
                dir.path(),
                TxConfig {
                    max_ops: 2,
                    ..TxConfig::default()
                },
            )
            .unwrap(),
        );
        let (users, _counters) = open_pair(&db);
        let tx = db
            .tx()
            .put(&users, &user_id(1), &User { active: true })
            .unwrap()
            .put(&users, &user_id(2), &User { active: true })
            .unwrap();
        let err = tx
            .put(&users, &user_id(3), &User { active: true })
            .err()
            .expect("third put should exceed max_ops");
        assert!(matches!(err, crate::DbError::Client(_)), "{err}");
    }

    #[test]
    fn tx_commit_applies_both_collections_or_neither() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, counters) = open_pair(&db);

        db.tx()
            .put(&users, &user_id(1), &User { active: true })
            .unwrap()
            .upsert(&counters, &user_id(1), |old: Option<&Counter>| {
                Upsert::Put(Counter {
                    n: old.map_or(1, |c| c.n + 1),
                })
            })
            .unwrap()
            .commit()
            .unwrap();

        assert_eq!(users.get(&user_id(1)).map(|u| u.active), Some(true));
        assert_eq!(counters.get(&user_id(1)).map(|c| c.n), Some(1));
    }

    #[test]
    fn tx_commit_guard_error_leaves_nothing_behind() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, counters) = open_pair(&db);

        let err = db
            .tx()
            .put(&users, &user_id(2), &User { active: true })
            .unwrap()
            .guard(&counters, &user_id(2), |cur: Option<&Counter>| {
                if cur.is_none() {
                    Err(crate::DbError::TxConflict)
                } else {
                    Ok(())
                }
            })
            .unwrap()
            .commit()
            .unwrap_err();

        assert!(matches!(err, crate::DbError::TxConflict), "{err}");
        assert!(
            users.get(&user_id(2)).is_none(),
            "no write survives a failed guard"
        );
        assert_eq!(
            db.pending_transactions().len(),
            0,
            "nothing reached the log"
        );
    }

    #[test]
    fn tx_commit_collapses_duplicate_keys_and_reads_its_own_writes() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (_users, counters) = open_pair(&db);

        db.tx()
            .upsert(&counters, &user_id(3), |_old: Option<&Counter>| {
                Upsert::Put(Counter { n: 5 })
            })
            .unwrap()
            .upsert(&counters, &user_id(3), |old: Option<&Counter>| {
                // Must observe the value the previous operation produced.
                Upsert::Put(Counter {
                    n: old.map_or(0, |c| c.n) * 2,
                })
            })
            .unwrap()
            .commit()
            .unwrap();

        assert_eq!(counters.get(&user_id(3)).map(|c| c.n), Some(10));
    }

    #[test]
    fn tx_commit_rejects_a_collection_from_another_db() {
        let dir_a = tempfile::tempdir().unwrap();
        let dir_b = tempfile::tempdir().unwrap();
        let db_a = Arc::new(Db::open_test(dir_a.path()).unwrap());
        let db_b = Arc::new(Db::open_test(dir_b.path()).unwrap());
        let (users_a, _) = open_pair(&db_a);
        let (_, counters_b) = open_pair(&db_b);

        let err = db_a
            .tx()
            .put(&users_a, &user_id(4), &User { active: true })
            .unwrap()
            .put(&counters_b, &user_id(4), &Counter { n: 1 })
            .unwrap()
            .commit()
            .unwrap_err();
        assert!(matches!(err, crate::DbError::Client(_)), "{err}");
    }

    #[test]
    fn tx_commit_durable_writes_the_log_before_applying() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, _counters) = open_pair(&db);

        db.tx()
            .put(&users, &user_id(5), &User { active: true })
            .unwrap()
            .durable()
            .commit()
            .unwrap();

        // The record is applied, so it is retired at the next checkpoint but
        // still present until then.
        assert!(dir.path().join("txlog").is_dir());
        assert_eq!(users.get(&user_id(5)).map(|u| u.active), Some(true));
    }

    pub(super) fn open_counters(
        db: &Arc<Db>,
    ) -> Arc<TypedMap<CounterId, Counter, RapiraCodec, TtlHook<Counter, NoHook>>> {
        db.open_typed_map::<Counter, RapiraCodec, _>(Config::test(), NoHook, &[])
            .unwrap()
    }

    #[test]
    fn tx_checkpoint_retires_applied_records_after_a_durable_flush() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(
            Db::open_with_tx_config(
                dir.path(),
                TxConfig {
                    segment_bytes: 512,
                    ..TxConfig::default()
                },
            )
            .unwrap(),
        );
        let (users, _counters) = open_pair(&db);

        for i in 0..20u64 {
            db.tx()
                .put(&users, &user_id(i), &User { active: true })
                .unwrap()
                .commit()
                .unwrap();
        }
        assert!(!db.pending_transactions().is_empty());

        db.checkpoint_now().unwrap();
        assert!(
            db.pending_transactions().is_empty(),
            "applied records retire once their participants are durable"
        );
    }

    #[test]
    fn tx_checkpoint_keeps_a_record_whose_participant_was_never_opened() {
        use std::sync::atomic::Ordering;

        let dir = tempfile::tempdir().unwrap();
        {
            let db = Arc::new(Db::open_test(dir.path()).unwrap());
            let (users, counters) = open_pair(&db);
            db.tx()
                .put(&users, &user_id(1), &User { active: true })
                .unwrap()
                .put(&counters, &user_id(1), &Counter { n: 1 })
                .unwrap()
                .durable()
                .commit()
                .unwrap();
            crate::txn::SKIP_APPLY.store(true, Ordering::SeqCst);
            db.tx()
                .put(&users, &user_id(2), &User { active: true })
                .unwrap()
                .durable()
                .commit()
                .unwrap();
            crate::txn::SKIP_APPLY.store(false, Ordering::SeqCst);
        }

        // Reopen and open ONLY the counters collection: the record touching
        // `users` cannot be replayed, so it must survive the checkpoint.
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let _counters = open_counters(&db);
        db.checkpoint_now().unwrap();
        assert_eq!(db.pending_transactions().len(), 1);
    }

    #[test]
    fn tx_partial_apply_error_keeps_record_in_flight() {
        use std::sync::atomic::Ordering;

        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, counters) = open_pair(&db);

        crate::txn::APPLY_FAIL_AT.store(1, Ordering::SeqCst);
        let err = db
            .tx()
            .put(&users, &user_id(20), &User { active: true })
            .unwrap()
            .put(&counters, &user_id(20), &Counter { n: 2 })
            .unwrap()
            .durable()
            .commit()
            .unwrap_err();
        crate::txn::APPLY_FAIL_AT.store(usize::MAX, Ordering::SeqCst);
        assert!(format!("{err}").contains("injected apply failure"), "{err}");
        assert_eq!(
            users.get(&user_id(20)).map(|u| u.active),
            Some(true),
            "the first apply must land"
        );
        assert!(
            counters.get(&user_id(20)).is_none(),
            "the second apply must not"
        );
        assert_eq!(
            db.pending_transactions().len(),
            1,
            "partial apply must not mark_applied — the record stays pending"
        );
        db.checkpoint_now().unwrap();
        assert_eq!(
            db.pending_transactions().len(),
            1,
            "checkpoint must not retire a record whose apply never fully succeeded"
        );

        // Opening any collection runs the open-time settle pass. That pass is a
        // statement about a *previous* process's replay — `tx_replay_done`
        // accumulates for the whole session, so by now it already names every
        // participant of the record above. It must not claim a record this
        // session left half-applied, or the half that never landed is lost.
        #[derive(Clone, Copy, Debug, PartialEq, Eq, GetType, Rapira)]
        struct Third {
            v: u32,
        }
        impl CollectionMeta for Third {
            type SelfId = [u8; 8];
            const NAME: &'static str = "thirds";
        }
        let _third = db
            .open_typed_map::<Third, RapiraCodec, _>(Config::test(), NoHook, &[])
            .unwrap();
        db.checkpoint_now().unwrap();
        assert_eq!(
            db.pending_transactions().len(),
            1,
            "opening an unrelated collection must not settle a half-applied record"
        );
    }

    #[test]
    fn tx_partial_apply_clears_in_flight_so_later_commits_can_checkpoint() {
        use std::sync::atomic::Ordering;

        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(
            Db::open_with_tx_config(
                dir.path(),
                TxConfig {
                    segment_bytes: 512,
                    ..TxConfig::default()
                },
            )
            .unwrap(),
        );
        let (users, counters) = open_pair(&db);

        crate::txn::APPLY_FAIL_AT.store(1, Ordering::SeqCst);
        let _ = db
            .tx()
            .put(&users, &user_id(30), &User { active: true })
            .unwrap()
            .put(&counters, &user_id(30), &Counter { n: 3 })
            .unwrap()
            .commit()
            .unwrap_err();
        crate::txn::APPLY_FAIL_AT.store(usize::MAX, Ordering::SeqCst);
        assert_eq!(db.pending_transactions().len(), 1);

        for i in 0..12u64 {
            db.tx()
                .put(&users, &user_id(100 + i), &User { active: true })
                .unwrap()
                .commit()
                .unwrap();
        }
        db.checkpoint_now().unwrap();
        assert_eq!(
            db.pending_transactions().len(),
            1,
            "cleared in_flight must not freeze the checkpoint cut"
        );
    }

    #[test]
    fn tx_tags_follow_write_order_not_collection_id_order() {
        // users is published before counters, so collection_id(users) <
        // collection_id(counters). Advance users' GSN watermark first; then a
        // counters-then-users write with SKIP_APPLY must still carry each
        // collection's own reservation. A zip bug attaches users' high GSN to
        // the counters op and counters' low GSN to users — reopen then skips
        // the users op (low GSN ≤ users watermark).
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, counters) = open_pair(&db);
        for i in 0..8u64 {
            db.tx()
                .put(&users, &user_id(50 + i), &User { active: false })
                .unwrap()
                .commit()
                .unwrap();
        }

        crate::txn::SKIP_APPLY.store(true, std::sync::atomic::Ordering::SeqCst);
        db.tx()
            .put(&counters, &user_id(40), &Counter { n: 7 })
            .unwrap()
            .put(&users, &user_id(40), &User { active: true })
            .unwrap()
            .durable()
            .commit()
            .unwrap();
        crate::txn::SKIP_APPLY.store(false, std::sync::atomic::Ordering::SeqCst);

        let pending = db.txlog().pending();
        let ops = &pending.last().unwrap().ops;
        assert_eq!(ops[0].name, "counters");
        assert_eq!(ops[1].name, "users");

        drop(users);
        drop(counters);
        drop(db);
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, counters) = open_pair(&db);
        assert_eq!(counters.get(&user_id(40)).map(|c| c.n), Some(7));
        assert_eq!(users.get(&user_id(40)).map(|u| u.active), Some(true));
    }

    #[test]
    fn tx_commit_durable_surfaces_sync_failure_after_unlock() {
        use std::sync::atomic::Ordering;

        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, _counters) = open_pair(&db);

        crate::txlog::segment::FAIL_SYNC.store(true, Ordering::SeqCst);
        let err = db
            .tx()
            .put(&users, &user_id(21), &User { active: true })
            .unwrap()
            .durable()
            .commit()
            .unwrap_err();
        crate::txlog::segment::FAIL_SYNC.store(false, Ordering::SeqCst);
        assert!(
            format!("{err}").contains("injected txlog sync failure"),
            "{err}"
        );
        assert_eq!(
            users.get(&user_id(21)).map(|u| u.active),
            Some(true),
            "apply still runs after a sync failure"
        );
    }

    #[test]
    fn tx_reentrant_commit_inside_atomic_errors_instead_of_hanging() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, counters) = open_pair(&db);

        let err = users
            .atomic(&user_id(1), |_shard| {
                db.tx()
                    .put(&counters, &user_id(1), &Counter { n: 1 })
                    .unwrap()
                    .commit()
            })
            .unwrap_err();
        assert!(matches!(err, crate::DbError::ReentrantTx), "{err}");
    }

    /// The other direction: `upsert`/`guard` closures run *under* the
    /// transaction's shard locks, so reaching back into a collection from one
    /// parks on a mutex this thread already owns — forever, with no poisoning
    /// and no timeout to break it.
    #[test]
    fn tx_collection_call_inside_an_upsert_closure_errors_instead_of_hanging() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, counters) = open_pair(&db);

        // Same collection, same key — so this is the transaction's *own* shard
        // lock, the case that hangs rather than merely being out of order.
        let reentrant = users.clone();
        let err = db
            .tx()
            .guard(&users, &user_id(1), move |_current: Option<&User>| {
                reentrant.atomic(&user_id(1), |_shard| Ok(()))
            })
            .unwrap()
            .commit()
            .unwrap_err();
        assert!(matches!(err, crate::DbError::ReentrantTx), "{err}");

        // A nested commit is rejected the same way.
        let nested_db = db.clone();
        let nested_counters = counters.clone();
        let err = db
            .tx()
            .guard(&users, &user_id(2), move |_current: Option<&User>| {
                nested_db
                    .tx()
                    .put(&nested_counters, &user_id(2), &Counter { n: 1 })?
                    .commit()
            })
            .unwrap()
            .commit()
            .unwrap_err();
        assert!(matches!(err, crate::DbError::ReentrantTx), "{err}");
    }

    /// A panic inside a `guard`/`upsert` closure unwinds out of `commit()` with
    /// the shard locks still held. They are released by `TxHeld`'s destructor —
    /// the same `ShardGuards` path `finish`/`abort` take, not a plain drop of
    /// the guard tuple (whose order is the S1 regression; `ctx.rs` owns that
    /// property's own tests). What this asserts is the liveness half: the
    /// database is still usable afterwards rather than wedged.
    #[test]
    fn tx_panic_under_the_locks_leaves_the_collection_usable() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, _counters) = open_pair(&db);
        users.put(&user_id(1), User { active: true }).unwrap();

        let previous = std::panic::take_hook();
        std::panic::set_hook(Box::new(|_| {})); // the panic below is expected
        let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            db.tx()
                .guard(&users, &user_id(1), |_current: Option<&User>| {
                    panic!("closure blew up under the locks")
                })
                .unwrap()
                .commit()
        }));
        std::panic::set_hook(previous);
        assert!(outcome.is_err(), "the closure must have panicked");

        // Same shard, straight after: a leaked guard would hang here forever.
        users.put(&user_id(1), User { active: false }).unwrap();
        assert_eq!(users.get(&user_id(1)).map(|u| u.active), Some(false));
        users.atomic(&user_id(1), |_shard| Ok(())).unwrap();

        // And nothing reached the log: the panic happened before the append.
        assert!(db.pending_transactions().is_empty());
    }

    // ── Writes that change nothing ─────────────────────────────────────────
    //
    // A delete of an absent key appends no entry in any family. That makes it
    // the one operation whose reserved GSN is never consumed and whose hook
    // event is never buffered — the two assumptions the commit path used to
    // make about every applied write.

    #[test]
    fn tx_delete_of_a_missing_key_does_not_strand_the_reserved_gsn() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, _counters) = open_pair(&db);

        // Nothing to delete: no entry is appended, so the armed GSN is unused.
        db.tx()
            .delete(&users, &user_id(1))
            .unwrap()
            .commit()
            .unwrap();

        // A stranded GSN trips `arm_gsn`'s single-shot debug assert here, and in
        // release builds silently lands on this write with a GSN below the ones
        // already on the shard.
        db.tx()
            .put(&users, &user_id(1), &User { active: true })
            .unwrap()
            .commit()
            .unwrap();
        assert_eq!(users.get(&user_id(1)).map(|u| u.active), Some(true));

        // Same thing without a transaction in between: a plain facade write must
        // not inherit a transaction's GSN either.
        db.tx()
            .delete(&users, &user_id(2))
            .unwrap()
            .commit()
            .unwrap();
        users.put(&user_id(3), User { active: false }).unwrap();
        assert_eq!(users.get(&user_id(3)).map(|u| u.active), Some(false));
    }

    #[test]
    fn tx_delete_miss_next_to_a_write_keeps_hooks_and_checkpoint_alive() {
        use std::sync::atomic::AtomicUsize;

        #[derive(
            Clone,
            Copy,
            Debug,
            PartialEq,
            armour_core::GetType,
            zerocopy::FromBytes,
            zerocopy::IntoBytes,
            zerocopy::Immutable,
        )]
        #[repr(C)]
        struct Row {
            n: u32,
        }
        impl CollectionMeta for Row {
            type SelfId = [u8; 8];
            const NAME: &'static str = "rows";
        }

        /// `NEEDS_WRITE` is what makes the family buffer events at all — the
        /// shape every hook-backed secondary index has.
        #[derive(Clone, Default)]
        struct CountingHook(Arc<AtomicUsize>);
        impl crate::TypedWriteHook<[u8; 8], Row> for CountingHook {
            const NEEDS_WRITE: bool = true;
            fn on_write(&self, _k: &[u8; 8], _old: Option<&Row>, _new: Option<&Row>) {
                self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            }
        }

        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let fired = Arc::new(AtomicUsize::new(0));
        let rows = db
            .open_zero_map::<Row, 4, _>(Config::test(), CountingHook(fired.clone()), &[])
            .unwrap();

        // One write that changes something and one that does not, in a single
        // transaction. The event count (1) is below the applied-write count (2).
        db.tx()
            .delete(&rows, &[9u8; 8])
            .unwrap()
            .put(&rows, &[1u8; 8], &Row { n: 5 })
            .unwrap()
            .commit()
            .expect("a delete of an absent key must not fail the commit");

        assert_eq!(rows.get(&[1u8; 8]).map(|r| r.n), Some(5));
        assert_eq!(
            fired.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "the write that changed something must still fire its hook"
        );

        // The record must be marked applied, so the cut can retire it. A commit
        // that errored out before `mark_applied` would pin the floor forever.
        db.checkpoint_now().unwrap();
        assert!(
            db.pending_transactions().is_empty(),
            "the transaction log must not be frozen by a write that changed nothing"
        );
    }

    /// A panic inside `apply` unwinds out of `commit()` between the log append
    /// and the point where the record's bookkeeping is normally settled. Nothing
    /// on that path — releasing the shard locks, `TxHeld::drop` — clears the
    /// in-flight pin, so `in_flight_floor` would stay frozen at this `tx_id` and
    /// the checkpoint could never retire this record or any later one, for the
    /// rest of the process's life.
    #[test]
    fn tx_panic_during_apply_still_settles_the_record() {
        use std::sync::atomic::Ordering;

        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(
            Db::open_with_tx_config(
                dir.path(),
                TxConfig {
                    segment_bytes: 512,
                    ..TxConfig::default()
                },
            )
            .unwrap(),
        );
        let blobs = open_blobs(&db);
        // The later commits go to a different collection: the panicked shard is
        // fenced afterwards, since it owes a committed write.
        let (users, _counters) = open_pair(&db);

        let previous = std::panic::take_hook();
        std::panic::set_hook(Box::new(|_| {})); // the panic below is expected
        crate::txn::APPLY_PANIC_AT.store(0, Ordering::SeqCst);
        let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            db.tx()
                .put(
                    &blobs,
                    &user_id(1),
                    &Blob {
                        payload: vec![5u8; 8],
                    },
                )
                .unwrap()
                .commit()
        }));
        crate::txn::APPLY_PANIC_AT.store(usize::MAX, Ordering::SeqCst);
        std::panic::set_hook(previous);
        assert!(outcome.is_err(), "apply must have panicked");

        // The panicked record stays pending — it was never fully applied — but it
        // must not pin the cut for everything committed after it.
        for i in 0..12u64 {
            db.tx()
                .put(&users, &user_id(100 + i), &User { active: true })
                .unwrap()
                .commit()
                .unwrap();
        }
        db.checkpoint_now().unwrap();
        assert_eq!(
            db.pending_transactions().len(),
            1,
            "a panic in apply must not freeze the checkpoint floor"
        );
    }

    /// Hooks run user code after the locks are released. A panic there unwinds
    /// straight out of `commit()`, so the record's bookkeeping has to be owed by
    /// a destructor — otherwise `tx_id` keeps pinning `in_flight_floor` and the
    /// checkpoint can never retire anything again.
    #[test]
    fn tx_panicking_hook_still_settles_the_record() {
        #[derive(
            Clone,
            Copy,
            Debug,
            PartialEq,
            armour_core::GetType,
            zerocopy::FromBytes,
            zerocopy::IntoBytes,
            zerocopy::Immutable,
        )]
        #[repr(C)]
        struct Boom {
            n: u32,
        }
        impl CollectionMeta for Boom {
            type SelfId = [u8; 8];
            const NAME: &'static str = "boom";
        }

        struct PanicHook;
        impl crate::TypedWriteHook<[u8; 8], Boom> for PanicHook {
            const NEEDS_WRITE: bool = true;
            fn on_write(&self, _k: &[u8; 8], _old: Option<&Boom>, _new: Option<&Boom>) {
                panic!("hook blew up");
            }
        }

        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let rows = db
            .open_zero_map::<Boom, 4, _>(Config::test(), PanicHook, &[])
            .unwrap();

        let previous = std::panic::take_hook();
        std::panic::set_hook(Box::new(|_| {})); // the panic below is expected
        let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            db.tx()
                .put(&rows, &[1u8; 8], &Boom { n: 1 })
                .unwrap()
                .commit()
        }));
        std::panic::set_hook(previous);
        assert!(outcome.is_err(), "the hook must have panicked");

        // The write itself landed before the hook ran, so the record is applied
        // and the cut must be free to retire it.
        assert_eq!(rows.get(&[1u8; 8]).map(|r| r.n), Some(1));
        db.checkpoint_now().unwrap();
        assert!(
            db.pending_transactions().is_empty(),
            "a panicking hook must not freeze the checkpoint floor"
        );
    }
}