armdb 0.12.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::{ClaimKind, DeferredReplay, PlannedOp, TxLocked, TxParticipant};
use crate::txn_depth::ATOMIC_DEPTH;
use crate::unique::{
    ClaimTarget, ExplicitClaim, UniqueIndex, UniqueNamespace, UniqueOwner, collection_tag,
    reservation_ttl_ms,
};
use crate::upsert::{IntoUpsert, Upsert};
use crate::watch::WatchWakeBatch;

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

pub use scope::{Selection, TxScope};

/// Fault injection for tests. Two kinds live here:
///
/// * **crash stages** (`STAGE`) — the child process sets one and `_exit`s when
///   the commit path reaches it, so the parent can reopen and assert;
/// * **error injection** (`APPLY_FAIL_AT`) — the commit fails in-process, so a
///   consumer can test what it does with [`crate::DbError::TxFailedAfterLog`].
///
/// **Reaching this module from outside the crate needs `feature = "armour"`
/// *and* `feature = "test-faults"`**: `pub mod txn` is declared under both
/// (`lib.rs`). `armour` is on by default, so a default build only needs
/// `--features test-faults`; a `--no-default-features` target must name both.
#[cfg(any(test, feature = "test-faults"))]
pub mod failpoints {
    use std::sync::atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering};

    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: AtomicU8 = AtomicU8::new(NONE);

    /// Fail the write at this index of the transaction's **write plan**;
    /// `usize::MAX` disables it.
    ///
    /// The plan is not the list of builder calls. `commit` turns the builder's
    /// operations into a write plan first, and three things move the indices:
    ///
    /// * repeated writes to the same `(collection, key)` **collapse** into one
    ///   plan entry, last value winning — two `put`s to one key are index 0
    ///   alone, not 0 and 1;
    /// * [`Tx::guard`](crate::Tx::guard), and a [`Tx::upsert`](crate::Tx::upsert)
    ///   closure returning [`Upsert::Keep`](crate::Upsert::Keep), write nothing
    ///   and so occupy no index at all;
    /// * secondary-index rows projected from those writes are **appended** after
    ///   them, followed by explicit writes into projection targets.
    ///
    /// The index therefore equals the builder-call index only for a transaction
    /// of distinct-key `put`/`delete` calls on collections with no projections.
    /// That is the usual shape of a test, which is why the simple case behaves
    /// as you would expect — but do not rely on it past that shape.
    ///
    /// **Index 0 needs no such reasoning:** it is always the first planned
    /// write, whatever the plan looks like, so it always leaves that write's
    /// shard fenced. That is the case worth testing.
    ///
    /// A failure puts its `(participant, shard)` out of service for the rest of
    /// the apply loop — later writes on that shard are skipped, writes on other
    /// shards still apply.
    ///
    /// Prefer [`ApplyFailGuard`] over storing directly.
    pub static APPLY_FAIL_AT: AtomicUsize = AtomicUsize::new(usize::MAX);

    /// Set while an [`ApplyFailGuard`] is alive. See [`Claim`].
    static APPLY_FAIL_HELD: AtomicBool = AtomicBool::new(false);

    /// Set while a [`LogSyncFailGuard`] is alive. See [`Claim`].
    static LOG_SYNC_FAIL_HELD: AtomicBool = AtomicBool::new(false);

    /// Exclusive ownership of one failpoint, released on drop.
    ///
    /// A failpoint is a single process-global value, so two live guards cannot
    /// both mean something: the second would overwrite the first's setting, and
    /// whichever dropped first would disarm the failpoint under the other. Both
    /// outcomes are silent, and a test that silently stops injecting is a test
    /// that passes for the wrong reason. Claiming the slot turns that into a
    /// panic at the second guard's construction instead.
    struct Claim(&'static AtomicBool);

    impl Claim {
        fn take(held: &'static AtomicBool, name: &str) -> Self {
            assert!(
                held.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
                    .is_ok(),
                "{name} is already armed by a live guard. Failpoints are \
                 process-global, so only one guard at a time is meaningful — run \
                 failpoint tests under `cargo nextest`, which gives each test its \
                 own process."
            );
            Self(held)
        }
    }

    impl Drop for Claim {
        fn drop(&mut self) {
            self.0.store(false, Ordering::SeqCst);
        }
    }

    /// Arms [`APPLY_FAIL_AT`] for as long as it is alive, and disarms it on drop
    /// — including when a failed assertion unwinds past it, which is the whole
    /// reason to prefer it over storing to the static by hand.
    ///
    /// **One at a time.** Constructing a second guard while one is alive panics.
    /// Allowing it would be worse than refusing: the second would overwrite the
    /// first's index, and whichever dropped first would disarm the failpoint under
    /// the other — silently, leaving a test that no longer injects anything and
    /// passes for the wrong reason.
    ///
    /// The guard scopes the failpoint to a *process*, not to the arming thread:
    /// under `cargo test`, which runs a binary's tests as threads, two overlapping
    /// failpoint tests now panic instead of interfering invisibly.
    #[must_use = "the failpoint disarms as soon as the guard is dropped"]
    pub struct ApplyFailGuard(Claim);

    impl ApplyFailGuard {
        /// Panics if another `ApplyFailGuard` is still alive.
        pub fn at(index: usize) -> Self {
            let claim = Claim::take(&APPLY_FAIL_HELD, "APPLY_FAIL_AT");
            APPLY_FAIL_AT.store(index, Ordering::SeqCst);
            Self(claim)
        }
    }

    impl Drop for ApplyFailGuard {
        fn drop(&mut self) {
            // The claim in `self.0` is released after this body returns, so the
            // failpoint is never free-to-claim while still armed.
            APPLY_FAIL_AT.store(usize::MAX, Ordering::SeqCst);
        }
    }

    /// Make the transaction log's fsync fail after a successful append.
    ///
    /// This substitutes the **call**, not the filesystem: it does not prove how
    /// a real `fsync` error behaves, but it makes the whole observable part of
    /// the [`crate::TxFailStage::Sync`] branch testable — the stage, the empty
    /// `fenced` list, the preserved source, and the fact that apply continues.
    pub static FAIL_LOG_SYNC: AtomicBool = AtomicBool::new(false);

    /// Arms [`FAIL_LOG_SYNC`] for as long as it is alive, and disarms it on drop,
    /// including on unwind.
    ///
    /// **One at a time**, on the same terms as [`ApplyFailGuard`]: a second live
    /// guard panics rather than silently taking the failpoint from the first. The
    /// two guards own *different* failpoints, so holding one of each is fine.
    #[must_use = "the failpoint disarms as soon as the guard is dropped"]
    pub struct LogSyncFailGuard(Claim);

    impl LogSyncFailGuard {
        /// Panics if another `LogSyncFailGuard` is still alive.
        pub fn arm() -> Self {
            let claim = Claim::take(&LOG_SYNC_FAIL_HELD, "FAIL_LOG_SYNC");
            FAIL_LOG_SYNC.store(true, Ordering::SeqCst);
            Self(claim)
        }
    }

    impl Drop for LogSyncFailGuard {
        fn drop(&mut self) {
            // Released before the claim in `self.0`, as in `ApplyFailGuard`.
            FAIL_LOG_SYNC.store(false, Ordering::SeqCst);
        }
    }

    #[cfg(feature = "test-faults")]
    pub(crate) fn hit(stage: u8) {
        if STAGE.load(Ordering::SeqCst) == stage {
            // No destructors, no flush, no hint file — a real crash.
            unsafe { libc::_exit(9) }
        }
    }

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

        #[test]
        fn apply_fail_guard_disarms_on_unwind() {
            assert_eq!(APPLY_FAIL_AT.load(Ordering::SeqCst), usize::MAX);
            let caught = std::panic::catch_unwind(|| {
                let _g = ApplyFailGuard::at(0);
                assert_eq!(APPLY_FAIL_AT.load(Ordering::SeqCst), 0);
                panic!("boom");
            });
            assert!(caught.is_err());
            assert_eq!(
                APPLY_FAIL_AT.load(Ordering::SeqCst),
                usize::MAX,
                "a failed assertion must not poison the next test"
            );
        }

        /// Run `f`, keeping a deliberate panic out of the test log.
        fn quietly<R>(f: impl FnOnce() -> R) -> std::thread::Result<R> {
            let previous = std::panic::take_hook();
            std::panic::set_hook(Box::new(|_| {}));
            let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));
            std::panic::set_hook(previous);
            outcome
        }

        /// A second guard must not quietly take the failpoint from the first —
        /// that is how a test stops injecting and passes for the wrong reason.
        #[test]
        fn a_second_live_guard_panics_instead_of_stealing_the_failpoint() {
            let first = ApplyFailGuard::at(0);
            assert!(
                quietly(|| ApplyFailGuard::at(1)).is_err(),
                "a second live ApplyFailGuard must panic"
            );
            assert_eq!(
                APPLY_FAIL_AT.load(Ordering::SeqCst),
                0,
                "the refused guard must not have moved the first one's index"
            );

            // The two guards own different failpoints, so this one is unaffected.
            let sync = LogSyncFailGuard::arm();
            assert!(
                quietly(LogSyncFailGuard::arm).is_err(),
                "a second live LogSyncFailGuard must panic"
            );
            assert!(FAIL_LOG_SYNC.load(Ordering::SeqCst));

            // And the slot is claimable again once the first guard is gone.
            drop(first);
            drop(sync);
            let again = ApplyFailGuard::at(2);
            assert_eq!(APPLY_FAIL_AT.load(Ordering::SeqCst), 2);
            drop(again);
            assert_eq!(APPLY_FAIL_AT.load(Ordering::SeqCst), usize::MAX);
            assert!(!FAIL_LOG_SYNC.load(Ordering::SeqCst));
        }
    }
}

/// 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. The same flag is the explicit escape hatch for a codec-only
    /// format migration whose pending records carry only an ambiguous type hash;
    /// those records are checkpointed before any shard rewrite. 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<()>>;

/// Arc-owned services used by queue-owned transaction builders.
pub(crate) struct TxRuntime {
    pub(crate) config: TxConfig,
    pub(crate) log: Arc<crate::txlog::TxLog>,
    pub(crate) collections: Arc<crate::sync::Mutex<Vec<Arc<dyn crate::armour::Collection>>>>,
    pub(crate) ranks: Arc<crate::sync::Mutex<std::collections::HashMap<&'static str, u8>>>,
    pub(crate) post_log_queue_producer_permits:
        crate::sync::Mutex<Vec<crate::queue::retention::ProducerPermit>>,
    pub(crate) _root_lease: Arc<std::fs::File>,
}

pub(crate) enum TxContext<'d> {
    Borrowed(&'d crate::armour::Db),
    Runtime(Arc<TxRuntime>),
}

impl TxContext<'_> {
    fn pin_post_log_queue_producer_permits(
        &self,
        permits: Vec<crate::queue::retention::ProducerPermit>,
    ) {
        let runtime = match self {
            Self::Borrowed(db) => &db.tx_runtime,
            Self::Runtime(runtime) => runtime,
        };
        crate::sync::lock(&runtime.post_log_queue_producer_permits).extend(permits);
    }
    pub(crate) fn config(&self) -> &TxConfig {
        match self {
            Self::Borrowed(db) => db.tx_config(),
            Self::Runtime(runtime) => &runtime.config,
        }
    }

    pub(crate) fn log(&self) -> &crate::txlog::TxLog {
        match self {
            Self::Borrowed(db) => db.txlog(),
            Self::Runtime(runtime) => &runtime.log,
        }
    }

    pub(crate) fn collection_id(&self, participant: &dyn TxParticipant) -> DbResult<u32> {
        match self {
            Self::Borrowed(db) => db.tx_collection_id(participant),
            Self::Runtime(runtime) => {
                let needle = participant as *const dyn TxParticipant as *const ();
                crate::sync::lock(&runtime.collections)
                    .iter()
                    .position(|collection| Arc::as_ptr(collection) as *const () == needle)
                    .map(|index| index as u32)
                    .ok_or(DbError::Client("collection was not opened through this Db"))
            }
        }
    }

    pub(crate) fn rank_of(&self, participant: &dyn TxParticipant) -> u8 {
        match self {
            Self::Borrowed(db) => db.tx_rank_of(participant),
            Self::Runtime(runtime) => crate::sync::lock(&runtime.ranks)
                .get(participant.name())
                .copied()
                .unwrap_or(0),
        }
    }

    pub(crate) fn maybe_checkpoint(&self) -> DbResult<()> {
        if self.log().total_bytes() <= self.config().checkpoint_bytes {
            return Ok(());
        }
        let cut = self.log().in_flight_floor();
        if cut == 0 {
            return Ok(());
        }
        match self {
            Self::Borrowed(db) => db.checkpoint_now().map(|_| ()),
            Self::Runtime(runtime) => {
                checkpoint::durable_flush_all(&runtime.collections)?;
                self.log().checkpoint(cut).map(|_| ())
            }
        }
    }
}

pub(crate) enum Action {
    Put(Vec<u8>),
    /// `insert`: write only if the key is absent; otherwise abort the commit
    /// with [`DbError::KeyExists`]. Decided on the **raw** stored bytes — the
    /// existing value is never decoded, matching what a collection-level
    /// `insert` does.
    Insert(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);

/// Skip post-migration projection reconciliation (tests only): leaves
/// `projections_dirty` set after `save_info`, simulating a crash between
/// migration and repair.
#[cfg(any(test, feature = "test-faults"))]
pub(crate) static SKIP_PROJECTION_RECONCILE: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

/// 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) context: TxContext<'d>,
    pub(crate) ops: Vec<Op>,
    /// Explicit `reserve`/`release` intents (T10 Task 8) — kept separate from
    /// `ops` because neither becomes a raw write: both land in
    /// `resolve_claims`'s stage instead, merged with whatever a projecting
    /// source in `ops` claims (see [`Tx::reserve`]).
    pub(crate) claim_ops: Vec<ExplicitClaim>,
    pub(crate) durable: bool,
    pub(crate) after_commit: Vec<Box<dyn FnOnce() + Send>>,
    pub(crate) queue_producer_permits: Vec<crate::queue::retention::ProducerPermit>,
    pub(crate) queue_depth_admissions: crate::queue::QueueDepthAdmissions,
}

/// 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(crate) fn from_runtime(runtime: Arc<TxRuntime>) -> Tx<'static> {
        let durable = runtime.config.durable_by_default;
        Tx {
            context: TxContext::Runtime(runtime),
            ops: Vec::new(),
            claim_ops: Vec::new(),
            durable,
            after_commit: Vec::new(),
            queue_producer_permits: Vec::new(),
            queue_depth_admissions: crate::queue::QueueDepthAdmissions::default(),
        }
    }

    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.claim_ops.len(),
            self.context.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)
    }

    /// Write `value` at `key` **only if the key is absent**, failing the whole
    /// transaction with [`DbError::KeyExists`] if it is not.
    ///
    /// The transactional twin of a collection's `insert`. Presence is decided
    /// at `commit()` under the transaction's shard locks, so two transactions
    /// racing for the same free key produce exactly one winner.
    ///
    /// # Failure is pre-log
    ///
    /// The check runs in the commit's evaluation round, before any GSN is
    /// reserved and long before the log append: on `KeyExists` nothing is
    /// written, no index row is projected, no counter moves, no hook fires and
    /// no shard is fenced. The caller may retry or surface a client error
    /// (`KeyExists` maps to HTTP 409).
    ///
    /// # Relationship to `guard` + `put`
    ///
    /// For a **decodable** stored value this is the same outcome as
    ///
    /// ```ignore
    /// tx.guard(&c, &k, |cur| if cur.is_none() { Ok(()) } else { Err(DbError::KeyExists) })?
    ///   .put(&c, &k, &v)?
    /// ```
    ///
    /// at one operation against `max_ops` instead of two, and with the key
    /// written once instead of twice.
    ///
    /// The two forms are **not** equivalent on an undecodable row.
    /// [`guard`](Self::guard) hands its closure a typed value and therefore
    /// decodes first, so a corrupt row makes it fail with the codec's error;
    /// `insert` tests only whether bytes are present and returns `KeyExists`.
    /// That is deliberate — it is what a collection-level `insert` does, since
    /// it rejects on the index before any decode. A uniqueness claim stays
    /// rejected whether or not what already occupies the key can still be read.
    ///
    /// `guard` remains the tool for the more general "absent **or** matching
    /// what we expect" precondition; `insert` does not serve that case.
    ///
    /// Within one transaction the check sees the transaction's own pending
    /// writes first: `put` then `insert` on one key is `KeyExists`, while
    /// `delete` then `insert` succeeds.
    pub fn insert<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.claim_ops.len(),
            self.context.config().max_ops,
            key_bytes.len(),
        )?;
        self.ops.push(Op {
            participant: collection.clone() as Arc<dyn TxParticipant>,
            key: key_bytes,
            action: Action::Insert(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.claim_ops.len(),
            self.context.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)
    }

    /// Queue bounded deletion of keys selected by `scope`.
    ///
    /// This only builds source delete operations; it never writes before
    /// [`commit`](Self::commit). `Selection::done` is false whenever the caller's
    /// limit or remaining source-operation capacity truncated the range. It is
    /// not a snapshot: a key inserted after selection can survive even when
    /// `done` is true. A zero limit always returns
    /// `Selection { queued: 0, done: false }`.
    ///
    /// The selected source operations may still expand at commit time. Final
    /// derived `max_ops`, record bytes, queue admission, and projection budgets
    /// can reject the transaction before log append; rebuilding the same scope
    /// with a smaller limit (normally halving it) can make those size-related
    /// refusals fit. `UniqueViolation`, `CounterTargetMissing`,
    /// `AmbiguousCounterTarget`, and decode failures are data validation errors:
    /// they are also pre-log, but retrying a smaller chunk does not repair them.
    /// A useful initial size is
    /// `chunk * (1 + derived rows per source row) <= max_ops`.
    ///
    /// Every call selects currently stored keys independently. Do not call this
    /// method repeatedly for the same scope on one transaction when the intent is
    /// to advance a cursor: queued deletes are not removed from its next storage
    /// walk. Commit one chunk, then start a new transaction for the next chunk.
    pub fn delete_scope<S, K, const N: usize>(
        mut self,
        scope: &TxScope<S, K, N>,
        limit: usize,
    ) -> DbResult<(Self, Selection)>
    where
        K: Key,
        S: TxTarget<Key = K> + TxParticipant + 'static,
    {
        if limit == 0 {
            return Ok((
                self,
                Selection {
                    queued: 0,
                    done: false,
                },
            ));
        }

        let capacity = self
            .context
            .config()
            .max_ops
            .saturating_sub(self.ops.len().saturating_add(self.claim_ops.len()));
        let source_limit = limit.min(capacity);
        if source_limit == 0 {
            return Ok((
                self,
                Selection {
                    queued: 0,
                    done: false,
                },
            ));
        }
        let mut keys = scope.keys();
        let mut queued = 0;
        for key in keys.by_ref().take(source_limit) {
            let key_bytes = key.as_bytes().to_vec();
            check_op_limits(
                self.ops.len() + self.claim_ops.len(),
                self.context.config().max_ops,
                key_bytes.len(),
            )?;
            self.ops.push(Op {
                participant: scope.collection.clone() as Arc<dyn TxParticipant>,
                key: key_bytes,
                action: Action::Delete,
            });
            queued += 1;
        }
        let done = queued < source_limit || keys.next().is_none();
        Ok((self, Selection { queued, done }))
    }

    /// 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.claim_ops.len(),
            self.context.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.claim_ops.len(),
            self.context.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)
    }

    /// Claim `key` in `ns` on behalf of `source`/`owner_key`, before that
    /// source row exists — a reservation held until `ttl` elapses, unless
    /// the source row itself claims the same key first (in this transaction
    /// or a later one), which promotes it to a permanent claim. That
    /// promotion is not a special case: it is the ordinary "owner is me, has
    /// a deadline" branch of [`unique::resolve::decide`](crate::unique::resolve::decide),
    /// the same rule an explicit [`reserve`](Self::reserve) and a projected
    /// claim on the same key resolve through in one transaction — see
    /// `txn::projection_commit::resolve_claims`'s doc for exactly how that
    /// ordering is made deterministic.
    ///
    /// Like [`put`](Self::put)/[`delete`](Self::delete), this only queues the
    /// intent — nothing is written until [`commit`](Self::commit). Unlike
    /// them, the queued intent is **not** a raw write: it is decided in the
    /// same claim-resolution stage a projected claim is
    /// (`txn::projection_commit::resolve_claims`), so it obeys the same
    /// release-before-acquisition partition instead of colliding with a
    /// projected claim on the same key through an ordinary last-write-wins
    /// overlay ([`Action::Put`]'s `set_write` merge).
    ///
    /// `ttl` is validated **here**, before any lock is taken:
    /// [`DbError::UniqueReservationInvalid`] with reason `"zero duration"`
    /// for `Duration::ZERO` (an expired-on-arrival reservation is
    /// indistinguishable from a leak — call [`release`](Self::release)
    /// instead), or `"deadline overflow"` for a duration whose millisecond
    /// count does not fit `u64`. A sub-millisecond, non-zero duration rounds
    /// up to 1 ms. The reservation's actual deadline is **not** `now() +
    /// ttl` computed here — the resolver stamps it at `commit()`, under the
    /// namespace's own lock, from one clock read shared by every reservation
    /// that commit resolves.
    pub fn reserve<N, S>(
        self,
        ns: &Arc<UniqueIndex<N>>,
        key: N::SelfId,
        source: &Arc<S>,
        owner_key: &S::Key,
        ttl: std::time::Duration,
    ) -> DbResult<Self>
    where
        N: UniqueNamespace,
        N::SelfId: Key + Send + Sync + std::hash::Hash + Eq,
        S: TxTarget + TxParticipant + 'static,
    {
        let ttl_ms = reservation_ttl_ms(ttl, N::NAME)?;
        self.push_explicit_claim(
            ns,
            key,
            source,
            owner_key,
            ClaimKind::Reserve {
                deadline_ms: ttl_ms,
            },
        )
    }

    /// Give up `key` in `ns` if `source`/`owner_key` currently holds it —
    /// permanent claim or reservation alike. Idempotent: releasing an
    /// already-free key is a no-op success at `commit()`. Refuses the whole
    /// commit with [`DbError::UniqueViolation`] if a *different* owner holds
    /// the key — a release never deletes a row it does not own, the same
    /// rule a projected release follows
    /// ([`unique::resolve::decide`](crate::unique::resolve::decide)'s
    /// foreign-owner arm, which every [`crate::txn::participant::ClaimKind`]
    /// shares).
    ///
    /// Like [`reserve`](Self::reserve), this queues a claim-stage intent, not
    /// a raw write — see that method's doc.
    pub fn release<N, S>(
        self,
        ns: &Arc<UniqueIndex<N>>,
        key: N::SelfId,
        source: &Arc<S>,
        owner_key: &S::Key,
    ) -> DbResult<Self>
    where
        N: UniqueNamespace,
        N::SelfId: Key + Send + Sync + std::hash::Hash + Eq,
        S: TxTarget + TxParticipant + 'static,
    {
        self.push_explicit_claim(ns, key, source, owner_key, ClaimKind::Release)
    }

    fn push_explicit_claim<N, S>(
        mut self,
        ns: &Arc<UniqueIndex<N>>,
        key: N::SelfId,
        source: &Arc<S>,
        owner_key: &S::Key,
        kind: ClaimKind,
    ) -> DbResult<Self>
    where
        N: UniqueNamespace,
        N::SelfId: Key + Send + Sync + std::hash::Hash + Eq,
        S: TxTarget + TxParticipant + 'static,
    {
        let key_bytes = key.as_bytes().to_vec();
        check_op_limits(
            self.ops.len() + self.claim_ops.len(),
            self.context.config().max_ops,
            key_bytes.len(),
        )?;
        // Review round I1: `ns` must be a claim target `source` actually
        // declared (`Projected::and_unique`/`UniqueTargets`) — otherwise the
        // row lands tagged with `source`'s own collection, blocking whatever
        // the *real* claiming source later tries to write, and nothing ever
        // reports it (unlike the projected path, whose `ClaimSink::claim`
        // and `project_rank0` both refuse an undeclared target loudly). A
        // `UniqueIndex` participant can only reach
        // `tx_preplan_projection_targets()` through `UniqueTargets`, so exact
        // membership by `Arc::ptr_eq` is the same check the projected path
        // gets for free, done here before any lock is taken.
        let participant = ns.as_tx_participant();
        let declared = source
            .tx_preplan_projection_targets()?
            .iter()
            .any(|t| Arc::ptr_eq(t, &participant));
        if !declared {
            return Err(DbError::UndeclaredUniqueTarget { namespace: N::NAME });
        }
        // The *source's* own key, not the namespace's — the same tag+key
        // pair a projected claim's owner carries (`into_tx_claim_ops`), built
        // with `expires_at = 0`: this is who holds the row, not whether the
        // hold expires — that is `kind`'s job, resolved later.
        let owner = UniqueOwner::new(collection_tag(source.name()), owner_key.as_bytes(), 0)
            .ok_or(DbError::Internal(
                "reservation owner key exceeds the owner capacity",
            ))?;
        self.claim_ops.push(ExplicitClaim {
            participant,
            namespace: N::NAME,
            key: key_bytes,
            owner,
            kind,
        });
        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(mut 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() && self.claim_ops.is_empty() {
            return Ok(());
        }

        let claim_ops = self.claim_ops;
        let projection_commit::PlanOutcome {
            entries,
            rank0_len,
            rank0_ops,
            rank1_ops,
        } = projection_commit::plan(&self.context, self.ops, &claim_ops)?;

        for op in rank0_ops.iter().chain(rank1_ops.iter()) {
            if op.participant.encrypted() && !self.context.log().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",
                ));
            }
        }
        for ec in &claim_ops {
            projection_commit::check_derived_encryption(
                &ec.participant,
                self.context.log().is_encrypted(),
            )?;
        }
        for entry in &entries[..rank0_len] {
            self.queue_producer_permits
                .extend(entry.participant.tx_prepare_streams()?);
        }

        let commit_guard = crate::txn_depth::TxCommitGuard::enter();
        let mut locked: Vec<Box<dyn TxLocked + '_>> = Vec::with_capacity(entries.len());
        for entry in &entries[..rank0_len] {
            match entry.participant.lock_shards(&entry.shards) {
                Ok(l) => locked.push(l),
                Err(e) => {
                    projection_commit::abort_locked(locked);
                    return Err(e);
                }
            }
        }

        let (mut writes, projection_inputs) =
            match projection_commit::eval_rank0(&entries, &mut locked, rank0_ops) {
                Ok(v) => v,
                Err(e) => {
                    projection_commit::abort_locked(locked);
                    return Err(e);
                }
            };

        let projection_commit::Rank0Derived {
            extra_shards,
            counters,
            claims,
            stream_rows,
        } = match projection_commit::project_rank0(
            &entries,
            &mut writes,
            projection_inputs,
            self.context.log().is_encrypted(),
        ) {
            Ok(s) => s,
            Err(e) => {
                projection_commit::abort_locked(locked);
                return Err(e);
            }
        };

        for (offset, entry) in entries[rank0_len..].iter().enumerate() {
            let entry_pos = rank0_len + offset;
            let all_shards;
            let shards = if entry.lock_all {
                all_shards = (0..entry.participant.shard_count()).collect::<Vec<_>>();
                &all_shards
            } else {
                all_shards = merge_lock_shards(&entry.shards, extra_shards.get(&entry_pos));
                &all_shards
            };
            match entry.participant.lock_shards(shards) {
                Ok(l) => locked.push(l),
                Err(e) => {
                    projection_commit::abort_locked(locked);
                    return Err(e);
                }
            }
        }

        if let Err(e) =
            projection_commit::eval_rank1(&entries, &mut locked, rank0_len, rank1_ops, &mut writes)
        {
            projection_commit::abort_locked(locked);
            return Err(e);
        }

        let after_success = match projection_commit::lower_streams(
            &entries,
            &mut writes,
            stream_rows,
            self.context.log().is_encrypted(),
            &mut self.queue_depth_admissions,
        ) {
            Ok(v) => v,
            Err(e) => {
                projection_commit::abort_locked(locked);
                return Err(e);
            }
        };
        self.after_commit.extend(after_success);

        // Claims before counters (T10 Task 7): a conflict is cheaper to
        // discover before counter arithmetic runs, and a refused transaction
        // must not have run any.
        //
        // T10 Task 8: explicit `reserve`/`release` ops are placed *ahead* of
        // whatever a projecting source in this same transaction claimed —
        // see `resolve_claims`'s own doc for why that order (not the order
        // the caller chained `.reserve()`/`.put()`) is what makes a
        // same-transaction takeover land correctly.
        let explicit_claims =
            match projection_commit::pending_claims_from_explicit(&entries, claim_ops) {
                Ok(v) => v,
                Err(e) => {
                    projection_commit::abort_locked(locked);
                    return Err(e);
                }
            };
        let mut all_claims = explicit_claims;
        all_claims.extend(claims);
        if let Err(e) = projection_commit::resolve_claims(&mut locked, all_claims, &mut writes) {
            projection_commit::abort_locked(locked);
            return Err(e);
        }

        // Counters last: a delta must see the transaction's own writes to the
        // target row, so it is resolved after every explicit write has been
        // decided — and under the target's lock, which is what lets the log
        // carry an absolute value instead of a change.
        if let Err(e) = projection_commit::resolve_counters(&mut locked, counters, &mut writes) {
            projection_commit::abort_locked(locked);
            return Err(e);
        }

        if writes.is_empty() {
            projection_commit::abort_locked(locked);
            return Ok(());
        }

        if let Err(e) = projection_commit::admit(&self.context, &entries, &writes) {
            projection_commit::abort_locked(locked);
            return Err(e);
        }

        // ── Reserve, encode, append
        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.context.log().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.context.config().max_bytes {
                return Err(projection_commit::max_bytes_err(&writes));
            }
            let committed = self.context.log().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.context.log(),
            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.
            #[cfg(any(test, feature = "test-faults"))]
            let injected = failpoints::FAIL_LOG_SYNC.load(std::sync::atomic::Ordering::SeqCst);
            #[cfg(not(any(test, feature = "test-faults")))]
            let injected = false;

            let outcome = if injected {
                Err(DbError::Internal("injected log sync failure"))
            } else {
                self.context.log().sync_upto(&committed)
            };
            if let Err(e) = outcome {
                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(any(test, feature = "test-faults")),
            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(any(test, feature = "test-faults"))]
            if _index == failpoints::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 transaction_wakes = WatchWakeBatch::default();
        let mut finish_err = None;
        // `locked[i]` is `entries[i]` — the reservation loop above indexes both
        // by the same `pos`/`w.entry`. That is where the collection name for the
        // report comes from.
        let mut fenced: Vec<crate::FencedShard> = Vec::new();
        for (i, l) in locked.into_iter().enumerate() {
            let collection = entries[i].participant.name();
            let mut local: Vec<(usize, u64)> = Vec::new();
            // `transaction_wakes` is passed in rather than taken off the
            // returned deferred, so a participant whose `finish` fails still
            // hands back the wakers of the events it already published.
            match l.finish(&mut local, &mut transaction_wakes) {
                Ok(d) => {
                    deferred.push(d);
                }
                Err(e) => {
                    tracing::error!(tx_id, "releasing a transaction participant failed: {e}");
                    finish_err.get_or_insert(e);
                }
            }
            fenced.extend(local.into_iter().map(|(shard, gsn)| crate::FencedShard {
                collection,
                shard,
                gsn,
            }));
        }
        // Deterministic order for tests and logs.
        fenced.sort_unstable_by_key(|f| (f.collection, f.shard));
        // 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
        transaction_wakes.drain();

        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.context.maybe_checkpoint() {
            tracing::error!(tx_id, "checkpoint after commit failed: {e}");
        }
        let after_commit_ready = apply_err.is_none() && finish_err.is_none();
        let result = match stage_and_source(apply_err, finish_err, sync_err) {
            Some((stage, source)) => Err(DbError::TxFailedAfterLog {
                tx_id,
                stage,
                fenced,
                source: Box::new(source),
            }),
            None => Ok(()),
        };
        if !after_commit_ready && matches!(result, Err(DbError::TxFailedAfterLog { .. })) {
            self.context
                .pin_post_log_queue_producer_permits(std::mem::take(
                    &mut self.queue_producer_permits,
                ));
        }
        if after_commit_ready {
            for callback in self.after_commit {
                callback();
            }
        }
        result
    }
}

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);
        }
    }
}

fn build_record(
    tx_id: u64,
    entries: &[projection_commit::Entry],
    writes: &[projection_commit::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 }
}

/// Pick the reported failure out of the three post-log-point sources.
///
/// The priority is `apply -> finish -> sync`, which is the order the commit
/// path has always used; the stage is derived from the same choice, so it can
/// never disagree with the error it accompanies.
fn stage_and_source(
    apply_err: Option<DbError>,
    finish_err: Option<DbError>,
    sync_err: Option<DbError>,
) -> Option<(crate::TxFailStage, DbError)> {
    apply_err
        .map(|e| (crate::TxFailStage::Apply, e))
        .or_else(|| finish_err.map(|e| (crate::TxFailStage::Finish, e)))
        .or_else(|| sync_err.map(|e| (crate::TxFailStage::Sync, e)))
}

/// Merge a rank-1 participant's declared shards with the shards its derived
/// (projected) writes landed in.
///
/// `TxParticipant::lock_shards` requires an **ascending, deduplicated** slice:
/// that global rank order is what makes the two-round commit deadlock-free.
/// Derived shard ids arrive in projection-emission order, so two transactions
/// projecting into the same pair of target shards in opposite orders would
/// otherwise take them as A-then-B and B-then-A.
fn merge_lock_shards(base: &[usize], extra: Option<&Vec<usize>>) -> Vec<usize> {
    let mut shards = base.to_vec();
    if let Some(extra) = extra {
        for sid in extra {
            if !shards.contains(sid) {
                shards.push(*sid);
            }
        }
    }
    shards.sort_unstable();
    shards
}

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

    #[test]
    fn merged_lock_shards_are_ascending_and_deduplicated() {
        // Emission order, not shard order — exactly what `project_rank0` returns.
        let merged = merge_lock_shards(&[5], Some(&vec![9, 2, 5, 7]));
        assert_eq!(merged, vec![2, 5, 7, 9]);

        // The mirror-image transaction must produce the same lock order.
        let mirrored = merge_lock_shards(&[5], Some(&vec![7, 5, 2, 9]));
        assert_eq!(merged, mirrored);

        assert_eq!(merge_lock_shards(&[3, 1], None), vec![1, 3]);
    }

    #[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}");
    }

    #[test]
    fn stage_follows_the_apply_finish_sync_priority() {
        use crate::TxFailStage::*;

        assert!(stage_and_source(None, None, None).is_none());

        let only_finish = stage_and_source(None, Some(DbError::Internal("f")), None).unwrap();
        assert_eq!(only_finish.0, Finish);
        assert!(format!("{}", only_finish.1).contains('f'));

        let only_sync = stage_and_source(None, None, Some(DbError::Internal("s"))).unwrap();
        assert_eq!(only_sync.0, Sync);

        // Apply wins over both, and the reported source is apply's.
        let all_three = stage_and_source(
            Some(DbError::Internal("a")),
            Some(DbError::Internal("f")),
            Some(DbError::Internal("s")),
        )
        .unwrap();
        assert_eq!(all_three.0, Apply);
        assert!(format!("{}", all_three.1).contains('a'));
    }
}

#[cfg(test)]
#[cfg(all(feature = "armour", feature = "rapira-codec"))]
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;
    use zerocopy::{FromBytes, Immutable, IntoBytes};

    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() {
        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.
            let _fail = crate::txn::failpoints::ApplyFailGuard::at(0);
            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();
            drop(_fail);
            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() {
        let dir = tempfile::tempdir().unwrap();
        {
            let db = Arc::new(Db::open_test(dir.path()).unwrap());
            let blobs = open_blobs(&db);

            let _fail = crate::txn::failpoints::ApplyFailGuard::at(0);
            let err = db
                .tx()
                .put(
                    &blobs,
                    &user_id(1),
                    &Blob {
                        payload: vec![7u8; 8],
                    },
                )
                .unwrap()
                .durable()
                .commit()
                .unwrap_err();
            drop(_fail);
            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"
        );
    }

    /// The post-log wrapper is what a consumer classifies on: an apply failure
    /// must name the stage and the shard that now owes the write, and that
    /// shard's `gsn` must be the one a later write reports back.
    #[test]
    fn tx_apply_failure_reports_stage_and_fenced_shard() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, counters) = open_pair(&db);

        let _fail = crate::txn::failpoints::ApplyFailGuard::at(0);
        let err = db
            .tx()
            .put(&users, &user_id(1), &User { active: true })
            .unwrap()
            .put(&counters, &user_id(1), &Counter { n: 1 })
            .unwrap()
            .commit()
            .unwrap_err();
        drop(_fail);

        let crate::DbError::TxFailedAfterLog {
            stage,
            fenced,
            source,
            ..
        } = &err
        else {
            panic!("expected TxFailedAfterLog, got {err}");
        };
        assert_eq!(*stage, crate::TxFailStage::Apply);
        assert!(
            format!("{source}").contains("injected apply failure"),
            "{source}"
        );
        assert_eq!(fenced.len(), 1, "{fenced:?}");
        assert_eq!(fenced[0].collection, "users");

        // The reported gsn is exactly what the fence then refuses on. Re-use the
        // *same* key: `Config::test()` gives the collection two shards
        // (`config.rs:203-205`), so another key could route to the unfenced one.
        let reported = fenced[0].gsn;
        let again = users.put(&user_id(1), User { active: false });
        match again {
            Err(crate::DbError::ShardFenced { gsn, .. }) => assert_eq!(gsn, reported),
            Ok(_) => panic!("expected ShardFenced, got Ok"),
            Err(other) => panic!("expected ShardFenced, got {other}"),
        }
    }

    /// A `durable()` transaction whose log fsync fails is still applied — the
    /// record is in the log and applying it is correct. The caller must not
    /// retry it, so it is wrapped; but nothing is fenced and no reopen helps.
    #[test]
    fn a_failed_log_sync_reports_the_sync_stage_and_fences_nothing() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, counters) = open_pair(&db);

        let _fail = crate::txn::failpoints::LogSyncFailGuard::arm();
        let err = db
            .tx()
            .put(&users, &user_id(40), &User { active: true })
            .unwrap()
            .put(&counters, &user_id(40), &Counter { n: 4 })
            .unwrap()
            .durable()
            .commit()
            .unwrap_err();
        drop(_fail);

        let crate::DbError::TxFailedAfterLog {
            stage,
            fenced,
            source,
            ..
        } = &err
        else {
            panic!("expected TxFailedAfterLog, got {err}");
        };
        assert_eq!(*stage, crate::TxFailStage::Sync);
        assert!(
            fenced.is_empty(),
            "a sync failure fences nothing: {fenced:?}"
        );
        assert!(
            format!("{source}").contains("injected log sync failure"),
            "{source}"
        );
        assert_eq!(err.status_code(), 500);

        // Apply continued: this is precisely why the caller must not retry.
        assert_eq!(users.get(&user_id(40)).map(|u| u.active), Some(true));
        assert_eq!(counters.get(&user_id(40)).map(|c| c.n), Some(4));
    }

    /// One injected failure fences exactly one shard: the apply loop skips only
    /// later writes on the *same* `(entry, shard)`, so other shards apply and
    /// spend their reservations. Two fenced shards therefore need two earlier
    /// transactions — and the third transaction proves the list holds both.
    ///
    /// It also pins the min-merge: the reported gsn comes from the *first*
    /// fence, not from the higher reservation the third transaction took.
    #[test]
    fn several_fenced_shards_are_reported_together_with_their_earliest_gsn() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, counters) = open_pair(&db);

        // Fence `users` on its own.
        let _fail = crate::txn::failpoints::ApplyFailGuard::at(0);
        let first = db
            .tx()
            .put(&users, &user_id(1), &User { active: true })
            .unwrap()
            .commit()
            .unwrap_err();
        let crate::DbError::TxFailedAfterLog { fenced: f1, .. } = &first else {
            panic!("expected TxFailedAfterLog, got {first}");
        };
        let users_gsn = f1[0].gsn;

        // Fence `counters` on its own.
        let second = db
            .tx()
            .put(&counters, &user_id(1), &Counter { n: 1 })
            .unwrap()
            .commit()
            .unwrap_err();
        let crate::DbError::TxFailedAfterLog { fenced: f2, .. } = &second else {
            panic!("expected TxFailedAfterLog, got {second}");
        };
        let counters_gsn = f2[0].gsn;
        drop(_fail);

        // No injection now: both applies fail on their own fences, so both
        // reservations stay unspent and both shards are reported.
        //
        // The keys are the *same* ones as above on purpose. `Config::test()`
        // gives each collection two shards (`config.rs:203-205`), so a
        // different key could route to the unfenced shard and apply cleanly.
        let third = db
            .tx()
            .put(&users, &user_id(1), &User { active: false })
            .unwrap()
            .put(&counters, &user_id(1), &Counter { n: 2 })
            .unwrap()
            .commit()
            .unwrap_err();
        let crate::DbError::TxFailedAfterLog { stage, fenced, .. } = &third else {
            panic!("expected TxFailedAfterLog, got {third}");
        };
        assert_eq!(*stage, crate::TxFailStage::Apply);
        assert_eq!(fenced.len(), 2, "{fenced:?}");

        let by_name = |name: &str| {
            fenced
                .iter()
                .find(|f| f.collection == name)
                .unwrap_or_else(|| panic!("{name} missing from {fenced:?}"))
        };
        // The earlier, lower fence wins — this is what a dropped return value
        // in one of the six `tx_fence` wrappers would break.
        assert_eq!(by_name("users").gsn, users_gsn);
        assert_eq!(by_name("counters").gsn, counters_gsn);
    }

    /// Two unapplied reservations on one shard are one fenced shard, reported
    /// with the lower gsn.
    ///
    /// This needs a **single-shard** collection, which is exactly what the
    /// existing `open_blobs` fixture builds (`shard_count(1)`); `Config::test()`
    /// gives two shards, and two keys there could land apart.
    #[test]
    fn one_shard_is_reported_once_even_with_two_unapplied_reservations() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let blobs = open_blobs(&db);

        let _fail = crate::txn::failpoints::ApplyFailGuard::at(0);
        let err = db
            .tx()
            .put(
                &blobs,
                &user_id(10),
                &Blob {
                    payload: vec![1, 2, 3],
                },
            )
            .unwrap()
            .put(
                &blobs,
                &user_id(11),
                &Blob {
                    payload: vec![4, 5, 6],
                },
            )
            .unwrap()
            .commit()
            .unwrap_err();
        drop(_fail);

        let crate::DbError::TxFailedAfterLog { fenced, .. } = &err else {
            panic!("expected TxFailedAfterLog, got {err}");
        };
        assert_eq!(fenced.len(), 1, "one shard, one entry: {fenced:?}");
        assert_eq!(fenced[0].collection, "blobs");

        let reported = fenced[0].gsn;
        let outcome = blobs.put(&user_id(12), Blob { payload: vec![7] });
        match outcome {
            Err(crate::DbError::ShardFenced { gsn, .. }) => assert_eq!(gsn, reported),
            Ok(_) => panic!("expected ShardFenced, got Ok"),
            Err(other) => panic!("expected ShardFenced, got {other}"),
        }
    }

    /// A single-shard FixedStore marker collection, used only by
    /// `a_fixed_participant_is_never_reported_as_fenced` below.
    #[derive(Clone, Copy, Debug, PartialEq, Eq, GetType, FromBytes, IntoBytes, Immutable)]
    #[repr(C)]
    pub(super) struct SeenMarker {
        pub at: u64,
    }
    impl CollectionMeta for SeenMarker {
        type SelfId = UserId;
        const NAME: &'static str = "seen";
    }

    pub(super) fn open_seen(
        db: &Arc<Db>,
    ) -> Arc<
        crate::ZeroMap<
            UserId,
            8,
            SeenMarker,
            TtlHook<SeenMarker, NoHook>,
            crate::durability::Fixed,
        >,
    > {
        db.open_zero_map_fixed::<SeenMarker, 8, _>(crate::FixedConfig::test(), NoHook, &[])
            .unwrap()
    }

    /// FixedStore is not fenced: its replay rule reads the reserved slot's
    /// exact state, so a mixed transaction that fails on the Bitcask side must
    /// report exactly one fenced shard — the Bitcask one, never the Fixed
    /// participant that shared the transaction.
    ///
    /// This is the Task 3 FixedStore test (brief §3). It lives here rather
    /// than in `armdb/tests/tx_fixed.rs` because it predates
    /// `failpoints::APPLY_FAIL_AT` becoming public (Task 4 publishes it
    /// through [`failpoints`], reachable from an integration test built with
    /// `feature = "test-faults"`); `mod commit_tests` can build the same
    /// mixed Bitcask+FixedStore transaction the fixture in `tx_fixed.rs`
    /// uses, so the property was verified here instead of being deferred.
    /// Task 6 adds the equivalent as an external contract test.
    #[test]
    fn a_fixed_participant_is_never_reported_as_fenced() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, _counters) = open_pair(&db);
        let seen = open_seen(&db);

        // The Bitcask write is inserted first, so index 0 targets it; the
        // Fixed write (index 1) applies normally and needs no fence.
        let _fail = crate::txn::failpoints::ApplyFailGuard::at(0);
        let err = db
            .tx()
            .put(&users, &user_id(1), &User { active: true })
            .unwrap()
            .put(&seen, &user_id(1), &SeenMarker { at: 1 })
            .unwrap()
            .commit()
            .unwrap_err();
        drop(_fail);

        let crate::DbError::TxFailedAfterLog { fenced, .. } = &err else {
            panic!("expected TxFailedAfterLog, got {err}");
        };
        assert_eq!(fenced.len(), 1, "{fenced:?}");
        assert_eq!(fenced[0].collection, "users");
        assert!(
            fenced.iter().all(|f| f.collection != SeenMarker::NAME),
            "a FixedStore participant must never be reported as fenced: {fenced:?}"
        );
    }

    /// A failure *before* the append is a plain error: nothing was written, so
    /// the caller may retry. Wrapping it would tell them the opposite.
    #[test]
    fn tx_failure_before_the_log_point_is_not_wrapped() {
        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(5), &User { active: true })
            .unwrap()
            // `Tx::guard` takes `impl FnOnce(Option<&S::Val>) -> DbResult<()>`
            // (`armdb/src/txn/mod.rs:257-262`); the parameter type is annotated
            // because a bare `|_|` gives inference nothing to work with.
            .guard(&users, &user_id(5), |_cur: Option<&User>| {
                Err(crate::DbError::Client("nope"))
            })
            .unwrap()
            .commit()
            .unwrap_err();

        assert!(
            !matches!(err, crate::DbError::TxFailedAfterLog { .. }),
            "{err}"
        );
        assert!(matches!(err, crate::DbError::Client("nope")), "{err}");
        assert!(users.get(&user_id(5)).is_none(), "nothing was written");
    }

    /// 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() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let (users, counters) = open_pair(&db);

        let _fail = crate::txn::failpoints::ApplyFailGuard::at(1);
        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();
        drop(_fail);
        assert!(format!("{err}").contains("injected apply failure"), "{err}");
        assert!(
            matches!(
                err,
                crate::DbError::TxFailedAfterLog {
                    stage: crate::TxFailStage::Apply,
                    ..
                }
            ),
            "{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() {
        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);

        let _fail = crate::txn::failpoints::ApplyFailGuard::at(1);
        let _ = db
            .tx()
            .put(&users, &user_id(30), &User { active: true })
            .unwrap()
            .put(&counters, &user_id(30), &Counter { n: 3 })
            .unwrap()
            .commit()
            .unwrap_err();
        drop(_fail);
        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.
    // Liveness after a panic taken *under* the shard locks: only a mutex that
    // does not poison can offer it. `std::sync::Mutex` and loom's both mark the
    // lock poisoned and `sync::lock` panics on every later acquisition, so the
    // property this asserts holds under `parking_lot` and nothing else. The gate
    // mirrors `sync`'s backend predicate: loom wins when both features are on.
    #[cfg(all(feature = "parking_lot", not(feature = "loom")))]
    #[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.
    // Liveness after a panic taken *under* the shard locks: only a mutex that
    // does not poison can offer it. `std::sync::Mutex` and loom's both mark the
    // lock poisoned and `sync::lock` panics on every later acquisition, so the
    // property this asserts holds under `parking_lot` and nothing else. The gate
    // mirrors `sync`'s backend predicate: loom wins when both features are on.
    #[cfg(all(feature = "parking_lot", not(feature = "loom")))]
    #[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"
        );
    }
}