armdb 0.14.2

//! `ConstTree`/`ConstMap` — and therefore `ZeroTree`/`ZeroMap`, which are
//! `#[repr(transparent)]` wrappers over them.

use std::sync::Arc;

use crate::const_map::{ConstMap, ConstMapShard};
use crate::const_tree::{ConstOwnedEvent, ConstShard, ConstTree};
use crate::ctx::Family;
use crate::durability::{Bitcask, Durability, Fixed};
use crate::error::{DbError, DbResult};
use crate::hook::WriteHook;
use crate::key::Key;
use crate::txlog::record::{OpKind, VersionTag};
use crate::txn::TxTarget;
use crate::txn::held::{TxFamily, TxHeld, TxRaw};
use crate::txn::participant::{
    ClaimOp, DeferredReplay, PlannedOp, TxBackend, TxLocked, TxParticipant,
};
use crate::watch::WatchWakeBatch;

/// Reserve one FixedStore version per operation, rolling back what was already
/// taken if a later key fails.
///
/// Each absent key takes a slot from the allocator, which marks the bitmap. A
/// failure part-way — a full disk makes `alloc_slot`'s `grow()` return `Err` —
/// returns no tags at all, so `TxHeld` never learns about the earlier slots and
/// `abort` cannot hand them back: they stay marked used until a restart rebuilds
/// the bitmap from disk. The rollback therefore belongs in this loop, which is
/// the only place that knows them.
///
/// Shared by `ConstTree` and `ConstMap` so the rule has one implementation
/// rather than two copies (libfunc/armour#133).
fn reserve_fixed_in_shard<S>(
    shard: &mut S,
    ops: &[PlannedOp<'_>],
    mut reserve: impl FnMut(&mut S, &[u8], bool) -> DbResult<(u32, u32)>,
    release: impl Fn(&mut S, VersionTag),
) -> DbResult<Vec<VersionTag>> {
    let mut tags: Vec<VersionTag> = Vec::with_capacity(ops.len());
    for op in ops {
        match reserve(shard, op.key, op.delete) {
            Ok((slot_id, target_meta)) => tags.push(VersionTag::Fixed {
                slot_id,
                target_meta,
            }),
            Err(e) => {
                for tag in tags.into_iter().rev() {
                    release(shard, tag);
                }
                return Err(e);
            }
        }
    }
    Ok(tags)
}

/// Arc-backed deferred hook owner. `ConstOwnedEvent` is fully owned, so no
/// seize guard is required after lock release — unlike the Typed families.
pub(crate) struct ConstDeferred<
    K: Key + 'static,
    const V: usize,
    H: WriteHook<K> + 'static,
    D: Durability + 'static,
> {
    tree: Arc<ConstTree<K, V, H, D>>,
    events: Vec<ConstOwnedEvent<K, V>>,
}

impl<K: Key + 'static, const V: usize, H: WriteHook<K> + 'static, D: Durability + 'static>
    DeferredReplay for ConstDeferred<K, V, H, D>
{
    fn event_count(&self) -> usize {
        self.events.len()
    }
    fn replay_one(&mut self, local: usize) {
        if let Some(ev) = self.events.get(local) {
            self.tree.fire_owned(ev);
        }
    }
}

pub(crate) struct ConstMapDeferred<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    H: WriteHook<K> + 'static,
    D: Durability + 'static,
> {
    map: Arc<ConstMap<K, V, H, D>>,
    events: Vec<<ConstMap<K, V, H, D> as Family>::Owned>,
}

impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    H: WriteHook<K> + 'static,
    D: Durability + 'static,
> DeferredReplay for ConstMapDeferred<K, V, H, D>
{
    fn event_count(&self) -> usize {
        self.events.len()
    }
    fn replay_one(&mut self, local: usize) {
        if let Some(ev) = self.events.get(local) {
            self.map.fire_owned(ev);
        }
    }
}

impl<K: Key + 'static, const V: usize, H: WriteHook<K> + 'static> TxFamily
    for ConstTree<K, V, H, Bitcask>
{
    type Shard<'a>
        = ConstShard<'a, K, V, H, Bitcask>
    where
        Self: 'a;
    type Deferred<'a>
        = ConstDeferred<K, V, H, Bitcask>
    where
        Self: 'a;
    type ReleaseExtra = ();

    fn tx_shard_count(&self) -> usize {
        self.shard_count()
    }
    fn tx_shard_of(&self, key: &[u8]) -> usize {
        self.shard_for(&K::from_bytes(key))
    }
    fn tx_make_shard(&self, sid: usize) -> Self::Shard<'_> {
        self.tx_shard(sid)
    }
    fn tx_release(shard: Self::Shard<'_>) -> (Vec<<Self as Family>::Owned>, WatchWakeBatch, ()) {
        // The guard returned here is dropped immediately; `tx_deferred` does
        // not need a fresh one because `Owned` events carry copied bytes.
        let (events, wakes, guard) = Self::tx_release_shard(shard);
        drop(guard);
        (events, wakes, ())
    }
    fn tx_deferred(
        &self,
        events: Vec<<Self as Family>::Owned>,
        (): (),
    ) -> DbResult<Box<dyn DeferredReplay>> {
        Ok(Box::new(ConstDeferred {
            tree: self.tx_arc()?,
            events,
        }))
    }
}

impl<K: Key + 'static, const V: usize, H: WriteHook<K> + 'static> TxRaw
    for ConstTree<K, V, H, Bitcask>
{
    fn raw_read(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
        Ok(shard.try_get(&K::from_bytes(key))?.map(|v| v.to_vec()))
    }
    fn raw_contains(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<bool> {
        shard.try_contains(&K::from_bytes(key))
    }
    fn raw_put(&self, shard: &mut Self::Shard<'_>, key: &[u8], value: &[u8]) -> DbResult<()> {
        let bytes: [u8; V] = value
            .try_into()
            .map_err(|_| DbError::Client("transaction value has the wrong fixed length"))?;
        shard.put(&K::from_bytes(key), &bytes).map(drop)
    }
    fn raw_delete(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<()> {
        shard.delete(&K::from_bytes(key)).map(drop)
    }
    fn reserve_in_shard(
        &self,
        _shard: &mut Self::Shard<'_>,
        ops: &[PlannedOp<'_>],
    ) -> DbResult<Vec<VersionTag>> {
        // One contiguous block, handed out in apply order, so this shard's
        // entries keep ascending GSNs — the property the replay rule relies on.
        let base = self.tx_reserve_gsns(ops.len());
        Ok((0..ops.len())
            .map(|i| VersionTag::Bitcask {
                gsn: base + i as u64,
            })
            .collect())
    }

    fn arm_tag(shard: &mut Self::Shard<'_>, tag: VersionTag) -> DbResult<()> {
        let VersionTag::Bitcask { gsn } = tag else {
            return Err(DbError::Internal(
                "Fixed version tag on a Bitcask participant",
            ));
        };
        shard.tx_arm_gsn(gsn);
        Ok(())
    }

    fn disarm_tag(shard: &mut Self::Shard<'_>) {
        shard.tx_clear_armed_gsn();
    }

    fn release_reservation(_shard: &mut Self::Shard<'_>, _tag: VersionTag) {
        // A GSN hole is harmless: the counter is monotonic and the replay rule
        // compares against a watermark, not against a dense sequence.
    }
    fn fence_shard(shard: &mut Self::Shard<'_>, tag: VersionTag) -> Option<u64> {
        match tag {
            VersionTag::Bitcask { gsn } => Some(shard.tx_fence(gsn)),
            VersionTag::Fixed { .. } => None,
        }
    }
    fn buffered_events(shard: &Self::Shard<'_>) -> usize {
        shard.tx_buffered_events()
    }

    fn needs_replay(
        &self,
        _shard: &mut Self::Shard<'_>,
        key: &[u8],
        _kind: OpKind,
        tag: VersionTag,
    ) -> DbResult<bool> {
        let VersionTag::Bitcask { gsn } = tag else {
            return Err(DbError::Internal(
                "Fixed version tag on a Bitcask participant",
            ));
        };
        // A shard's entries are appended in ascending GSN order under its lock
        // and recovery truncates at the first invalid entry, so a GSN at or below
        // the watermark is durably present — or was superseded, which is equally
        // a reason not to replay.
        let sid = <Self as TxFamily>::tx_shard_of(self, key);
        Ok(gsn > self.tx_durable_recovered_gsn(sid))
    }
}

impl<K: Key + 'static, const V: usize, H: WriteHook<K> + 'static> TxParticipant
    for ConstTree<K, V, H, Bitcask>
{
    fn name(&self) -> &'static str {
        self.name()
    }
    fn backend(&self) -> TxBackend {
        TxBackend::Bitcask
    }
    fn schema_fingerprint(&self) -> u64 {
        self.typ_hash()
    }
    fn encrypted(&self) -> bool {
        self.tx_encrypted()
    }
    fn max_entry_bytes(&self) -> Option<usize> {
        Some(self.config().write_buffer_size)
    }
    fn shard_of(&self, key: &[u8]) -> DbResult<usize> {
        Ok(<Self as TxFamily>::tx_shard_of(self, key))
    }
    fn shard_count(&self) -> usize {
        <Self as TxFamily>::tx_shard_count(self)
    }
    fn lock_shards<'a>(&'a self, shards: &[usize]) -> DbResult<Box<dyn TxLocked + 'a>> {
        Ok(Box::new(TxHeld::lock(self, shards)))
    }
    fn tx_project(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
        self.tx_project_writes(key, old, new)
    }
    fn tx_stream_writes(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<crate::txn::participant::TxStreamWrites> {
        ConstTree::tx_stream_writes(self, key, old, new)
    }
    fn tx_counter_deltas(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
        self.tx_counter_deltas_for(key, old, new)
    }
    fn tx_claims(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<ClaimOp>> {
        self.tx_claims_for(self.name(), key, old, new)
    }
    fn tx_preplan_projection_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        self.tx_preplan_projection_targets()
    }
    fn tx_preplan_stream_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        Ok(ConstTree::tx_preplan_stream_targets(self))
    }
    fn tx_prepare_streams(&self) -> DbResult<Vec<crate::queue::retention::ProducerPermit>> {
        ConstTree::tx_prepare_streams(self)
    }
}

impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    H: WriteHook<K> + 'static,
> TxFamily for ConstMap<K, V, H, Bitcask>
{
    type Shard<'a>
        = ConstMapShard<'a, K, V, H, Bitcask>
    where
        Self: 'a;
    type Deferred<'a>
        = ConstMapDeferred<K, V, H, Bitcask>
    where
        Self: 'a;
    type ReleaseExtra = ();

    fn tx_shard_count(&self) -> usize {
        self.shard_count()
    }
    fn tx_shard_of(&self, key: &[u8]) -> usize {
        self.shard_for(&K::from_bytes(key))
    }
    fn tx_make_shard(&self, sid: usize) -> Self::Shard<'_> {
        self.tx_shard(sid)
    }
    fn tx_release(shard: Self::Shard<'_>) -> (Vec<<Self as Family>::Owned>, WatchWakeBatch, ()) {
        let (events, wakes) = Self::tx_release_shard(shard);
        (events, wakes, ())
    }
    fn tx_deferred(
        &self,
        events: Vec<<Self as Family>::Owned>,
        (): (),
    ) -> DbResult<Box<dyn DeferredReplay>> {
        Ok(Box::new(ConstMapDeferred {
            map: self.tx_arc()?,
            events,
        }))
    }
}

impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    H: WriteHook<K> + 'static,
> TxRaw for ConstMap<K, V, H, Bitcask>
{
    fn raw_read(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
        Ok(shard.try_get(&K::from_bytes(key))?.map(|v| v.to_vec()))
    }
    fn raw_contains(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<bool> {
        shard.try_contains(&K::from_bytes(key))
    }
    fn raw_put(&self, shard: &mut Self::Shard<'_>, key: &[u8], value: &[u8]) -> DbResult<()> {
        let bytes: [u8; V] = value
            .try_into()
            .map_err(|_| DbError::Client("transaction value has the wrong fixed length"))?;
        shard.put(&K::from_bytes(key), &bytes).map(drop)
    }
    fn raw_delete(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<()> {
        shard.delete(&K::from_bytes(key)).map(drop)
    }
    fn reserve_in_shard(
        &self,
        _shard: &mut Self::Shard<'_>,
        ops: &[PlannedOp<'_>],
    ) -> DbResult<Vec<VersionTag>> {
        // One contiguous block, handed out in apply order, so this shard's
        // entries keep ascending GSNs — the property the replay rule relies on.
        let base = self.tx_reserve_gsns(ops.len());
        Ok((0..ops.len())
            .map(|i| VersionTag::Bitcask {
                gsn: base + i as u64,
            })
            .collect())
    }

    fn arm_tag(shard: &mut Self::Shard<'_>, tag: VersionTag) -> DbResult<()> {
        let VersionTag::Bitcask { gsn } = tag else {
            return Err(DbError::Internal(
                "Fixed version tag on a Bitcask participant",
            ));
        };
        shard.tx_arm_gsn(gsn);
        Ok(())
    }

    fn disarm_tag(shard: &mut Self::Shard<'_>) {
        shard.tx_clear_armed_gsn();
    }

    fn release_reservation(_shard: &mut Self::Shard<'_>, _tag: VersionTag) {
        // A GSN hole is harmless: the counter is monotonic and the replay rule
        // compares against a watermark, not against a dense sequence.
    }
    fn fence_shard(shard: &mut Self::Shard<'_>, tag: VersionTag) -> Option<u64> {
        match tag {
            VersionTag::Bitcask { gsn } => Some(shard.tx_fence(gsn)),
            VersionTag::Fixed { .. } => None,
        }
    }
    fn buffered_events(shard: &Self::Shard<'_>) -> usize {
        shard.tx_buffered_events()
    }

    fn needs_replay(
        &self,
        _shard: &mut Self::Shard<'_>,
        key: &[u8],
        _kind: OpKind,
        tag: VersionTag,
    ) -> DbResult<bool> {
        let VersionTag::Bitcask { gsn } = tag else {
            return Err(DbError::Internal(
                "Fixed version tag on a Bitcask participant",
            ));
        };
        let sid = <Self as TxFamily>::tx_shard_of(self, key);
        Ok(gsn > self.tx_durable_recovered_gsn(sid))
    }
}

impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    H: WriteHook<K> + 'static,
> TxParticipant for ConstMap<K, V, H, Bitcask>
{
    fn name(&self) -> &'static str {
        self.name()
    }
    fn backend(&self) -> TxBackend {
        TxBackend::Bitcask
    }
    fn schema_fingerprint(&self) -> u64 {
        self.typ_hash()
    }
    fn encrypted(&self) -> bool {
        self.tx_encrypted()
    }
    fn max_entry_bytes(&self) -> Option<usize> {
        Some(self.config().write_buffer_size)
    }
    fn shard_of(&self, key: &[u8]) -> DbResult<usize> {
        Ok(<Self as TxFamily>::tx_shard_of(self, key))
    }
    fn lock_shards<'a>(&'a self, shards: &[usize]) -> DbResult<Box<dyn TxLocked + 'a>> {
        Ok(Box::new(TxHeld::lock(self, shards)))
    }
    fn tx_project(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
        self.tx_project_writes(key, old, new)
    }
    fn tx_stream_writes(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<crate::txn::participant::TxStreamWrites> {
        ConstMap::tx_stream_writes(self, key, old, new)
    }
    fn tx_counter_deltas(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
        self.tx_counter_deltas_for(key, old, new)
    }
    fn tx_claims(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<ClaimOp>> {
        self.tx_claims_for(self.name(), key, old, new)
    }
    fn tx_preplan_projection_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        self.tx_preplan_projection_targets()
    }
    fn tx_preplan_stream_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        Ok(ConstMap::tx_preplan_stream_targets(self))
    }
    fn tx_prepare_streams(&self) -> DbResult<Vec<crate::queue::retention::ProducerPermit>> {
        ConstMap::tx_prepare_streams(self)
    }
}

impl<K, const V: usize, T, H> TxParticipant for crate::zero_tree::ZeroTree<K, V, T, H, Bitcask>
where
    K: Key + 'static,
    T: armour_core::GetType
        + crate::txn::participant::TxCollectionName
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Send
        + Sync
        + 'static,
    H: crate::hook::TypedWriteHook<K, T> + 'static,
{
    fn name(&self) -> &'static str {
        T::tx_collection_name()
    }
    fn backend(&self) -> TxBackend {
        TxBackend::Bitcask
    }
    fn schema_fingerprint(&self) -> u64 {
        T::TYPE.h()
    }
    fn encrypted(&self) -> bool {
        TxParticipant::encrypted(self.tx_inner())
    }
    fn max_entry_bytes(&self) -> Option<usize> {
        TxParticipant::max_entry_bytes(self.tx_inner())
    }
    fn shard_of(&self, key: &[u8]) -> DbResult<usize> {
        TxParticipant::shard_of(self.tx_inner(), key)
    }
    fn shard_count(&self) -> usize {
        TxParticipant::shard_count(self.tx_inner())
    }
    fn lock_shards<'a>(&'a self, shards: &[usize]) -> DbResult<Box<dyn TxLocked + 'a>> {
        TxParticipant::lock_shards(self.tx_inner(), shards)
    }
    fn tx_project(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
        self.tx_project_writes(key, old, new)
    }
    fn tx_stream_writes(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<crate::txn::participant::TxStreamWrites> {
        TxParticipant::tx_stream_writes(self.tx_inner(), key, old, new)
    }
    fn tx_counter_deltas(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
        self.tx_counter_deltas_for(key, old, new)
    }
    fn tx_claims(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<ClaimOp>> {
        self.tx_claims_for(self.name(), key, old, new)
    }
    fn tx_preplan_projection_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        self.tx_preplan_projection_targets()
    }
    fn tx_preplan_stream_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        TxParticipant::tx_preplan_stream_targets(self.tx_inner())
    }
    fn tx_prepare_streams(&self) -> DbResult<Vec<crate::queue::retention::ProducerPermit>> {
        TxParticipant::tx_prepare_streams(self.tx_inner())
    }
}

impl<K, const V: usize, T, H> TxParticipant for crate::zero_map::ZeroMap<K, V, T, H, Bitcask>
where
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    T: armour_core::GetType
        + crate::txn::participant::TxCollectionName
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Send
        + Sync
        + 'static,
    H: crate::hook::TypedWriteHook<K, T> + 'static,
{
    fn name(&self) -> &'static str {
        T::tx_collection_name()
    }
    fn backend(&self) -> TxBackend {
        TxBackend::Bitcask
    }
    fn schema_fingerprint(&self) -> u64 {
        T::TYPE.h()
    }
    fn encrypted(&self) -> bool {
        TxParticipant::encrypted(self.tx_inner())
    }
    fn max_entry_bytes(&self) -> Option<usize> {
        TxParticipant::max_entry_bytes(self.tx_inner())
    }
    fn shard_of(&self, key: &[u8]) -> DbResult<usize> {
        TxParticipant::shard_of(self.tx_inner(), key)
    }
    fn shard_count(&self) -> usize {
        TxParticipant::shard_count(self.tx_inner())
    }
    fn lock_shards<'a>(&'a self, shards: &[usize]) -> DbResult<Box<dyn TxLocked + 'a>> {
        TxParticipant::lock_shards(self.tx_inner(), shards)
    }
    fn tx_project(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
        self.tx_project_writes(key, old, new)
    }
    fn tx_stream_writes(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<crate::txn::participant::TxStreamWrites> {
        TxParticipant::tx_stream_writes(self.tx_inner(), key, old, new)
    }
    fn tx_counter_deltas(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
        self.tx_counter_deltas_for(key, old, new)
    }
    fn tx_claims(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<ClaimOp>> {
        self.tx_claims_for(self.name(), key, old, new)
    }
    fn tx_preplan_projection_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        self.tx_preplan_projection_targets()
    }
    fn tx_preplan_stream_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        TxParticipant::tx_preplan_stream_targets(self.tx_inner())
    }
    fn tx_prepare_streams(&self) -> DbResult<Vec<crate::queue::retention::ProducerPermit>> {
        TxParticipant::tx_prepare_streams(self.tx_inner())
    }
}

impl<K, const V: usize, T, H> TxTarget for crate::zero_tree::ZeroTree<K, V, T, H, Bitcask>
where
    K: Key + 'static,
    T: Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Send
        + Sync
        + 'static,
    H: crate::hook::TypedWriteHook<K, T> + 'static,
{
    type Key = K;
    type Val = T;

    fn tx_encode(&self, value: &T, out: &mut Vec<u8>) -> DbResult<()> {
        out.extend_from_slice(crate::zero_tree::to_bytes::<V, T>(value).as_slice());
        Ok(())
    }

    fn tx_decode(&self, bytes: &[u8]) -> DbResult<T> {
        let fixed: [u8; V] = bytes
            .try_into()
            .map_err(|_| DbError::Client("transaction value has the wrong fixed length"))?;
        Ok(crate::zero_tree::from_value_bytes::<V, T>(&fixed))
    }
}

impl<K, const V: usize, T, H> TxTarget for crate::zero_map::ZeroMap<K, V, T, H, Bitcask>
where
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    T: Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Send
        + Sync
        + 'static,
    H: crate::hook::TypedWriteHook<K, T> + 'static,
{
    type Key = K;
    type Val = T;

    fn tx_encode(&self, value: &T, out: &mut Vec<u8>) -> DbResult<()> {
        out.extend_from_slice(crate::zero_tree::to_bytes::<V, T>(value).as_slice());
        Ok(())
    }

    fn tx_decode(&self, bytes: &[u8]) -> DbResult<T> {
        let fixed: [u8; V] = bytes
            .try_into()
            .map_err(|_| DbError::Client("transaction value has the wrong fixed length"))?;
        Ok(crate::zero_tree::from_value_bytes::<V, T>(&fixed))
    }
}

impl<K: Key + 'static, const V: usize, H: WriteHook<K> + 'static> TxFamily
    for ConstTree<K, V, H, Fixed>
{
    type Shard<'a>
        = ConstShard<'a, K, V, H, Fixed>
    where
        Self: 'a;
    type Deferred<'a>
        = ConstDeferred<K, V, H, Fixed>
    where
        Self: 'a;
    type ReleaseExtra = ();

    fn tx_shard_count(&self) -> usize {
        self.shard_count()
    }
    fn tx_shard_of(&self, key: &[u8]) -> usize {
        self.shard_for(&K::from_bytes(key))
    }
    fn tx_make_shard(&self, sid: usize) -> Self::Shard<'_> {
        self.tx_shard(sid)
    }
    fn tx_release(shard: Self::Shard<'_>) -> (Vec<<Self as Family>::Owned>, WatchWakeBatch, ()) {
        let (events, wakes, guard) = Self::tx_release_shard(shard);
        drop(guard);
        (events, wakes, ())
    }
    fn tx_deferred(
        &self,
        events: Vec<<Self as Family>::Owned>,
        (): (),
    ) -> DbResult<Box<dyn DeferredReplay>> {
        Ok(Box::new(ConstDeferred {
            tree: self.tx_arc()?,
            events,
        }))
    }
}

impl<K: Key + 'static, const V: usize, H: WriteHook<K> + 'static> TxRaw
    for ConstTree<K, V, H, Fixed>
{
    fn raw_read(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
        Ok(shard.try_get(&K::from_bytes(key))?.map(|v| v.to_vec()))
    }
    fn raw_contains(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<bool> {
        shard.try_contains(&K::from_bytes(key))
    }
    fn raw_put(&self, shard: &mut Self::Shard<'_>, key: &[u8], value: &[u8]) -> DbResult<()> {
        let bytes: [u8; V] = value
            .try_into()
            .map_err(|_| DbError::Client("transaction value has the wrong fixed length"))?;
        shard.put(&K::from_bytes(key), &bytes).map(drop)
    }
    fn raw_delete(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<()> {
        shard.delete(&K::from_bytes(key)).map(drop)
    }

    fn reserve_in_shard(
        &self,
        shard: &mut Self::Shard<'_>,
        ops: &[PlannedOp<'_>],
    ) -> DbResult<Vec<VersionTag>> {
        reserve_fixed_in_shard(
            shard,
            ops,
            |shard, key, delete| self.tx_reserve_fixed(shard, key, delete),
            Self::release_reservation,
        )
    }

    fn arm_tag(shard: &mut Self::Shard<'_>, tag: VersionTag) -> DbResult<()> {
        let VersionTag::Fixed {
            slot_id,
            target_meta,
        } = tag
        else {
            return Err(DbError::Internal(
                "Bitcask version tag on a FixedStore participant",
            ));
        };
        if slot_id == crate::fixed::slot::NO_SLOT {
            return Ok(());
        }
        let fresh = Self::tx_fixed_slot_is_free(shard, slot_id)?;
        Self::tx_arm_slot(shard, slot_id, target_meta, fresh);
        Ok(())
    }

    fn disarm_tag(shard: &mut Self::Shard<'_>) {
        Self::tx_clear_armed_slot(shard);
    }

    fn release_reservation(shard: &mut Self::Shard<'_>, tag: VersionTag) {
        if let VersionTag::Fixed { slot_id, .. } = tag
            && slot_id != crate::fixed::slot::NO_SLOT
        {
            let fresh = Self::tx_fixed_slot_is_free(shard, slot_id).unwrap_or(false);
            Self::tx_release_slot(shard, slot_id, fresh);
        }
    }
    fn fence_shard(_shard: &mut Self::Shard<'_>, _tag: VersionTag) -> Option<u64> {
        // FixedStore replay reads the reserved slot's exact state, which no
        // later write can disturb into a false "already applied".
        None
    }

    fn buffered_events(shard: &Self::Shard<'_>) -> usize {
        shard.tx_buffered_events()
    }

    fn needs_replay(
        &self,
        shard: &mut Self::Shard<'_>,
        key: &[u8],
        kind: OpKind,
        tag: VersionTag,
    ) -> DbResult<bool> {
        use crate::fixed::slot::{NO_SLOT, STATUS_DELETED, is_newer, status_of, version_of};

        let VersionTag::Fixed {
            slot_id,
            target_meta,
        } = tag
        else {
            return Err(DbError::Internal(
                "Bitcask version tag on a FixedStore participant",
            ));
        };
        if slot_id == NO_SLOT {
            return Ok(false); // a logged no-op has nothing to apply
        }
        // A failed apply can leave this reservation allocated while a later
        // commit for the same key lands in a different slot. Replaying into the
        // orphaned slot would overwrite that newer incarnation.
        let superseded_elsewhere = matches!(
            self.tx_fixed_index_slot(shard, key),
            Some(other) if other != slot_id
        );
        let state = Self::tx_fixed_slot_state(shard, slot_id)?;
        let current = state.meta;

        // Already exactly the committed state?
        if current == target_meta {
            if let OpKind::Put = kind
                && state.crc_valid
                && state.key.as_deref() != Some(key)
            {
                // The slot holds a *different* intact row at our exact target
                // version: the reservation was recycled while the record stayed
                // durable. Replaying would overwrite a live row that is not ours.
                if superseded_elsewhere {
                    return Ok(false); // a newer incarnation of our key won
                }
                return Err(DbError::TxFixedSlot {
                    collection: TxParticipant::name(self),
                    slot: slot_id,
                    current,
                    target: target_meta,
                });
            }
            let settled = match kind {
                OpKind::Put => state.crc_valid,
                OpKind::Delete => status_of(current) == STATUS_DELETED,
            };
            if settled {
                return Ok(false);
            }
            // Same version, different reality: a torn write recovers as FREE with
            // its version preserved. Replay it — unless the key moved elsewhere.
            return Ok(!superseded_elsewhere);
        }

        let (cur_v, tgt_v) = (version_of(current), version_of(target_meta));
        if is_newer(cur_v, tgt_v) {
            return Ok(false); // the slot moved on; our write is superseded
        }
        if is_newer(tgt_v, cur_v) {
            return Ok(!superseded_elsewhere); // our write never landed
        }
        if cur_v == tgt_v {
            return Ok(!superseded_elsewhere); // equal version, different status/CRC
        }
        if superseded_elsewhere {
            return Ok(false);
        }
        Err(DbError::TxFixedSlot {
            collection: TxParticipant::name(self),
            slot: slot_id,
            current,
            target: target_meta,
        })
    }
}

impl<K: Key + 'static, const V: usize, H: WriteHook<K> + 'static> TxParticipant
    for ConstTree<K, V, H, Fixed>
{
    fn name(&self) -> &'static str {
        self.name()
    }
    fn backend(&self) -> TxBackend {
        TxBackend::Fixed
    }
    fn schema_fingerprint(&self) -> u64 {
        self.typ_hash()
    }
    fn encrypted(&self) -> bool {
        self.tx_encrypted()
    }
    fn max_entry_bytes(&self) -> Option<usize> {
        None
    }
    fn shard_of(&self, key: &[u8]) -> DbResult<usize> {
        Ok(<Self as TxFamily>::tx_shard_of(self, key))
    }
    fn shard_count(&self) -> usize {
        <Self as TxFamily>::tx_shard_count(self)
    }
    fn lock_shards<'a>(&'a self, shards: &[usize]) -> DbResult<Box<dyn TxLocked + 'a>> {
        Ok(Box::new(TxHeld::lock(self, shards)))
    }
    fn tx_project(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
        self.tx_project_writes(key, old, new)
    }
    fn tx_stream_writes(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<crate::txn::participant::TxStreamWrites> {
        ConstTree::tx_stream_writes(self, key, old, new)
    }
    fn tx_counter_deltas(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
        self.tx_counter_deltas_for(key, old, new)
    }
    fn tx_claims(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<ClaimOp>> {
        self.tx_claims_for(self.name(), key, old, new)
    }
    fn tx_preplan_projection_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        self.tx_preplan_projection_targets()
    }
    fn tx_preplan_stream_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        Ok(ConstTree::tx_preplan_stream_targets(self))
    }
    fn tx_prepare_streams(&self) -> DbResult<Vec<crate::queue::retention::ProducerPermit>> {
        ConstTree::tx_prepare_streams(self)
    }
}

impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    H: WriteHook<K> + 'static,
> TxFamily for ConstMap<K, V, H, Fixed>
{
    type Shard<'a>
        = ConstMapShard<'a, K, V, H, Fixed>
    where
        Self: 'a;
    type Deferred<'a>
        = ConstMapDeferred<K, V, H, Fixed>
    where
        Self: 'a;
    type ReleaseExtra = ();

    fn tx_shard_count(&self) -> usize {
        self.shard_count()
    }
    fn tx_shard_of(&self, key: &[u8]) -> usize {
        self.shard_for(&K::from_bytes(key))
    }
    fn tx_make_shard(&self, sid: usize) -> Self::Shard<'_> {
        self.tx_shard(sid)
    }
    fn tx_release(shard: Self::Shard<'_>) -> (Vec<<Self as Family>::Owned>, WatchWakeBatch, ()) {
        let (events, wakes) = Self::tx_release_shard(shard);
        (events, wakes, ())
    }
    fn tx_deferred(
        &self,
        events: Vec<<Self as Family>::Owned>,
        (): (),
    ) -> DbResult<Box<dyn DeferredReplay>> {
        Ok(Box::new(ConstMapDeferred {
            map: self.tx_arc()?,
            events,
        }))
    }
}

impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    H: WriteHook<K> + 'static,
> TxRaw for ConstMap<K, V, H, Fixed>
{
    fn raw_read(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
        Ok(shard.try_get(&K::from_bytes(key))?.map(|v| v.to_vec()))
    }
    fn raw_contains(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<bool> {
        shard.try_contains(&K::from_bytes(key))
    }
    fn raw_put(&self, shard: &mut Self::Shard<'_>, key: &[u8], value: &[u8]) -> DbResult<()> {
        let bytes: [u8; V] = value
            .try_into()
            .map_err(|_| DbError::Client("transaction value has the wrong fixed length"))?;
        shard.put(&K::from_bytes(key), &bytes).map(drop)
    }
    fn raw_delete(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<()> {
        shard.delete(&K::from_bytes(key)).map(drop)
    }

    fn reserve_in_shard(
        &self,
        shard: &mut Self::Shard<'_>,
        ops: &[PlannedOp<'_>],
    ) -> DbResult<Vec<VersionTag>> {
        reserve_fixed_in_shard(
            shard,
            ops,
            |shard, key, delete| self.tx_reserve_fixed(shard, key, delete),
            Self::release_reservation,
        )
    }

    fn arm_tag(shard: &mut Self::Shard<'_>, tag: VersionTag) -> DbResult<()> {
        let VersionTag::Fixed {
            slot_id,
            target_meta,
        } = tag
        else {
            return Err(DbError::Internal(
                "Bitcask version tag on a FixedStore participant",
            ));
        };
        if slot_id == crate::fixed::slot::NO_SLOT {
            return Ok(());
        }
        let fresh = Self::tx_fixed_slot_is_free(shard, slot_id)?;
        Self::tx_arm_slot(shard, slot_id, target_meta, fresh);
        Ok(())
    }

    fn disarm_tag(shard: &mut Self::Shard<'_>) {
        Self::tx_clear_armed_slot(shard);
    }

    fn release_reservation(shard: &mut Self::Shard<'_>, tag: VersionTag) {
        if let VersionTag::Fixed { slot_id, .. } = tag
            && slot_id != crate::fixed::slot::NO_SLOT
        {
            let fresh = Self::tx_fixed_slot_is_free(shard, slot_id).unwrap_or(false);
            Self::tx_release_slot(shard, slot_id, fresh);
        }
    }
    fn fence_shard(_shard: &mut Self::Shard<'_>, _tag: VersionTag) -> Option<u64> {
        // FixedStore replay reads the reserved slot's exact state, which no
        // later write can disturb into a false "already applied".
        None
    }

    fn buffered_events(shard: &Self::Shard<'_>) -> usize {
        shard.tx_buffered_events()
    }

    fn needs_replay(
        &self,
        shard: &mut Self::Shard<'_>,
        key: &[u8],
        kind: OpKind,
        tag: VersionTag,
    ) -> DbResult<bool> {
        use crate::fixed::slot::{NO_SLOT, STATUS_DELETED, is_newer, status_of, version_of};

        let VersionTag::Fixed {
            slot_id,
            target_meta,
        } = tag
        else {
            return Err(DbError::Internal(
                "Bitcask version tag on a FixedStore participant",
            ));
        };
        if slot_id == NO_SLOT {
            return Ok(false);
        }
        let superseded_elsewhere = matches!(
            Self::tx_fixed_index_slot(shard, key),
            Some(other) if other != slot_id
        );
        let state = Self::tx_fixed_slot_state(shard, slot_id)?;
        let current = state.meta;

        if current == target_meta {
            if let OpKind::Put = kind
                && state.crc_valid
                && state.key.as_deref() != Some(key)
            {
                // The slot holds a *different* intact row at our exact target
                // version: the reservation was recycled while the record stayed
                // durable. Replaying would overwrite a live row that is not ours.
                if superseded_elsewhere {
                    return Ok(false); // a newer incarnation of our key won
                }
                return Err(DbError::TxFixedSlot {
                    collection: TxParticipant::name(self),
                    slot: slot_id,
                    current,
                    target: target_meta,
                });
            }
            let settled = match kind {
                OpKind::Put => state.crc_valid,
                OpKind::Delete => status_of(current) == STATUS_DELETED,
            };
            if settled {
                return Ok(false);
            }
            return Ok(!superseded_elsewhere);
        }

        let (cur_v, tgt_v) = (version_of(current), version_of(target_meta));
        if is_newer(cur_v, tgt_v) {
            return Ok(false);
        }
        if is_newer(tgt_v, cur_v) {
            return Ok(!superseded_elsewhere);
        }
        if cur_v == tgt_v {
            return Ok(!superseded_elsewhere);
        }
        if superseded_elsewhere {
            return Ok(false);
        }
        Err(DbError::TxFixedSlot {
            collection: TxParticipant::name(self),
            slot: slot_id,
            current,
            target: target_meta,
        })
    }
}

impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    H: WriteHook<K> + 'static,
> TxParticipant for ConstMap<K, V, H, Fixed>
{
    fn name(&self) -> &'static str {
        self.name()
    }
    fn backend(&self) -> TxBackend {
        TxBackend::Fixed
    }
    fn schema_fingerprint(&self) -> u64 {
        self.typ_hash()
    }
    fn encrypted(&self) -> bool {
        self.tx_encrypted()
    }
    fn max_entry_bytes(&self) -> Option<usize> {
        None
    }
    fn shard_of(&self, key: &[u8]) -> DbResult<usize> {
        Ok(<Self as TxFamily>::tx_shard_of(self, key))
    }
    fn lock_shards<'a>(&'a self, shards: &[usize]) -> DbResult<Box<dyn TxLocked + 'a>> {
        Ok(Box::new(TxHeld::lock(self, shards)))
    }
    fn tx_project(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
        self.tx_project_writes(key, old, new)
    }
    fn tx_stream_writes(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<crate::txn::participant::TxStreamWrites> {
        ConstMap::tx_stream_writes(self, key, old, new)
    }
    fn tx_counter_deltas(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
        self.tx_counter_deltas_for(key, old, new)
    }
    fn tx_claims(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<ClaimOp>> {
        self.tx_claims_for(self.name(), key, old, new)
    }
    fn tx_preplan_projection_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        self.tx_preplan_projection_targets()
    }
    fn tx_preplan_stream_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        Ok(ConstMap::tx_preplan_stream_targets(self))
    }
    fn tx_prepare_streams(&self) -> DbResult<Vec<crate::queue::retention::ProducerPermit>> {
        ConstMap::tx_prepare_streams(self)
    }
}

impl<K, const V: usize, T, H> TxParticipant for crate::zero_tree::ZeroTree<K, V, T, H, Fixed>
where
    K: Key + 'static,
    T: armour_core::GetType
        + crate::txn::participant::TxCollectionName
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Send
        + Sync
        + 'static,
    H: crate::hook::TypedWriteHook<K, T> + 'static,
{
    fn name(&self) -> &'static str {
        T::tx_collection_name()
    }
    fn backend(&self) -> TxBackend {
        TxBackend::Fixed
    }
    fn schema_fingerprint(&self) -> u64 {
        T::TYPE.h()
    }
    fn encrypted(&self) -> bool {
        TxParticipant::encrypted(self.tx_inner())
    }
    fn max_entry_bytes(&self) -> Option<usize> {
        TxParticipant::max_entry_bytes(self.tx_inner())
    }
    fn shard_of(&self, key: &[u8]) -> DbResult<usize> {
        TxParticipant::shard_of(self.tx_inner(), key)
    }
    fn shard_count(&self) -> usize {
        TxParticipant::shard_count(self.tx_inner())
    }
    fn lock_shards<'a>(&'a self, shards: &[usize]) -> DbResult<Box<dyn TxLocked + 'a>> {
        TxParticipant::lock_shards(self.tx_inner(), shards)
    }
    fn tx_project(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
        self.tx_project_writes(key, old, new)
    }
    fn tx_stream_writes(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<crate::txn::participant::TxStreamWrites> {
        TxParticipant::tx_stream_writes(self.tx_inner(), key, old, new)
    }
    fn tx_counter_deltas(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
        self.tx_counter_deltas_for(key, old, new)
    }
    fn tx_claims(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<ClaimOp>> {
        self.tx_claims_for(self.name(), key, old, new)
    }
    fn tx_preplan_projection_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        self.tx_preplan_projection_targets()
    }
    fn tx_preplan_stream_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        TxParticipant::tx_preplan_stream_targets(self.tx_inner())
    }
    fn tx_prepare_streams(&self) -> DbResult<Vec<crate::queue::retention::ProducerPermit>> {
        TxParticipant::tx_prepare_streams(self.tx_inner())
    }
}

impl<K, const V: usize, T, H> TxParticipant for crate::zero_map::ZeroMap<K, V, T, H, Fixed>
where
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    T: armour_core::GetType
        + crate::txn::participant::TxCollectionName
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Send
        + Sync
        + 'static,
    H: crate::hook::TypedWriteHook<K, T> + 'static,
{
    fn name(&self) -> &'static str {
        T::tx_collection_name()
    }
    fn backend(&self) -> TxBackend {
        TxBackend::Fixed
    }
    fn schema_fingerprint(&self) -> u64 {
        T::TYPE.h()
    }
    fn encrypted(&self) -> bool {
        TxParticipant::encrypted(self.tx_inner())
    }
    fn max_entry_bytes(&self) -> Option<usize> {
        TxParticipant::max_entry_bytes(self.tx_inner())
    }
    fn shard_of(&self, key: &[u8]) -> DbResult<usize> {
        TxParticipant::shard_of(self.tx_inner(), key)
    }
    fn shard_count(&self) -> usize {
        TxParticipant::shard_count(self.tx_inner())
    }
    fn lock_shards<'a>(&'a self, shards: &[usize]) -> DbResult<Box<dyn TxLocked + 'a>> {
        TxParticipant::lock_shards(self.tx_inner(), shards)
    }
    fn tx_project(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
        self.tx_project_writes(key, old, new)
    }
    fn tx_stream_writes(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<crate::txn::participant::TxStreamWrites> {
        TxParticipant::tx_stream_writes(self.tx_inner(), key, old, new)
    }
    fn tx_counter_deltas(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
        self.tx_counter_deltas_for(key, old, new)
    }
    fn tx_claims(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<ClaimOp>> {
        self.tx_claims_for(self.name(), key, old, new)
    }
    fn tx_preplan_projection_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        self.tx_preplan_projection_targets()
    }
    fn tx_preplan_stream_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        TxParticipant::tx_preplan_stream_targets(self.tx_inner())
    }
    fn tx_prepare_streams(&self) -> DbResult<Vec<crate::queue::retention::ProducerPermit>> {
        TxParticipant::tx_prepare_streams(self.tx_inner())
    }
}

impl<K, const V: usize, T, H> TxTarget for crate::zero_tree::ZeroTree<K, V, T, H, Fixed>
where
    K: Key + 'static,
    T: Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Send
        + Sync
        + 'static,
    H: crate::hook::TypedWriteHook<K, T> + 'static,
{
    type Key = K;
    type Val = T;

    fn tx_encode(&self, value: &T, out: &mut Vec<u8>) -> DbResult<()> {
        out.extend_from_slice(crate::zero_tree::to_bytes::<V, T>(value).as_slice());
        Ok(())
    }

    fn tx_decode(&self, bytes: &[u8]) -> DbResult<T> {
        let fixed: [u8; V] = bytes
            .try_into()
            .map_err(|_| DbError::Client("transaction value has the wrong fixed length"))?;
        Ok(crate::zero_tree::from_value_bytes::<V, T>(&fixed))
    }
}

impl<K, const V: usize, T, H> TxTarget for crate::zero_map::ZeroMap<K, V, T, H, Fixed>
where
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    T: Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Send
        + Sync
        + 'static,
    H: crate::hook::TypedWriteHook<K, T> + 'static,
{
    type Key = K;
    type Val = T;

    fn tx_encode(&self, value: &T, out: &mut Vec<u8>) -> DbResult<()> {
        out.extend_from_slice(crate::zero_tree::to_bytes::<V, T>(value).as_slice());
        Ok(())
    }

    fn tx_decode(&self, bytes: &[u8]) -> DbResult<T> {
        let fixed: [u8; V] = bytes
            .try_into()
            .map_err(|_| DbError::Client("transaction value has the wrong fixed length"))?;
        Ok(crate::zero_tree::from_value_bytes::<V, T>(&fixed))
    }
}

#[cfg(test)]
#[path = "fixed_replay_tests.rs"]
mod fixed_replay_tests;

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    use crate::txn::participant::{PlannedOp, TxBackend, TxParticipant};
    use crate::{Config, WriteHook};

    #[derive(Clone, Default)]
    pub(super) struct CountingHook(Arc<AtomicUsize>);
    impl WriteHook<[u8; 8]> for CountingHook {
        const NEEDS_WRITE: bool = true;
        fn on_write(&self, _key: &[u8; 8], _old: Option<&[u8]>, _new: Option<&[u8]>) {
            self.0.fetch_add(1, Ordering::SeqCst);
        }
    }

    fn open_tree(
        dir: &std::path::Path,
        hook: CountingHook,
    ) -> Arc<crate::ConstTree<[u8; 8], 16, CountingHook>> {
        let tree = Arc::new(crate::ConstTree::open_hooked(dir, Config::test(), hook).unwrap());
        crate::ConstTree::bind_tx_self(&tree);
        tree
    }

    #[test]
    fn tx_const_locked_reads_reserves_applies_and_defers_hooks() {
        let dir = tempfile::tempdir().unwrap();
        let fired = Arc::new(AtomicUsize::new(0));
        let tree = open_tree(dir.path(), CountingHook(fired.clone()));
        tree.put(&[1u8; 8], &[9u8; 16]).unwrap();
        fired.store(0, Ordering::SeqCst);

        assert_eq!(tree.backend(), TxBackend::Bitcask);
        let sid = tree.shard_of(&[1u8; 8]).unwrap();
        let mut locked = tree.lock_shards(&[sid]).unwrap();

        assert_eq!(
            locked.read(&[1u8; 8]).unwrap().as_deref(),
            Some(&[9u8; 16][..])
        );
        assert!(locked.read(&[2u8; 8]).unwrap().is_none());

        let tags = locked
            .reserve(&[PlannedOp {
                key: &[1u8; 8],
                delete: false,
            }])
            .unwrap();
        locked.apply_put(&[1u8; 8], &[7u8; 16], tags[0]).unwrap();
        assert_eq!(
            fired.load(Ordering::SeqCst),
            0,
            "hooks must not fire under the lock"
        );

        let mut deferred = locked
            .finish(
                &mut Vec::new(),
                &mut crate::watch::WatchWakeBatch::default(),
            )
            .unwrap();
        assert_eq!(deferred.event_count(), 1);
        deferred.replay_one(0);
        assert_eq!(
            fired.load(Ordering::SeqCst),
            1,
            "the hook fires after release"
        );

        assert_eq!(tree.get(&[1u8; 8]), Some([7u8; 16]));
    }

    #[test]
    fn tx_const_reserved_gsn_is_the_one_that_lands() {
        let dir = tempfile::tempdir().unwrap();
        let tree = open_tree(dir.path(), CountingHook::default());
        let sid = tree.shard_of(&[3u8; 8]).unwrap();
        let mut locked = tree.lock_shards(&[sid]).unwrap();
        let tags = locked
            .reserve(&[PlannedOp {
                key: &[3u8; 8],
                delete: false,
            }])
            .unwrap();
        let crate::txlog::record::VersionTag::Bitcask { gsn } = tags[0] else {
            panic!("Bitcask participant must reserve a Bitcask tag");
        };
        locked.apply_put(&[3u8; 8], &[1u8; 16], tags[0]).unwrap();
        drop(
            locked
                .finish(
                    &mut Vec::new(),
                    &mut crate::watch::WatchWakeBatch::default(),
                )
                .unwrap(),
        );
        assert_eq!(tree.get(&[3u8; 8]), Some([1u8; 16]));

        let mut again = tree.lock_shards(&[sid]).unwrap();
        let next = again
            .reserve(&[PlannedOp {
                key: &[3u8; 8],
                delete: false,
            }])
            .unwrap();
        let crate::txlog::record::VersionTag::Bitcask { gsn: gsn2 } = next[0] else {
            unreachable!("Bitcask participant")
        };
        assert!(gsn2 > gsn, "reserved GSNs must not repeat: {gsn2} <= {gsn}");
        again.abort();
    }

    #[test]
    fn tx_const_failed_apply_clears_armed_gsn() {
        let dir = tempfile::tempdir().unwrap();
        let tree = open_tree(dir.path(), CountingHook::default());
        let sid = tree.shard_of(&[5u8; 8]).unwrap();
        let mut locked = tree.lock_shards(&[sid]).unwrap();
        let tags = locked
            .reserve(&[PlannedOp {
                key: &[5u8; 8],
                delete: false,
            }])
            .unwrap();
        assert!(
            locked.apply_put(&[5u8; 8], &[1u8; 8], tags[0]).is_err(),
            "wrong-length value must fail apply"
        );

        let tags2 = locked
            .reserve(&[PlannedOp {
                key: &[5u8; 8],
                delete: false,
            }])
            .unwrap();
        let crate::txlog::record::VersionTag::Bitcask { gsn } = tags2[0] else {
            unreachable!("Bitcask participant");
        };
        locked.apply_put(&[5u8; 8], &[2u8; 16], tags2[0]).unwrap();
        drop(
            locked
                .finish(
                    &mut Vec::new(),
                    &mut crate::watch::WatchWakeBatch::default(),
                )
                .unwrap(),
        );
        assert_eq!(tree.get(&[5u8; 8]), Some([2u8; 16]));

        let mut again = tree.lock_shards(&[sid]).unwrap();
        let next = again
            .reserve(&[PlannedOp {
                key: &[5u8; 8],
                delete: false,
            }])
            .unwrap();
        let crate::txlog::record::VersionTag::Bitcask { gsn: gsn2 } = next[0] else {
            unreachable!("Bitcask participant");
        };
        assert!(
            gsn2 > gsn,
            "a failed apply must not leave an armed GSN: {gsn2} <= {gsn}"
        );
        again.abort();
    }

    #[test]
    fn tx_const_abort_writes_nothing() {
        let dir = tempfile::tempdir().unwrap();
        let tree = open_tree(dir.path(), CountingHook::default());
        let sid = tree.shard_of(&[4u8; 8]).unwrap();
        let mut locked = tree.lock_shards(&[sid]).unwrap();
        let _tags = locked
            .reserve(&[PlannedOp {
                key: &[4u8; 8],
                delete: false,
            }])
            .unwrap();
        locked.abort();

        assert_eq!(tree.get(&[4u8; 8]), None, "a reservation is not a write");
    }

    #[test]
    fn tx_const_abort_releases_reservations_before_release() {
        // Bitcask has nothing to release, but the *call* must happen for every
        // unapplied reservation — Task 4's Fixed impl relies on it. Applying one
        // of two reservations must leave exactly one to release.
        let dir = tempfile::tempdir().unwrap();
        let tree = open_tree(dir.path(), CountingHook::default());
        let sid = tree.shard_of(&[6u8; 8]).unwrap();
        let mut locked = tree.lock_shards(&[sid]).unwrap();
        let tags = locked
            .reserve(&[
                PlannedOp {
                    key: &[6u8; 8],
                    delete: false,
                },
                PlannedOp {
                    key: &[6u8; 8],
                    delete: false,
                },
            ])
            .unwrap();
        locked.apply_put(&[6u8; 8], &[1u8; 16], tags[0]).unwrap();
        assert_eq!(
            locked.unapplied_reservations(),
            1,
            "an applied tag is no longer a live reservation"
        );
        locked.abort();
        assert_eq!(tree.get(&[6u8; 8]), Some([1u8; 16]));
    }

    pub(super) fn open_fixed_tree(
        dir: &std::path::Path,
        hook: CountingHook,
    ) -> Arc<crate::ConstTree<[u8; 8], 16, CountingHook, crate::durability::Fixed>> {
        let cfg = crate::fixed::config::FixedConfig {
            shard_count: 1,
            grow_step: 16,
            ..crate::fixed::config::FixedConfig::test()
        };
        let tree = Arc::new(crate::ConstTree::open_with_hook(dir, cfg, hook).unwrap());
        crate::ConstTree::bind_tx_self(&tree);
        tree
    }

    #[test]
    fn tx_fixed_locked_reserves_the_slot_the_apply_writes() {
        use crate::fixed::slot::{STATUS_OCCUPIED, status_of, version_of};

        let dir = tempfile::tempdir().unwrap();
        let fired = Arc::new(AtomicUsize::new(0));
        let tree = open_fixed_tree(dir.path(), CountingHook(fired.clone()));
        tree.put(&[1u8; 8], &[9u8; 16]).unwrap(); // slot 0, version 1
        fired.store(0, Ordering::SeqCst);

        assert_eq!(tree.backend(), TxBackend::Fixed);
        let sid = tree.shard_of(&[1u8; 8]).unwrap();
        let mut locked = tree.lock_shards(&[sid]).unwrap();

        assert_eq!(
            locked.read(&[1u8; 8]).unwrap().as_deref(),
            Some(&[9u8; 16][..])
        );

        let tags = locked
            .reserve(&[PlannedOp {
                key: &[1u8; 8],
                delete: false,
            }])
            .unwrap();
        let crate::txlog::record::VersionTag::Fixed {
            slot_id,
            target_meta,
        } = tags[0]
        else {
            panic!("a Fixed participant must reserve a Fixed tag");
        };
        assert_eq!(status_of(target_meta), STATUS_OCCUPIED);
        assert_eq!(
            version_of(target_meta),
            2,
            "reserved from the slot's current version"
        );

        locked.apply_put(&[1u8; 8], &[7u8; 16], tags[0]).unwrap();
        assert_eq!(
            fired.load(Ordering::SeqCst),
            0,
            "hooks must not fire under the lock"
        );
        let mut deferred = locked
            .finish(
                &mut Vec::new(),
                &mut crate::watch::WatchWakeBatch::default(),
            )
            .unwrap();
        assert_eq!(deferred.event_count(), 1);
        deferred.replay_one(0);
        assert_eq!(fired.load(Ordering::SeqCst), 1);

        assert_eq!(tree.get(&[1u8; 8]), Some([7u8; 16]));
        // And the slot carries exactly the meta that was logged.
        let state = tree.tx_debug_slot_state(sid, slot_id).unwrap();
        assert_eq!(state.meta, target_meta);
        assert!(state.crc_valid);
    }

    #[test]
    fn tx_fixed_abort_frees_a_freshly_reserved_slot() {
        let dir = tempfile::tempdir().unwrap();
        let tree = open_fixed_tree(dir.path(), CountingHook::default());
        let sid = tree.shard_of(&[2u8; 8]).unwrap();

        let mut locked = tree.lock_shards(&[sid]).unwrap();
        let tags = locked
            .reserve(&[PlannedOp {
                key: &[2u8; 8],
                delete: false,
            }])
            .unwrap();
        let crate::txlog::record::VersionTag::Fixed { slot_id, .. } = tags[0] else {
            unreachable!("Fixed participant")
        };
        locked.abort();

        assert_eq!(tree.get(&[2u8; 8]), None, "a reservation is not a write");
        let state = tree.tx_debug_slot_state(sid, slot_id).unwrap();
        assert_eq!(
            crate::fixed::slot::status_of(state.meta),
            crate::fixed::slot::STATUS_FREE,
            "the released slot is untouched and free"
        );

        // The same slot is handed out again, proving the allocator took it back.
        let mut again = tree.lock_shards(&[sid]).unwrap();
        let tags = again
            .reserve(&[PlannedOp {
                key: &[3u8; 8],
                delete: false,
            }])
            .unwrap();
        let crate::txlog::record::VersionTag::Fixed {
            slot_id: reused, ..
        } = tags[0]
        else {
            unreachable!("Fixed participant")
        };
        assert_eq!(reused, slot_id);
        again.abort();
    }

    /// Past the log point the record is durable and replay owns the reserved
    /// slot. Handing it back to the allocator lets an unrelated key take it, so
    /// replay would either overwrite that key's row or be refused.
    #[test]
    fn tx_fixed_reservation_survives_an_abort_past_the_log_point() {
        let dir = tempfile::tempdir().unwrap();
        let tree = open_fixed_tree(dir.path(), CountingHook::default());
        let key = [12u8; 8];
        let sid = tree.shard_of(&key).unwrap();

        let mut locked = tree.lock_shards(&[sid]).unwrap();
        let tags = locked
            .reserve(&[PlannedOp {
                key: &key,
                delete: false,
            }])
            .unwrap();
        let crate::txlog::record::VersionTag::Fixed { slot_id, .. } = tags[0] else {
            unreachable!("Fixed participant")
        };
        locked.mark_logged();
        locked.abort();

        let mut again = tree.lock_shards(&[sid]).unwrap();
        let next = again
            .reserve(&[PlannedOp {
                key: &[13u8; 8],
                delete: false,
            }])
            .unwrap();
        let crate::txlog::record::VersionTag::Fixed {
            slot_id: other_slot,
            ..
        } = next[0]
        else {
            unreachable!("Fixed participant")
        };
        assert_ne!(
            other_slot, slot_id,
            "a reservation past the log point must stay allocated for replay"
        );
        again.abort();
    }

    /// A panic between arming and the mutation must not leave the arm on the
    /// shard: the next unrelated write would take it and land in a transaction's
    /// slot at a transaction's version.
    // Both `parking_lot` and the poison-recovering std adapter continue after a
    // guard-escaping panic. Loom deliberately stops at that boundary.
    #[cfg(not(feature = "loom"))]
    #[test]
    fn tx_fixed_panic_leaves_no_armed_slot_behind() {
        use std::panic::AssertUnwindSafe;

        let dir = tempfile::tempdir().unwrap();
        let tree = open_fixed_tree(dir.path(), CountingHook::default());
        let key = [14u8; 8];
        let sid = tree.shard_of(&key).unwrap();

        let prev = std::panic::take_hook();
        std::panic::set_hook(Box::new(|_| {})); // the panic below is the point
        let outcome = std::panic::catch_unwind(AssertUnwindSafe(|| {
            let mut locked = tree.lock_shards(&[sid]).unwrap();
            let tags = locked
                .reserve(&[PlannedOp {
                    key: &key,
                    delete: false,
                }])
                .unwrap();
            locked.tx_debug_arm(&key, tags[0]).unwrap();
            panic!("the mutation panicked with the shard armed");
        }));
        std::panic::set_hook(prev);
        assert!(outcome.is_err(), "the closure must have panicked");

        assert!(
            tree.tx_debug_armed_slot(sid).is_none(),
            "a panic must not leave a transaction's slot armed on the shard"
        );

        // An unrelated write lands in its own slot at its own version.
        tree.put(&[15u8; 8], &[4u8; 16]).unwrap();
        assert_eq!(tree.get(&[15u8; 8]), Some([4u8; 16]));
    }

    #[test]
    fn tx_fixed_delete_of_an_absent_key_reserves_no_slot() {
        let dir = tempfile::tempdir().unwrap();
        let tree = open_fixed_tree(dir.path(), CountingHook::default());
        let sid = tree.shard_of(&[4u8; 8]).unwrap();
        let mut locked = tree.lock_shards(&[sid]).unwrap();
        let tags = locked
            .reserve(&[PlannedOp {
                key: &[4u8; 8],
                delete: true,
            }])
            .unwrap();
        let crate::txlog::record::VersionTag::Fixed { slot_id, .. } = tags[0] else {
            unreachable!("Fixed participant")
        };
        assert_eq!(
            slot_id,
            crate::fixed::slot::NO_SLOT,
            "there is no slot to delete; the operation is a logged no-op"
        );
        assert_eq!(locked.apply_delete(&[4u8; 8], tags[0]).unwrap(), 0);
        drop(
            locked
                .finish(
                    &mut Vec::new(),
                    &mut crate::watch::WatchWakeBatch::default(),
                )
                .unwrap(),
        );
    }
}