armdb 0.14.2

use crate::Key;
use crate::batch::Applied;
use crate::compaction::{CompactionIndex, compact_shard};
use crate::config::Config;
use crate::ctx::{Ctx, Family, Held, Owned, ShardGuards};
use crate::disk_loc::DiskLoc;
use crate::durability::{Bitcask, Durability, DurabilityInner};
use crate::engine::Engine;
use crate::error::{DbError, DbResult};
use crate::hook::{NoHook, WriteHook};
use crate::key::Location;
use crate::map_index::{Dir, dir_from};
use crate::recovery::recover_const_tree;
use crate::scope::DirRangeIter;
use crate::skiplist::iter::{SkipKeyIter, SkipKeyKeys};
use crate::skiplist::node::{ConstNode, SkipNode, random_height};
use crate::skiplist::{BackFinger, InsertResult, SkipList};
use crate::sync::{self, MutexGuard};
use crate::ttl::{SweepOutcome, TtlSlots};
use crate::upsert::Upsert;
use crate::watch::{WatchEventInner, WatchWakeBatch};
use std::mem::size_of;
use std::ops::Bound;
use std::sync::{Arc, OnceLock, Weak};

#[cfg(test)]
thread_local! {
    static UPDATE_MANY_FAIL_AFTER: std::cell::Cell<usize> = const { std::cell::Cell::new(usize::MAX) };
}

/// Scoped test-only fault for `ConstTree::update_many` applied-prefix tests.
#[cfg(test)]
pub(crate) struct UpdateManyFailAfter(usize);

#[cfg(test)]
impl UpdateManyFailAfter {
    pub(crate) fn after(applied: usize) -> Self {
        let previous = UPDATE_MANY_FAIL_AFTER.with(|remaining| {
            let previous = remaining.get();
            remaining.set(applied);
            previous
        });
        Self(previous)
    }
}

#[cfg(test)]
impl Drop for UpdateManyFailAfter {
    fn drop(&mut self) {
        UPDATE_MANY_FAIL_AFTER.with(|remaining| remaining.set(self.0));
    }
}

enum ConstSweepAction<K: Key> {
    /// `(key, candidate_deadline)` — deadline kept for requeue if delete fails.
    Delete(K, u64),
    Reschedule(K, u64),
    Drop,
    Retry(K),
}

/// A tree with fixed-size keys and values. All values are stored inline in SkipList nodes.
/// Reads never touch disk — zero I/O reads.
///
/// Generic over `D: Durability` (default: [`Bitcask`]). Use [`Fixed`] backend for
/// frequent updates without compaction (`FixedTree` is a type alias for `ConstTree<K, V, Fixed>`).
///
/// # Batch operations
///
/// [`get_many`](Self::get_many) does one sorted finger-walk over the index for a
/// list of keys; [`update_many`](Self::update_many) applies a per-key closure
/// grouped by shard (per-shard isolation, no global atomicity). See
/// `docs/superpowers/specs/26-06-24-armdb-batch-tree-ops.md`.
///
/// Each `ConstTree` owns its storage engine — one tree = one database directory.
///
/// # Write hooks
///
/// Uses [`WriteHook<K>`]. `on_write` fires on `put`/`insert`/`delete`/`cas`/`compare_delete`/`update`/`fetch_update`,
/// and on each mutation inside `atomic()` (replayed after the lock is released, in application order).
/// `on_init` fires once per live entry
/// during `migrate()` or `replay_init()`. Old value is always provided in
/// `on_write` — `NEEDS_OLD_VALUE` is ignored (value is inline, zero cost).
///
/// # Usage
///
/// ```ignore
/// let tree = ConstTree::<16, 64>::open("data/users", Config::balanced().build())?;
/// tree.put(&key, &value)?;
/// tree.close()?;
/// ```
///
/// # Iteration
///
/// `iter()`, `range()`, and `prefix_iter()` all return [`ConstIter`] which
/// implements `Iterator + DoubleEndedIterator` with `Item = (K, [u8; V])`.
/// Lock-free, zero disk I/O.
///
/// ```ignore
/// for (key, value) in tree.iter() { }
/// let latest = tree.prefix_iter(&user_id).take(20).collect::<Vec<_>>();
/// let oldest = tree.iter().rev().take(5);  // DoubleEndedIterator
/// ```
pub struct ConstTree<K: Key, const V: usize, H: WriteHook<K> = NoHook, D: Durability = Bitcask> {
    index: SkipList<ConstNode<K, V, D::Loc>>,
    durability: D,
    shard_prefix_bits: usize,
    reversed: bool,
    pub(crate) hook: H,
    /// Weak back-link for [`tx_deferred`](Self::tx_deferred): set once via
    /// [`bind_tx_self`](Self::bind_tx_self) when the tree lives in an `Arc`.
    tx_self: OnceLock<Weak<Self>>,
    ttl: TtlSlots<K>,
    /// `counters_dirty` marker (T9 Task 7) for the `ZeroTree` this
    /// engine backs, bound once at managed open via
    /// `ManagedOpen::bind_counter_dirty`. Lives here (not on the hook) so a
    /// user-authored wrapper hook that forgets to forward some hook method can
    /// never silently disconnect it — see `counter_dirty.rs`. `ZeroTree`
    /// itself has no separate write path to host this on: its facade/`atomic()`
    /// calls ride this engine's `Family`/`Ctx` seam directly.
    pub(crate) counter_dirty: OnceLock<crate::counter_dirty::CounterDirty>,
}

impl<K: Key, const V: usize, H: WriteHook<K>, D: Durability> ConstTree<K, V, H, D> {
    pub(crate) fn watch_channel(&self) -> Option<&std::sync::Arc<crate::watch::WatchChannel<K>>> {
        self.hook.watch_channel()
    }
    /// Shards holding writes no durability barrier covers yet — the size of
    /// this collection's power-loss window right now.
    ///
    /// Counted shard by shard without a consistent cut, so a concurrent writer
    /// or barrier can move the answer mid-count; read it as a gauge.
    pub fn dirty_shards(&self) -> usize {
        self.durability.dirty_shards()
    }
}

// ==========================================================================
// Bitcask-specific: open, close, compact, sync_hints, config, flush_buffers
// ==========================================================================

impl<K: Key, const V: usize> ConstTree<K, V, NoHook, Bitcask> {
    /// Open or create a `ConstTree` at the given path.
    /// Recovers the index from existing data files on disk.
    ///
    #[doc = include_str!("../docs/exclusive-access.md")]
    pub fn open(path: impl AsRef<std::path::Path>, config: Config) -> DbResult<Self> {
        Self::open_hooked(path, config, NoHook)
    }
}

impl<K: Key, const V: usize, H: WriteHook<K>> ConstTree<K, V, H, Bitcask> {
    pub(crate) fn floor_recovered_gsns(&self, gsn_high: u64, shard_gsn_high: &[u64]) {
        self.durability.engine.gsn().fetch_max(
            gsn_high.saturating_add(1),
            std::sync::atomic::Ordering::Relaxed,
        );
        for (shard_id, high) in shard_gsn_high.iter().copied().enumerate() {
            if let Some(shard) = self.durability.engine.shards().get(shard_id) {
                shard.set_durable_recovered_gsn(shard.durable_recovered_gsn().max(high));
            }
        }
    }

    /// Open or create a `ConstTree` with a write hook for secondary index maintenance.
    ///
    #[doc = include_str!("../docs/exclusive-access.md")]
    pub fn open_hooked(
        path: impl AsRef<std::path::Path>,
        config: Config,
        hook: H,
    ) -> DbResult<Self> {
        #[cfg(feature = "armour")]
        if H::EMITS {
            return Err(DbError::ChangeStreamRequiresManagedOpen);
        }
        let config = config.with_resolved_hints(false);
        Self::open_inner(path, config, hook)
    }

    fn open_inner(path: impl AsRef<std::path::Path>, config: Config, hook: H) -> DbResult<Self> {
        let compaction_threshold = config.compaction_threshold;
        let shard_prefix_bits = config.shard_prefix_bits;
        let reversed = config.reversed;
        let engine = Engine::open(path, config)?;

        let durability = Bitcask {
            engine,
            compaction_threshold,
        };

        let tree = Self {
            index: SkipList::new(reversed),
            shard_prefix_bits,
            reversed,
            hook,
            durability,
            tx_self: OnceLock::new(),
            ttl: TtlSlots::new(),
            counter_dirty: OnceLock::new(),
        };

        // Recover index from disk
        let shard_dirs = tree.durability.engine.shard_dirs();
        let shard_dir_refs = Engine::shard_dir_refs(&shard_dirs);
        let shard_ids = tree.durability.engine.shard_ids();

        let hints = tree.durability.engine.hints();
        let outcome = recover_const_tree::<K, V>(
            &shard_dir_refs,
            &shard_ids,
            tree.index(),
            hints,
            #[cfg(feature = "encryption")]
            tree.durability.engine.shard_ciphers(),
        )?;
        for tail in &outcome.active_tails {
            tree.durability.engine.shards()[tail.shard_idx].apply_recovery_tail(tail)?;
        }
        for (shard_idx, dead) in outcome.shard_dead_bytes {
            tree.durability.engine.shards()[shard_idx].install_dead_bytes(dead);
        }
        for (shard_idx, file_seqs) in outcome.shard_file_seqs {
            tree.durability.engine.shards()[shard_idx].install_recovery_file_seqs(file_seqs);
        }
        // The transaction replay rule compares a reserved GSN against this
        // watermark, so it must exist in every build — not only with
        // `replication` on.
        for (shard_idx, max_gsn) in &outcome.shard_max_gsns {
            tree.durability.engine.shards()[*shard_idx].set_durable_recovered_gsn(*max_gsn);
        }
        let max_gsn = outcome.max_gsn;

        tree.durability
            .engine
            .gsn()
            .fetch_max(max_gsn + 1, std::sync::atomic::Ordering::Relaxed);
        if hints {
            for shard in tree.durability.engine.shards().iter() {
                shard.set_key_len(size_of::<K>());
            }
        }
        tracing::info!(
            key_size = size_of::<K>(),
            V,
            entries = tree.len(),
            "const_tree recovered"
        );

        Ok(tree)
    }

    /// Full clean shutdown: flush buffers, write hint files (if enabled), fsync.
    pub fn clean_shutdown(&self) -> DbResult<()> {
        self.flush_buffers()?;
        if self.durability.engine.hints() {
            self.sync_hints()?;
        }
        self.durability.close()
    }

    /// Graceful shutdown: write hint files (if enabled), flush write buffers + fsync.
    pub fn close(self) -> DbResult<()> {
        self.clean_shutdown()
    }

    /// Flush all shard write buffers to disk (without fsync).
    pub fn flush_buffers(&self) -> DbResult<()> {
        self.durability.engine.flush_buffers()
    }

    /// Durable flush: write buffers + fsync dirty shards.
    ///
    /// `Engine::flush` already flushes each shard's buffer inside `barrier()`,
    /// and keeps going after a per-shard error. A leading `flush_buffers()?`
    /// would abort the whole pass on the first ENOSPC and skip every fsync.
    pub(crate) fn durable_flush(&self) -> DbResult<()> {
        self.durability.flush()
    }

    #[cfg(test)]
    pub(crate) fn selected_compaction_file_ids_for_test(&self) -> Vec<u32> {
        crate::compaction::selected_file_ids_for_test(
            &self.durability.engine.shards()[0],
            self.durability.compaction_threshold,
        )
    }

    #[cfg(test)]
    pub(crate) fn dead_bytes_for_test(&self, file_id: u32) -> u64 {
        self.durability.engine.shards()[0]
            .lock()
            .dead_bytes
            .get(&file_id)
            .copied()
            .unwrap_or(0)
    }

    /// Get the database configuration.
    pub fn config(&self) -> &Config {
        self.durability.engine.config()
    }

    pub(crate) fn tx_reserve_gsns(&self, count: usize) -> u64 {
        self.durability
            .engine
            .gsn()
            .fetch_add(count as u64, std::sync::atomic::Ordering::Relaxed)
    }

    pub(crate) fn tx_encrypted(&self) -> bool {
        #[cfg(feature = "encryption")]
        {
            self.durability.engine.config().encryption_key.is_some()
        }
        #[cfg(not(feature = "encryption"))]
        {
            false
        }
    }

    pub(crate) fn tx_durable_recovered_gsn(&self, sid: usize) -> u64 {
        self.durability.engine.shards()[sid].durable_recovered_gsn()
    }

    /// Trigger a background compaction pass across all shards.
    pub fn compact(&self) -> DbResult<usize> {
        let mut total_compacted = 0;
        for shard in self.durability.engine.shards().iter() {
            total_compacted += compact_shard(shard, self, self.durability.compaction_threshold)?;
        }
        Ok(total_compacted)
    }

    /// Write hint files for all active shard files. Call during graceful shutdown.
    pub fn sync_hints(&self) -> DbResult<()> {
        for shard in self.durability.engine.shards().iter() {
            shard.write_active_hint(size_of::<K>())?;
        }
        Ok(())
    }
}

impl<K: Key, const V: usize, H: WriteHook<K>> CompactionIndex<K> for ConstTree<K, V, H, Bitcask> {
    fn update_if_match(&self, key: &K, old_loc: DiskLoc, new_loc: DiskLoc) -> bool {
        let guard = self.index.collector().enter();
        if let Some(node) = self.index.get(key.as_bytes(), &guard)
            && node.read_loc() == old_loc
        {
            node.write_loc(new_loc);
            return true;
        }
        false
    }

    fn contains_key(&self, key: &K) -> bool {
        self.contains(key)
    }

    fn is_live(&self, _shard_id: u8, key: &K, loc: DiskLoc) -> bool {
        let guard = self.index.collector().enter();
        self.index
            .get(key.as_bytes(), &guard)
            .is_some_and(|node| node.read_loc() == loc)
    }
}

// ==========================================================================
// Fixed-specific: open, close
// ==========================================================================

use crate::durability::Fixed;
use crate::fixed::config::FixedConfig;

impl<K: Key, const V: usize> ConstTree<K, V, NoHook, Fixed> {
    /// Open or create a `ConstTree` with Fixed (fixed-slot) backend.
    /// Recovers the index from existing data files on disk.
    ///
    #[doc = include_str!("../docs/exclusive-access.md")]
    pub fn open(path: impl AsRef<std::path::Path>, config: FixedConfig) -> DbResult<Self> {
        Self::open_fixed_inner(path, config, NoHook)
    }
}

impl<K: Key, const V: usize, H: WriteHook<K>> ConstTree<K, V, H, Fixed> {
    /// Open or create a `ConstTree` with a write hook, using Fixed (fixed-slot) backend.
    ///
    #[doc = include_str!("../docs/exclusive-access.md")]
    pub fn open_with_hook(
        path: impl AsRef<std::path::Path>,
        config: FixedConfig,
        hook: H,
    ) -> DbResult<Self> {
        #[cfg(feature = "armour")]
        if H::EMITS {
            return Err(DbError::ChangeStreamRequiresManagedOpen);
        }
        Self::open_fixed_inner(path, config, hook)
    }

    fn open_fixed_inner(
        path: impl AsRef<std::path::Path>,
        config: FixedConfig,
        hook: H,
    ) -> DbResult<Self> {
        let shard_prefix_bits = config.shard_prefix_bits;
        let reversed = config.reversed;
        let dur = Fixed::open(path, config, size_of::<K>(), V)?;
        let index = SkipList::new(reversed);

        // Recovery: rebuild the in-memory SkipList from on-disk data.
        let index_ref = &index;
        let total_recovered = dur.recover_entries(move |_shard_idx| {
            // One epoch guard per shard thread (hoisted out of the per-entry loop).
            let guard = index_ref.collector().enter();
            move |key_bytes: &[u8], value_bytes: &[u8], slot_id: u32| -> Option<u32> {
                let key = K::from_bytes(key_bytes);
                let mut value = [0u8; V];
                value.copy_from_slice(value_bytes);
                let height = random_height();
                let node = ConstNode::<K, V, u32>::alloc(key, value, slot_id, height);
                match index_ref.insert(node, &guard) {
                    InsertResult::Inserted => None,
                    InsertResult::Exists(existing) => {
                        // Post-crash duplicate key across two OCCUPIED slots
                        // (F1/F2). Last-scanned-wins: adopt the new slot+value
                        // into the existing node, free the freshly-allocated node
                        // (else it leaks — every other call-site deallocs on
                        // `Exists`), and report the previous slot so the scan loop
                        // frees it. This policy matches ConstMap's recovery.
                        let evicted = existing.read_loc();
                        existing.write_data(slot_id, &value);
                        unsafe {
                            ConstNode::<K, V, u32>::dealloc_node(node);
                        }
                        Some(evicted)
                    }
                }
            }
        })?;

        tracing::info!(
            key_size = size_of::<K>(),
            V,
            entries = total_recovered,
            "fixed_tree recovered"
        );

        Ok(Self {
            index,
            durability: dur,
            shard_prefix_bits,
            reversed,
            hook,
            tx_self: OnceLock::new(),
            ttl: TtlSlots::new(),
            counter_dirty: OnceLock::new(),
        })
    }

    /// Full clean shutdown: per-shard sidecar + clean-shutdown flag.
    pub fn clean_shutdown(&self) -> DbResult<()> {
        self.durability.close()
    }

    /// Perform a clean shutdown (Fixed backend).
    pub fn close(self) -> DbResult<()> {
        self.clean_shutdown()
    }

    /// Resolve one planned operation to a `(slot, target_meta)` pair, under the
    /// shard lock the caller already holds.
    ///
    /// The key's current slot comes from the index; a key that is not there
    /// takes a slot from the allocator (a put) or resolves to `NO_SLOT` (a
    /// delete, which has nothing to do).
    pub(crate) fn tx_reserve_fixed(
        &self,
        shard: &mut ConstShard<'_, K, V, H, Fixed>,
        key: &[u8],
        delete: bool,
    ) -> DbResult<(u32, u32)> {
        let existing = self
            .index
            .get(key, &shard.guard)
            .map(|node| node.read_loc());
        if delete && existing.is_none() {
            return Ok((crate::fixed::slot::NO_SLOT, 0));
        }
        shard.held.locked.reserve_slot(existing, delete)
    }

    pub(crate) fn tx_arm_slot(
        shard: &mut ConstShard<'_, K, V, H, Fixed>,
        slot_id: u32,
        meta: u32,
        fresh: bool,
    ) {
        shard.held.locked.arm_slot(slot_id, meta, fresh);
    }

    pub(crate) fn tx_clear_armed_slot(shard: &mut ConstShard<'_, K, V, H, Fixed>) {
        shard.held.locked.clear_armed_slot();
    }

    pub(crate) fn tx_release_slot(
        shard: &mut ConstShard<'_, K, V, H, Fixed>,
        slot_id: u32,
        fresh: bool,
    ) {
        shard.held.locked.release_slot_reservation(slot_id, fresh);
    }

    #[allow(dead_code)] // shard-side slot lookup for future callers; arm/release use `tx_fixed_slot_is_free`
    #[allow(dead_code)] // shard-side slot lookup for future callers; arm/release use `tx_fixed_slot_is_free`
    pub(crate) fn tx_fixed_slot_state(
        shard: &ConstShard<'_, K, V, H, Fixed>,
        slot_id: u32,
    ) -> DbResult<crate::fixed::shard::SlotState> {
        shard.held.locked.slot_state(slot_id)
    }

    pub(crate) fn tx_fixed_slot_is_free(
        shard: &ConstShard<'_, K, V, H, Fixed>,
        slot_id: u32,
    ) -> DbResult<bool> {
        let state = shard.held.locked.slot_state(slot_id)?;
        Ok(crate::fixed::slot::status_of(state.meta) != crate::fixed::slot::STATUS_OCCUPIED)
    }

    /// Current index slot for `key`, if any. Used by Fixed `needs_replay` to
    /// detect a newer incarnation that landed in a different slot after a
    /// failed apply left this reservation orphaned.
    pub(crate) fn tx_fixed_index_slot(
        &self,
        shard: &ConstShard<'_, K, V, H, Fixed>,
        key: &[u8],
    ) -> Option<u32> {
        self.index
            .get(key, &shard.guard)
            .map(|node| node.read_loc())
    }

    /// FixedStore has no encryption (`FixedConfig` carries no key; the meta flag
    /// is reserved), so a Fixed participant never needs an encrypted log.
    pub(crate) fn tx_encrypted(&self) -> bool {
        false
    }

    /// The shard's armed reservation, if any (tests only): what a transaction
    /// left behind on an exit path.
    #[cfg(test)]
    pub(crate) fn tx_debug_armed_slot(&self, sid: usize) -> Option<crate::fixed::shard::ArmedSlot> {
        self.durability.lock_shard(sid).armed_slot()
    }

    /// Slots the allocator considers taken (tests only): a reservation that was
    /// never released shows up here and nowhere else.
    #[cfg(test)]
    pub(crate) fn tx_debug_occupied_slots(&self, sid: usize) -> u32 {
        self.durability.lock_shard(sid).occupied_slots()
    }

    /// Slot state for a key-less caller (tests).
    #[doc(hidden)]
    #[cfg_attr(not(test), allow(dead_code))]
    pub(crate) fn tx_debug_slot_state(
        &self,
        sid: usize,
        slot_id: u32,
    ) -> DbResult<crate::fixed::shard::SlotState> {
        let locked = self.durability.lock_shard(sid);
        locked.slot_state(slot_id)
    }

    #[doc(hidden)]
    #[cfg(any(test, feature = "test-faults"))]
    pub fn tx_debug_force_slot_meta(&self, sid: usize, slot_id: u32, meta: u32) {
        let mut locked = self.durability.lock_shard(sid);
        let _ = locked.force_slot_meta(slot_id, meta);
    }
}

// ==========================================================================
// Fixed-specific replication apply helpers
// ==========================================================================

#[cfg(feature = "replication")]
impl<K: Key, const V: usize, H: WriteHook<K>> ConstTree<K, V, H, Fixed> {
    /// Accessor for the Fixed durability backend — used by the
    /// `FixedReplicationTarget` impl in `crate::fixed_replication::apply`.
    pub(crate) fn fixed_durability(&self) -> &Fixed {
        &self.durability
    }

    /// Return an `Arc<dyn FixedEngineAccess>` for this tree's engine.
    /// Used by integration tests to construct a `FixedReplicationServer`.
    pub fn fixed_engine_access(
        &self,
    ) -> std::sync::Arc<dyn crate::fixed_replication::FixedEngineAccess> {
        self.durability.engine.clone()
    }

    /// Return the on-disk slot id for `key`, if present. O(log n).
    ///
    /// Used by the Fixed replication apply path to detect stale entries that
    /// occupy a slot now reassigned to a different key.
    pub(crate) fn get_slot_id(&self, key: &K) -> Option<u32> {
        let guard = self.index.collector().enter();
        let node = self.index.get(key.as_bytes(), &guard)?;
        Some(node.read_loc())
    }

    /// Remove `key` from the in-memory index only if its current slot matches
    /// `slot_id`. Returns `true` if a removal happened.
    ///
    /// Bypasses durability — the caller is responsible for the underlying slot
    /// state. This is the whole point: preserve a newer mapping at the same
    /// slot if the key has since moved.
    pub(crate) fn remove_key_if_slot_matches(&self, key: &K, slot_id: u32) -> bool {
        let guard = self.index.collector().enter();
        let node = match self.index.get(key.as_bytes(), &guard) {
            Some(n) => n,
            None => return false,
        };
        if node.read_loc() != slot_id {
            return false;
        }
        self.index.remove(key.as_bytes(), &guard);
        true
    }

    /// Upsert `key → (value, slot_id)` in the in-memory index. Bypasses
    /// durability — replication has already rewritten the slot on disk.
    ///
    /// - existing node at the same `slot_id` → in-place SeqLock write
    /// - existing node at a different slot   → remove + insert fresh node
    /// - absent                               → allocate + insert
    pub(crate) fn upsert_replicated(&self, key: &K, value: [u8; V], slot_id: u32) {
        let guard = self.index.collector().enter();
        if let Some(existing) = self.index.get(key.as_bytes(), &guard) {
            if existing.read_loc() == slot_id {
                existing.write_data(slot_id, &value);
                return;
            }
            // Key moved to a new slot: drop stale mapping then insert fresh.
            self.index.remove(key.as_bytes(), &guard);
        }

        let height = random_height();
        let node_ptr = ConstNode::<K, V, u32>::alloc(*key, value, slot_id, height);
        match self.index.insert(node_ptr, &guard) {
            InsertResult::Inserted => {}
            InsertResult::Exists(existing) => {
                // Caller is holding the shard mutex externally, so a concurrent
                // insert of the same key is not expected.
                debug_assert!(
                    false,
                    "upsert_replicated: unexpected concurrent insert for key"
                );
                existing.write_data(slot_id, &value);
                unsafe {
                    ConstNode::<K, V, u32>::dealloc_node(node_ptr);
                }
            }
        }
    }
}

// ==========================================================================
// Family / Ctx seam
// ==========================================================================

/// New-value half of a Const write event: borrowed on the facade path, owned
/// when `Held` buffers for replay (or when the new bytes are produced under the lock).
pub(crate) enum ConstNew<'v, const V: usize> {
    Borrowed(&'v [u8; V]),
    Owned([u8; V]),
}

type ConstWriteEvent<'v, K, const V: usize> = (K, Option<[u8; V]>, Option<ConstNew<'v, V>>);
pub(crate) type ConstOwnedEvent<K, const V: usize> = (K, Option<[u8; V]>, Option<[u8; V]>);
type ConstUpsertEvent<const V: usize> = (Option<[u8; V]>, Option<[u8; V]>);

impl<K: Key, const V: usize, H: WriteHook<K>, D: Durability> Family for ConstTree<K, V, H, D> {
    type Locked<'a>
        = MutexGuard<'a, D::Inner>
    where
        Self: 'a;
    type Event<'v> = ConstWriteEvent<'v, K, V>;
    type Owned = ConstOwnedEvent<K, V>;

    #[inline]
    fn lock_shard(&self, sid: usize) -> Self::Locked<'_> {
        self.durability.lock_shard(sid)
    }

    fn should_sync<'a>(locked: &Self::Locked<'a>) -> bool
    where
        Self: 'a,
    {
        locked.should_sync()
    }

    fn sync(&self, sid: usize) -> DbResult<()> {
        self.durability.sync_shard(sid)
    }

    fn fire(&self, ev: Self::Event<'_>) {
        let (k, old, new) = ev;
        let new = match &new {
            Some(ConstNew::Borrowed(v)) => Some(&v[..]),
            Some(ConstNew::Owned(v)) => Some(&v[..]),
            None => None,
        };
        self.hook.on_write(&k, old.as_ref().map(|v| &v[..]), new);
    }

    fn own(ev: Self::Event<'_>) -> Self::Owned {
        let (k, old, new) = ev;
        let new = new.map(|n| match n {
            ConstNew::Borrowed(v) => *v,
            ConstNew::Owned(v) => v,
        });
        (k, old, new)
    }

    fn fire_owned(&self, ev: &Self::Owned) {
        let (k, old, new) = ev;
        self.hook.on_write(
            k,
            old.as_ref().map(|v| &v[..]),
            new.as_ref().map(|v| &v[..]),
        );
    }

    fn publish_watch(&self, ev: &Self::Event<'_>, active: bool, wakes: &mut WatchWakeBatch) {
        if !H::WATCHES {
            return;
        }
        let Some(channel) = self.hook.watch_channel() else {
            return;
        };
        if !active {
            return;
        }
        let (key, old, new) = ev;
        let old_existed = old.is_some();
        let old = self
            .hook
            .watch_old()
            .then(|| old.map(|value| value.to_vec().into_boxed_slice()))
            .flatten();
        let new = new.as_ref().map(|value| match value {
            ConstNew::Borrowed(value) => value.to_vec().into_boxed_slice(),
            ConstNew::Owned(value) => value.to_vec().into_boxed_slice(),
        });
        channel.publish(
            Arc::new(WatchEventInner::from_write(
                *key,
                old,
                new,
                self.hook.watch_old(),
                old_existed,
            )),
            wakes,
        );
    }

    const WATCHES: bool = H::WATCHES;

    fn watch_enter(&self) -> crate::watch::WatchProducer<'_> {
        if H::WATCHES {
            self.hook
                .watch_channel()
                .and_then(|channel| channel.enter_producer())
                .unwrap_or_else(crate::watch::WatchProducer::inactive)
        } else {
            crate::watch::WatchProducer::inactive()
        }
    }

    fn apply_projection_for_event(&self, ev: &Self::Event<'_>) -> DbResult<()> {
        let (k, old, new) = ev;
        let new_bytes = new.as_ref().map(|n| match n {
            ConstNew::Borrowed(v) => &v[..],
            ConstNew::Owned(v) => &v[..],
        });
        self.apply_projection_for_bytes(k, old.as_ref().map(|v| &v[..]), new_bytes)
    }

    fn apply_projection_for_owned(&self, ev: &Self::Owned) -> DbResult<()> {
        let (k, old, new) = ev;
        self.apply_projection_for_bytes(
            k,
            old.as_ref().map(|v| &v[..]),
            new.as_ref().map(|v| &v[..]),
        )
    }
}

// ==========================================================================
// Generic impl block — works with any D: Durability
// ==========================================================================

impl<K: Key, const V: usize, H: WriteHook<K>, D: Durability> ConstTree<K, V, H, D> {
    /// Collection name for the transaction log. Standalone trees return `""`.
    pub(crate) fn name(&self) -> &'static str {
        ""
    }

    /// `typ_hash` for the transaction log. Standalone trees return `0`.
    pub(crate) fn typ_hash(&self) -> u64 {
        0
    }

    pub(crate) fn shard_count(&self) -> usize {
        self.durability.shard_count()
    }

    /// Register the `Arc` handle used to build a `'static` deferred owner.
    #[cfg_attr(not(test), allow(dead_code))]
    pub(crate) fn bind_tx_self(arc: &Arc<Self>) {
        let _ = arc.tx_self.set(Arc::downgrade(arc));
    }

    pub(crate) fn tx_arc(&self) -> DbResult<Arc<Self>> {
        self.tx_self
            .get()
            .and_then(Weak::upgrade)
            .ok_or(DbError::Internal(
                "ConstTree::bind_tx_self must be called before transaction use",
            ))
    }

    /// Register projection targets and verify same-`Db` membership (T8 T5).
    pub(crate) fn tx_install_projection_metadata(&self, db: &crate::armour::Db) -> DbResult<()> {
        if H::PROJECTS
            && let Some(targets) = self.hook.projection_targets()
        {
            db.tx_register_projection_targets(targets)?;
        }
        if H::COUNTS
            && let Some(targets) = self.hook.counter_targets()
        {
            db.tx_register_counter_targets(targets)?;
        }
        if H::CLAIMS
            && let Some(targets) = self.hook.claim_targets()
        {
            db.tx_register_claim_targets(targets)?;
        }
        if H::EMITS
            && let Some(targets) = self.hook.stream_targets()
        {
            db.tx_register_stream_targets(targets)?;
        }
        Ok(())
    }

    /// Rank-1 participants a write here can reach: declared secondary-index
    /// targets, declared counter targets, **and** declared unique-namespace
    /// targets. All three are written under a lock the commit takes in its
    /// second round, so all three must be in the plan before the first lock
    /// is taken (T9 Task 9; unique namespaces added T10 Task 6). A namespace
    /// is preplanned whether or not this particular row claims — the plan
    /// cannot know what the projection will emit before it runs, and a
    /// namespace absent from the plan cannot be locked in the second round.
    pub(crate) fn tx_preplan_projection_targets(
        &self,
    ) -> DbResult<Vec<Arc<dyn crate::txn::participant::TxParticipant>>> {
        let mut out = Vec::new();
        if H::PROJECTS
            && let Some(targets) = self.hook.projection_targets()
        {
            out.extend(
                targets
                    .erased_targets()
                    .iter()
                    .filter_map(|h| targets.resolve_tx_participant(h.name)),
            );
        }
        if H::COUNTS
            && let Some(targets) = self.hook.counter_targets()
        {
            out.extend(targets.all().map(|m| m.as_tx_participant()));
        }
        if H::CLAIMS
            && let Some(targets) = self.hook.claim_targets()
        {
            out.extend(targets.all().map(|t| t.as_tx_participant()));
        }
        if H::EMITS
            && let Some(targets) = self.hook.stream_targets()
        {
            out.extend(targets.tx_participants());
        }
        Ok(out)
    }
    pub(crate) fn tx_preplan_stream_targets(
        &self,
    ) -> Vec<Arc<dyn crate::txn::participant::TxParticipant>> {
        self.hook
            .stream_targets()
            .map(|t| t.tx_participants())
            .unwrap_or_default()
    }
    pub(crate) fn tx_prepare_streams(
        &self,
    ) -> DbResult<Vec<crate::queue::retention::ProducerPermit>> {
        self.hook
            .stream_targets()
            .map_or(Ok(Vec::new()), crate::QueueTargets::prepare_streams)
    }

    pub(crate) fn tx_stream_writes(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<crate::txn::participant::TxStreamWrites> {
        if !H::EMITS {
            return Ok(crate::txn::participant::TxStreamWrites::empty());
        }
        let Some(targets) = self.hook.stream_targets() else {
            return Ok(crate::txn::participant::TxStreamWrites::empty());
        };
        let key = K::from_bytes(key);
        let (max_rows, max_bytes) = self.effective_projection_limits();
        let mut sink = crate::EmitSink::new(targets, max_rows, max_bytes);
        self.hook.stream_project(&key, old, new, &mut sink)?;
        targets.lower(&sink)
    }

    pub(crate) fn tx_project_writes(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
        if !H::PROJECTS {
            return Ok(Vec::new());
        }
        let Some(targets) = self.hook.projection_targets() else {
            return Ok(Vec::new());
        };
        let key_typed = K::from_bytes(key);
        let (max_rows, max_bytes) = self.effective_projection_limits();
        let plan = crate::projection::diff_projection_bytes(
            &self.hook, targets, max_rows, max_bytes, &key_typed, old, new,
        )?;
        plan.into_tx_derived(targets)
    }

    /// Counter deltas for one decided write inside a transaction — the counter
    /// twin of [`tx_project_writes`](Self::tx_project_writes).
    ///
    /// Deltas, not writes: the absolute value a delta resolves to is only
    /// knowable under the target's own lock, which the commit takes later.
    pub(crate) fn tx_counter_deltas_for(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
        if !H::COUNTS {
            return Ok(Vec::new());
        }
        let Some(targets) = self.hook.counter_targets() else {
            return Ok(Vec::new());
        };
        let key_typed = K::from_bytes(key);
        let (max_rows, max_bytes) = self.effective_projection_limits();
        let deltas = crate::counter::diff_counters_bytes(
            &self.hook, targets, max_rows, max_bytes, &key_typed, old, new,
        )?;
        crate::counter::into_tx_counter_deltas(targets, deltas)
    }

    /// Claim ops for one decided write inside a transaction — the claim twin
    /// of [`tx_project_writes`](Self::tx_project_writes)/
    /// [`tx_counter_deltas_for`](Self::tx_counter_deltas_for) (T10 Task 6).
    ///
    /// `name` is the wrapping collection's own name — supplied by the caller,
    /// since a standalone `ConstTree` (this type) has no name of its own (see
    /// [`Self::name`]); `ZeroTree`'s `TxParticipant` impl passes `T::NAME`.
    pub(crate) fn tx_claims_for(
        &self,
        name: &'static str,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::ClaimOp>> {
        if !H::CLAIMS {
            return Ok(Vec::new());
        }
        let Some(targets) = self.hook.claim_targets() else {
            return Ok(Vec::new());
        };
        let key_typed = K::from_bytes(key);
        let (max_rows, max_bytes) = self.effective_projection_limits();
        let delta = crate::unique::claim_delta_bytes(
            &self.hook, targets, max_rows, max_bytes, &key_typed, old, new,
        )?;
        crate::unique::into_tx_claim_ops(targets, name, key, delta)
    }

    /// Lock `sid` and build the shard object, exactly as [`atomic`] does but
    /// without the closure scope, so a transaction can hold several shards.
    pub(crate) fn tx_shard(&self, sid: usize) -> ConstShard<'_, K, V, H, D> {
        let locked = self.lock_shard(sid);
        let guard = self.index.collector().enter();
        ConstShard {
            tree: self,
            held: Held {
                locked,
                sid,
                events: Vec::new(),
                watch_wakes: crate::watch::WatchWakeBatch::default(),
                publish_watches: true,
            },
            guard,
        }
    }

    /// Release a shard built by [`tx_shard`], returning its buffered events.
    /// The `seize` guard is dropped by the caller's `Deferred` owner, not here.
    pub(crate) fn tx_release_shard(
        shard: ConstShard<'_, K, V, H, D>,
    ) -> (
        Vec<ConstOwnedEvent<K, V>>,
        crate::watch::WatchWakeBatch,
        seize::LocalGuard<'_>,
    ) {
        let ConstShard { held, guard, .. } = shard;
        let Held {
            locked,
            events,
            watch_wakes,
            ..
        } = held;
        locked.release();
        (events, watch_wakes, guard)
    }

    /// Build per-shard TTL indexes by walking live in-memory entries.
    #[doc(hidden)]
    pub fn build_ttl_index(&self) -> DbResult<()> {
        if !H::EXPIRES {
            return Ok(());
        }
        let n = self.durability.shard_count();
        let slots = self.ttl.install_and_clear(n);

        let _guard = self.index.collector().enter();
        let mut current = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        while !current.is_null() {
            let node = unsafe { &*current };
            current = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
            if node.is_marked() {
                continue;
            }
            let sid = self.shard_for(&node.key);
            // Shard lock before reading value — see TypedTree::build_ttl_index.
            let _shard = self.durability.lock_shard(sid);
            let value = node.read_value();
            let deadline = self.hook.deadline(&node.key, &value[..]);
            sync::lock(&slots[sid]).record(&node.key, deadline);
        }
        Ok(())
    }

    /// Deadline for a row, or `None` when TTL is off for this collection.
    ///
    /// Call this *before* the durable write, not after: the extractor is user
    /// code and may panic, and a panic between a committed row and its index
    /// entry leaks the row — it stays live with no deadline recorded, invisible
    /// until it fails to expire. Computing first makes a panicking extractor
    /// abort the write instead, which is what the Typed families already do.
    #[inline]
    fn deadline_for(&self, key: &K, value: &[u8; V]) -> Option<u64> {
        if H::EXPIRES {
            self.hook.deadline(key, &value[..])
        } else {
            None
        }
    }

    /// Record a deadline when `H::EXPIRES` and the index is built.
    /// Caller must hold the engine shard lock for `shard_id` first.
    fn record_ttl(&self, shard_id: usize, key: &K, expires_at: Option<u64>) {
        if !H::EXPIRES {
            return;
        }
        self.ttl.record(shard_id, key, expires_at);
    }

    #[doc(hidden)]
    pub fn ttl_debug_len(&self) -> usize {
        self.ttl.debug_len()
    }

    #[doc(hidden)]
    pub fn ttl_debug_deadline(&self, key: &K) -> Option<u64> {
        self.ttl.debug_deadline(self.shard_for(key), key)
    }

    pub fn sweep_expired(&self, now_ms: u64, budget: usize) -> DbResult<SweepOutcome> {
        if !H::EXPIRES {
            return Ok(SweepOutcome::default());
        }
        let Some(slots) = self.ttl.get() else {
            return Ok(SweepOutcome::default());
        };
        let Some(_pass) = self.ttl.begin_pass() else {
            return Ok(SweepOutcome::default());
        };
        let shard_count = slots.len();
        let mut remaining = budget;
        let mut outcome = SweepOutcome::default();

        while remaining > 0 && self.ttl.any_due(now_ms) {
            let sid = self.ttl.next_shard(shard_count);
            let mut shard_events = Vec::new();
            let mut wakes = WatchWakeBatch::default();
            let mut locked = self.lock_shard(sid);
            let producer = self.watch_enter();
            let active = producer.active();
            let context_active = H::WATCHES.then(|| crate::watch::ContextActiveGuard::set(active));
            let guard = self.index.collector().enter();
            let mut idx = sync::lock(&slots[sid]);

            while remaining > 0 {
                let Some((candidate_dl, key)) = idx.pop_due(now_ms) else {
                    break;
                };
                remaining -= 1;

                let action = match self.index.get(key.as_bytes(), &guard) {
                    Some(node) => {
                        let value = node.read_value();
                        match crate::ttl::catch_deadline(|| self.hook.deadline(&key, &value[..])) {
                            Some(Some(dl)) if dl <= now_ms => {
                                ConstSweepAction::Delete(key, candidate_dl)
                            }
                            Some(Some(dl)) => ConstSweepAction::Reschedule(key, dl),
                            Some(None) => ConstSweepAction::Drop,
                            None => ConstSweepAction::Retry(key),
                        }
                    }
                    None => ConstSweepAction::Drop,
                };

                match action {
                    ConstSweepAction::Delete(key, candidate_dl) => {
                        drop(idx);
                        match self.delete_locked(sid, &mut *locked, &guard, &key) {
                            Ok(old) => {
                                if let Some(old) = old {
                                    if let Some(event) =
                                        self.defer_write_event(&key, Some(old), None)
                                    {
                                        self.publish_watch(&event, active, &mut wakes);
                                        shard_events.push(Self::own(event));
                                    }
                                    outcome.deleted += 1;
                                }
                                idx = sync::lock(&slots[sid]);
                            }
                            Err(e) => {
                                sync::lock(&slots[sid]).record(&key, Some(candidate_dl));
                                // Do not lose what this pass already did: `failed` counts rows that were
                                // rescheduled after a read/decode failure, and the warning is the only
                                // signal that a collection is degrading.
                                crate::ttl::log_sweep_failures(None, &outcome);
                                let needs_sync = Self::should_sync(&locked);
                                locked.release();
                                drop(guard);
                                drop(context_active);
                                let _ = crate::projection::replay_owned_events(
                                    self,
                                    &shard_events,
                                    needs_sync,
                                    sid,
                                );
                                wakes.drain();
                                return Err(e);
                            }
                        }
                    }
                    ConstSweepAction::Reschedule(key, dl) => {
                        idx.record(&key, Some(dl));
                    }
                    ConstSweepAction::Retry(key) => {
                        outcome.failed += 1;
                        idx.record(&key, Some(now_ms + crate::ttl::TTL_RETRY_BACKOFF_MS));
                    }
                    ConstSweepAction::Drop => {}
                }
            }

            drop(idx);
            let needs_sync = Self::should_sync(&locked);
            locked.release();
            drop(guard);
            drop(context_active);
            let replay =
                crate::projection::replay_owned_events(self, &shard_events, needs_sync, sid);
            wakes.drain();
            replay?;
        }

        outcome.more = self.ttl.any_due(now_ms);
        crate::ttl::log_sweep_failures(None, &outcome);
        Ok(outcome)
    }

    pub fn set_ttl_sweep(&self, enabled: bool) -> DbResult<()> {
        if H::EXPIRES {
            self.ttl.set_enabled(enabled)
        } else {
            Ok(())
        }
    }

    pub fn mark_replication_follower(&self) {
        if H::EXPIRES {
            self.ttl.mark_replication_follower();
        }
    }

    pub fn rebuild_ttl_index(&self) -> DbResult<()> {
        if !H::EXPIRES {
            return Err(DbError::Client("collection has no TTL policy"));
        }
        self.build_ttl_index()?;
        self.ttl.clear_rebuild_requirement();
        Ok(())
    }

    /// Get a value by key. Lock-free, zero disk I/O.
    ///
    /// **O(log n)** — SkipList point lookup.
    pub fn get(&self, key: &K) -> Option<[u8; V]> {
        metrics::counter!("armdb.ops", "op" => "get", "tree" => "const_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_tree.get");
        let guard = self.index.collector().enter();
        let node = self.index.get(key.as_bytes(), &guard)?;
        Some(node.read_value())
    }

    /// Get a value by key, returning `Err(KeyNotFound)` if absent.
    pub fn get_or_err(&self, key: &K) -> DbResult<[u8; V]> {
        self.get(key).ok_or(DbError::KeyNotFound)
    }

    /// Buffer a write-hook event when hooks are enabled (owned new bytes — Held / update).
    #[inline]
    fn defer_write_event<'v>(
        &self,
        key: &K,
        old: Option<[u8; V]>,
        new: Option<[u8; V]>,
    ) -> Option<ConstWriteEvent<'v, K, V>> {
        if H::NEEDS_WRITE || (H::WATCHES && crate::watch::context_active()) {
            Some((*key, old, new.map(ConstNew::Owned)))
        } else {
            None
        }
    }

    /// Facade path: borrow the caller's new value (no `[u8; V]` copy into the event).
    #[inline]
    fn defer_borrowed_event<'v>(
        &self,
        key: &K,
        old: Option<[u8; V]>,
        new: Option<&'v [u8; V]>,
    ) -> Option<ConstWriteEvent<'v, K, V>> {
        if H::NEEDS_WRITE || (H::WATCHES && crate::watch::context_active()) {
            Some((*key, old, new.map(ConstNew::Borrowed)))
        } else {
            None
        }
    }

    /// Insert or update a key-value pair. Returns the old value if the key existed.
    pub fn put(&self, key: &K, value: &[u8; V]) -> DbResult<Option<[u8; V]>> {
        metrics::counter!("armdb.ops", "op" => "put", "tree" => "const_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_tree.put");
        let shard_id = self.shard_for(key);
        // F2: hook observes in-memory commit — mutation is already visible to readers,
        // so it fires even when the subsequent `sync()` returns `Err` (precedent: closure
        // `Err` in `atomic()`, see `hook.rs`). Owned::with_shard fires before sync.
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            // Enter seize *inside* the closure, i.e. after `with_shard` took the mutex.
            // Entering before would pin the epoch for the whole time this thread blocks
            // on a contended shard, stalling `try_retire` for every other thread — the
            // same hazard `atomic()` avoids by locking first.
            let guard = self.index.collector().enter();
            let old = self.put_locked(shard_id, &mut *locked, &guard, key, value)?;
            *ev_out = self.defer_borrowed_event(key, old, Some(value));
            Ok(old)
        })
    }

    /// Insert a key-value pair only if the key does not exist.
    /// Returns `Err(KeyExists)` if the key is already present.
    pub fn insert(&self, key: &K, value: &[u8; V]) -> DbResult<()> {
        metrics::counter!("armdb.ops", "op" => "insert", "tree" => "const_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_tree.insert");
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let guard = self.index.collector().enter();
            self.insert_locked(shard_id, &mut *locked, &guard, key, value)?;
            *ev_out = self.defer_borrowed_event(key, None, Some(value));
            Ok(())
        })
    }

    /// Delete a key. Returns the old value if the key existed.
    pub fn delete(&self, key: &K) -> DbResult<Option<[u8; V]>> {
        metrics::counter!("armdb.ops", "op" => "delete", "tree" => "const_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_tree.delete");
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let guard = self.index.collector().enter();
            let old = self.delete_locked(shard_id, &mut *locked, &guard, key)?;
            let ev = match &old {
                Some(o) => self.defer_write_event(key, Some(*o), None),
                None => None,
            };
            *ev_out = ev;
            Ok(old)
        })
    }

    /// Atomically execute multiple operations on a single shard.
    /// All keys must route to the same shard as `shard_key`.
    /// The closure must be short — shard lock is held for its duration.
    pub fn atomic<R>(
        &self,
        shard_key: &K,
        f: impl FnOnce(&mut ConstShard<'_, K, V, H, D>) -> DbResult<R>,
    ) -> DbResult<R> {
        let _atomic_depth = crate::AtomicDepthGuard::try_enter()?;
        let shard_id = self.shard_for(shard_key);
        let mut shard = self.tx_shard(shard_id);
        let result = f(&mut shard);
        let needs_sync = Self::should_sync(&shard.held.locked);
        let (events, wakes, guard) = Self::tx_release_shard(shard);
        let result =
            crate::projection::finish_atomic_teardown(self, shard_id, result, needs_sync, &events);
        drop(guard);
        wakes.drain();
        result
    }

    /// Persist `counters_dirty = true` for this collection (idempotent per
    /// session — see `CounterDirty::mark`). No-op if nothing is bound (a
    /// direct open, or no counter targets declared).
    ///
    /// Called once, from `Db::open_managed`, *before* the collection is
    /// published — never from the write path. See `counter_dirty.rs`.
    pub(crate) fn mark_counter_dirty(&self) -> DbResult<()> {
        match self.counter_dirty.get() {
            Some(d) => d.mark(),
            None => Ok(()),
        }
    }

    fn apply_projection_for_bytes(
        &self,
        key: &K,
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<()> {
        if !H::PROJECTS && !H::COUNTS {
            return Ok(());
        }
        let (max_rows, max_bytes) = self.effective_projection_limits();

        let mut plan = match (H::PROJECTS, self.hook.projection_targets()) {
            (true, Some(targets)) => crate::projection::diff_projection_bytes(
                &self.hook, targets, max_rows, max_bytes, key, old, new,
            )?,
            _ => crate::projection::PendingProjection::default(),
        };
        if H::COUNTS
            && let Some(counter_targets) = self.hook.counter_targets()
        {
            let deltas = crate::counter::diff_counters_bytes(
                &self.hook,
                counter_targets,
                max_rows,
                max_bytes,
                key,
                old,
                new,
            )?;
            if !deltas.is_empty() {
                plan.set_deltas(deltas, counter_targets.clone());
            }
        }
        plan.apply_or_incomplete()
    }

    /// Decide-and-write: insert, update, keep, or delete based on the closure's
    /// [`Upsert`](crate::Upsert) decision. The closure must be short — the shard
    /// lock is held across it, so it must not panic and must not call back into
    /// this collection (the shard mutex is not reentrant).
    pub fn upsert<U: crate::IntoUpsert<[u8; V]>>(
        &self,
        key: &K,
        f: impl FnOnce(Option<&[u8; V]>) -> U,
    ) -> DbResult<U::Out> {
        metrics::counter!("armdb.ops", "op" => "upsert", "tree" => "const_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_tree.upsert");
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let guard = self.index.collector().enter();
            let existing = self.index.get(key.as_bytes(), &guard);
            let cur = existing.map(|n| n.read_value());
            let (decision, out) = f(cur.as_ref()).into_parts();
            if let Some((old, new)) =
                self.upsert_locked(shard_id, &mut *locked, &guard, key, decision, existing)?
            {
                *ev_out = self.defer_write_event(key, old, new);
            }
            Ok(out)
        })
    }

    /// Generic [`upsert_via_held`] path — retained for agreement tests against the fused body.
    ///
    /// # Unstable
    /// Not part of the public API: it is `pub` only because `tests/upsert_agreement.rs`
    /// is an integration test. May change or disappear in any release — call
    /// [`upsert`](Self::upsert).
    #[doc(hidden)]
    pub fn upsert_generic<U: crate::IntoUpsert<[u8; V]>>(
        &self,
        key: &K,
        f: impl FnOnce(Option<&[u8; V]>) -> U,
    ) -> DbResult<U::Out> {
        self.atomic(key, |s| crate::upsert::upsert_via_held(s, key, f))
    }

    /// Fused upsert body: one index lookup is done by the caller and passed as
    /// `existing`. Returns `Some((old, new))` when a hook event is needed.
    fn upsert_locked(
        &self,
        shard_id: usize,
        inner: &mut D::Inner,
        guard: &seize::LocalGuard<'_>,
        key: &K,
        decision: crate::upsert::Upsert<[u8; V]>,
        existing: Option<&ConstNode<K, V, D::Loc>>,
    ) -> DbResult<Option<ConstUpsertEvent<V>>> {
        use crate::upsert::Upsert;
        // T10 Task 5: before any mutation. Covers both `ConstTree::upsert`
        // and `ConstShard::upsert` (atomic) — the single fused body both
        // route through. `Keep` makes no mutation, so it needs no guard.
        //
        // Gated on `claim_targets().is_some()` too (F1 fix, review):
        // `Projected` sets `H::CLAIMS = true` for every carrier, including
        // plain index/counter sources — without this, they would pay an
        // extra `ConstNode::read_value()` per `upsert` for a check that
        // always no-ops.
        if H::EMITS || (H::CLAIMS && self.hook.claim_targets().is_some()) {
            #[cfg(feature = "ctx-gates")]
            crate::ctx_gates_sync::record_claim_pre_read();
            match &decision {
                Upsert::Put(value) => {
                    let old = existing.map(|n| n.read_value());
                    self.claim_guard_bytes(key, old.as_ref().map(|v| &v[..]), Some(&value[..]))?;
                }
                Upsert::Delete => {
                    if let Some(node) = existing {
                        let old = node.read_value();
                        self.claim_guard_bytes(key, Some(&old[..]), None)?;
                    }
                }
                Upsert::Keep => {}
            }
        }
        match decision {
            Upsert::Put(value) => {
                // Before the first `?` — see `deadline_for`.
                let expires_at = self.deadline_for(key, &value);
                let old = match existing {
                    Some(node) => {
                        let old_value = node.read_value();
                        let old_loc = node.read_loc();
                        let new_loc =
                            inner.write_update(shard_id as u8, old_loc, key.as_bytes(), &value)?;
                        node.write_data(new_loc, &value);
                        Some(old_value)
                    }
                    None => {
                        let loc = inner.write_new(shard_id as u8, key.as_bytes(), &value)?;
                        let height = random_height();
                        let node_ptr = ConstNode::<K, V, D::Loc>::alloc(*key, value, loc, height);
                        match self.index.insert(node_ptr, guard) {
                            InsertResult::Inserted => None,
                            InsertResult::Exists(existing) => {
                                inner.write_discard(loc, key.as_bytes())?;
                                let old_value = existing.read_value();
                                let old_loc = existing.read_loc();
                                let new_loc = inner.write_update(
                                    shard_id as u8,
                                    old_loc,
                                    key.as_bytes(),
                                    &value,
                                )?;
                                existing.write_data(new_loc, &value);
                                unsafe {
                                    ConstNode::<K, V, D::Loc>::dealloc_node(node_ptr);
                                }
                                Some(old_value)
                            }
                        }
                    }
                };
                self.record_ttl(shard_id, key, expires_at);
                Ok(Some((old, Some(value))))
            }
            Upsert::Delete => {
                if let Some(node) = existing {
                    let old_value = node.read_value();
                    let old_loc = node.read_loc();
                    inner.write_tombstone(shard_id as u8, old_loc, key.as_bytes())?;
                    self.index.remove(key.as_bytes(), guard);
                    self.record_ttl(shard_id, key, None);
                    Ok(Some((Some(old_value), None)))
                } else {
                    Ok(None)
                }
            }
            Upsert::Keep => Ok(None),
        }
    }

    fn put_locked(
        &self,
        shard_id: usize,
        inner: &mut D::Inner,
        guard: &seize::LocalGuard<'_>,
        key: &K,
        value: &[u8; V],
    ) -> DbResult<Option<[u8; V]>> {
        // Before the first `?` — see `deadline_for`.
        let expires_at = self.deadline_for(key, value);
        // Fast path: key exists — SeqLock write, no node allocation
        if let Some(existing) = self.index.get(key.as_bytes(), guard) {
            let old_value = existing.read_value();
            self.claim_guard_bytes(key, Some(&old_value[..]), Some(&value[..]))?;
            let old_loc = existing.read_loc();
            let new_loc = inner.write_update(shard_id as u8, old_loc, key.as_bytes(), value)?;
            existing.write_data(new_loc, value);
            self.record_ttl(shard_id, key, expires_at);
            return Ok(Some(old_value));
        }
        self.claim_guard_bytes(key, None, Some(&value[..]))?;

        // Slow path: new key — allocate node + take write_lock via insert
        let loc = inner.write_new(shard_id as u8, key.as_bytes(), value)?;
        let height = random_height();
        let node_ptr = ConstNode::<K, V, D::Loc>::alloc(*key, *value, loc, height);

        match self.index.insert(node_ptr, guard) {
            InsertResult::Inserted => {
                self.record_ttl(shard_id, key, expires_at);
                Ok(None)
            }
            InsertResult::Exists(existing) => {
                // Race: another shard inserted same key between get and insert
                inner.write_discard(loc, key.as_bytes())?;
                let old_value = existing.read_value();
                let old_loc = existing.read_loc();
                let new_loc = inner.write_update(shard_id as u8, old_loc, key.as_bytes(), value)?;
                existing.write_data(new_loc, value);
                // Deallocate the unused new node
                unsafe {
                    ConstNode::<K, V, D::Loc>::dealloc_node(node_ptr);
                }
                self.record_ttl(shard_id, key, expires_at);
                Ok(Some(old_value))
            }
        }
    }

    fn insert_locked(
        &self,
        shard_id: usize,
        inner: &mut D::Inner,
        guard: &seize::LocalGuard<'_>,
        key: &K,
        value: &[u8; V],
    ) -> DbResult<()> {
        if self.index.get(key.as_bytes(), guard).is_some() {
            return Err(DbError::KeyExists);
        }
        // An insert only ever reaches here on an absent key, so `old` is
        // always `None` — still routes through the guard: a source that
        // claims must reserve its very first claim inside `db.tx()` too.
        self.claim_guard_bytes(key, None, Some(&value[..]))?;

        // Before the first `?` — see `deadline_for`.
        let expires_at = self.deadline_for(key, value);
        let loc = inner.write_new(shard_id as u8, key.as_bytes(), value)?;
        let height = random_height();
        let node_ptr = ConstNode::<K, V, D::Loc>::alloc(*key, *value, loc, height);
        match self.index.insert(node_ptr, guard) {
            InsertResult::Inserted => {
                self.record_ttl(shard_id, key, expires_at);
                Ok(())
            }
            InsertResult::Exists(_) => {
                inner.write_discard(loc, key.as_bytes())?;
                unsafe {
                    ConstNode::<K, V, D::Loc>::dealloc_node(node_ptr);
                }
                Err(DbError::KeyExists)
            }
        }
    }

    fn delete_locked(
        &self,
        shard_id: usize,
        inner: &mut D::Inner,
        guard: &seize::LocalGuard<'_>,
        key: &K,
    ) -> DbResult<Option<[u8; V]>> {
        if let Some(existing) = self.index.get(key.as_bytes(), guard) {
            let old_value = existing.read_value();
            // T10 Task 5: a delete is a pure release ({k} -> ∅) — refused
            // outside `db.tx()` exactly like an insert's pure claim.
            self.claim_guard_bytes(key, Some(&old_value[..]), None)?;
            let old_loc = existing.read_loc();
            inner.write_tombstone(shard_id as u8, old_loc, key.as_bytes())?;
            self.index.remove(key.as_bytes(), guard);
            self.record_ttl(shard_id, key, None);
            return Ok(Some(old_value));
        }
        Ok(None)
    }

    fn put_no_hook(&self, key: &K, value: &[u8; V]) -> DbResult<Option<[u8; V]>> {
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, _ev_out| {
            let guard = self.index.collector().enter();
            let old = self.put_locked(shard_id, &mut *locked, &guard, key, value)?;
            Ok(old)
        })
    }

    fn delete_no_hook(&self, key: &K) -> DbResult<Option<[u8; V]>> {
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, _ev_out| {
            let guard = self.index.collector().enter();
            let old = self.delete_locked(shard_id, &mut *locked, &guard, key)?;
            Ok(old)
        })
    }

    /// Compare-and-swap: if current value == expected, replace with new_value.
    /// Returns `Ok(())` on success, `Err(CasMismatch)` if current != expected,
    /// `Err(KeyNotFound)` if key doesn't exist.
    /// Zero I/O — values are inline in SkipList nodes.
    pub fn cas(&self, key: &K, expected: &[u8; V], new_value: &[u8; V]) -> DbResult<()> {
        metrics::counter!("armdb.ops", "op" => "cas", "tree" => "const_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_tree.cas");
        // Outside the shard lock: the new bytes are final regardless of whether
        // the compare succeeds, and the extractor must not run under the lock.
        let expires_at = self.deadline_for(key, new_value);
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let guard = self.index.collector().enter();
            let existing = self
                .index
                .get(key.as_bytes(), &guard)
                .ok_or(DbError::KeyNotFound)?;

            let current_value = existing.read_value();
            if current_value != *expected {
                return Err(DbError::CasMismatch);
            }
            // T10 Task 5: `cas` has its own inline body (does not go through
            // `put_locked`/`upsert_locked`), so it needs its own guard call.
            self.claim_guard_bytes(key, Some(&current_value[..]), Some(&new_value[..]))?;

            let old_loc = existing.read_loc();
            let new_loc =
                locked.write_update(shard_id as u8, old_loc, key.as_bytes(), new_value)?;
            existing.write_data(new_loc, new_value);
            self.record_ttl(shard_id, key, expires_at);
            // `expected` / `new_value` outlive the closure, so the new half rides
            // borrowed. The old half cannot: an event outlives the closure while a
            // value read from the node under the lock does not, so `Option<[u8; V]>`
            // by value is structural here, not an oversight.
            *ev_out = self.defer_borrowed_event(key, Some(*expected), Some(new_value));
            Ok(())
        })
    }

    /// Compare-and-delete: if the current value == `expected`, delete the key.
    /// Returns `Ok(())` on success, `Err(CasMismatch)` if current != expected,
    /// `Err(KeyNotFound)` if the key doesn't exist.
    /// Zero I/O on the comparison — values are inline in SkipList nodes.
    /// Works on both Bitcask and FixedStore backends.
    pub fn compare_delete(&self, key: &K, expected: &[u8; V]) -> DbResult<()> {
        metrics::counter!("armdb.ops", "op" => "compare_delete", "tree" => "const_tree")
            .increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_tree.compare_delete");
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let guard = self.index.collector().enter();
            let existing = self
                .index
                .get(key.as_bytes(), &guard)
                .ok_or(DbError::KeyNotFound)?;

            let current_value = existing.read_value();
            if current_value != *expected {
                return Err(DbError::CasMismatch);
            }
            // T10 Task 5: `compare_delete` has its own inline body — a pure
            // release ({k} -> ∅), refused outside `db.tx()`.
            self.claim_guard_bytes(key, Some(&current_value[..]), None)?;

            let old_loc = existing.read_loc();
            locked.write_tombstone(shard_id as u8, old_loc, key.as_bytes())?;
            self.index.remove(key.as_bytes(), &guard);
            self.record_ttl(shard_id, key, None);
            *ev_out = self.defer_write_event(key, Some(current_value), None);
            Ok(())
        })
    }

    /// Atomically read-modify-write. Returns `Some(new_value)` if key existed, `None` otherwise.
    /// Zero I/O — values are inline in SkipList nodes.
    /// The closure must not be heavy (shard lock is held).
    pub fn update(
        &self,
        key: &K,
        f: impl FnOnce(&[u8; V]) -> [u8; V],
    ) -> DbResult<Option<[u8; V]>> {
        self.update_inner(key, f, false)
    }

    /// Like [`update()`](Self::update), but returns `Some(old_value)` instead of the new one.
    pub fn fetch_update(
        &self,
        key: &K,
        f: impl FnOnce(&[u8; V]) -> [u8; V],
    ) -> DbResult<Option<[u8; V]>> {
        self.update_inner(key, f, true)
    }

    fn update_locked(
        &self,
        shard_id: usize,
        inner: &mut D::Inner,
        guard: &seize::LocalGuard<'_>,
        key: &K,
        f: impl FnOnce(&[u8; V]) -> [u8; V],
    ) -> DbResult<Option<([u8; V], [u8; V])>> {
        let existing = match self.index.get(key.as_bytes(), guard) {
            Some(n) => n,
            None => return Ok(None),
        };
        let old_value = existing.read_value();
        let new_value = f(&old_value);
        self.claim_guard_bytes(key, Some(&old_value[..]), Some(&new_value[..]))?;
        // Before the first `?` — see `deadline_for`. The new value only exists
        // under the lock here, so this one cannot be hoisted out of it.
        let expires_at = self.deadline_for(key, &new_value);
        let old_loc = existing.read_loc();
        let new_loc = inner.write_update(shard_id as u8, old_loc, key.as_bytes(), &new_value)?;
        existing.write_data(new_loc, &new_value);
        self.record_ttl(shard_id, key, expires_at);
        Ok(Some((old_value, new_value)))
    }

    fn update_inner(
        &self,
        key: &K,
        f: impl FnOnce(&[u8; V]) -> [u8; V],
        return_old: bool,
    ) -> DbResult<Option<[u8; V]>> {
        metrics::counter!("armdb.ops", "op" => "update", "tree" => "const_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_tree.update");
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let guard = self.index.collector().enter();
            let (old_value, new_value) =
                match self.update_locked(shard_id, &mut *locked, &guard, key, f)? {
                    Some(pair) => pair,
                    None => return Ok(None),
                };
            let result = Some(if return_old { old_value } else { new_value });
            *ev_out = self.defer_write_event(key, Some(old_value), Some(new_value));
            Ok(result)
        })
    }

    /// Check if a key exists.
    pub fn contains(&self, key: &K) -> bool {
        self.get(key).is_some()
    }

    /// Return the first entry in index order, or `None` if empty.
    /// With `reversed=true` (default): the entry with the largest key.
    /// O(1) — follows head's level-0 pointer, skipping marked nodes.
    pub fn first(&self) -> Option<(K, [u8; V])> {
        let guard = self.index.collector().enter();
        let mut ptr = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        while !ptr.is_null() {
            let node = unsafe { &*ptr };
            if !node.is_marked() {
                return Some((node.key, node.read_value()));
            }
            ptr = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
        }
        let _ = guard;
        None
    }

    /// Return the last entry in index order, or `None` if empty.
    /// With `reversed=true` (default): the entry with the smallest key.
    pub fn last(&self) -> Option<(K, [u8; V])> {
        self.iter().next_back()
    }

    // -- Range helpers (front/back pointer positioning) -------------------------

    /// ASC front: position at lower bound.
    fn resolve_front_asc(
        &self,
        bound: &Bound<&K>,
        guard: &seize::LocalGuard<'_>,
    ) -> *mut ConstNode<K, V, D::Loc> {
        match bound {
            Bound::Included(k) => self.index.find_first_ge(k.as_bytes(), guard),
            Bound::Excluded(k) => {
                let ge = self.index.find_first_ge(k.as_bytes(), guard);
                if !ge.is_null()
                    && !unsafe { &*ge }.is_marked()
                    && unsafe { &*ge }.key_bytes() == k.as_bytes()
                {
                    crate::skiplist::strip_mark(unsafe {
                        (*ge).tower(0).load(std::sync::atomic::Ordering::Acquire)
                    })
                } else {
                    ge
                }
            }
            Bound::Unbounded => crate::skiplist::strip_mark(unsafe {
                (*self.index.head_ptr())
                    .tower(0)
                    .load(std::sync::atomic::Ordering::Acquire)
            }),
        }
    }

    /// DESC front: position at upper bound (start of descending iteration).
    fn resolve_front_rev(
        &self,
        bound: &Bound<&K>,
        guard: &seize::LocalGuard<'_>,
    ) -> *mut ConstNode<K, V, D::Loc> {
        match bound {
            Bound::Included(k) => self.index.find_first_ge(k.as_bytes(), guard),
            Bound::Excluded(k) => {
                let ge = self.index.find_first_ge(k.as_bytes(), guard);
                if !ge.is_null()
                    && !unsafe { &*ge }.is_marked()
                    && unsafe { &*ge }.key_bytes() == k.as_bytes()
                {
                    crate::skiplist::strip_mark(unsafe {
                        (*ge).tower(0).load(std::sync::atomic::Ordering::Acquire)
                    })
                } else {
                    ge
                }
            }
            Bound::Unbounded => crate::skiplist::strip_mark(unsafe {
                (*self.index.head_ptr())
                    .tower(0)
                    .load(std::sync::atomic::Ordering::Acquire)
            }),
        }
    }

    fn prefix_bounds(&self, prefix: &[u8]) -> (K, Bound<K>) {
        if self.reversed {
            let mut search = K::zeroed();
            search.as_bytes_mut().fill(0xFF);
            search.as_bytes_mut()[..prefix.len()].copy_from_slice(prefix);
            let mut end_key = K::zeroed();
            end_key.as_bytes_mut()[..prefix.len()].copy_from_slice(prefix);
            (search, Bound::Included(end_key))
        } else {
            let mut search = K::zeroed();
            search.as_bytes_mut()[..prefix.len()].copy_from_slice(prefix);
            let end = crate::scope::key_math::prefix_to_end_bound::<K>(prefix);
            (search, end)
        }
    }

    /// Iterate entries whose keys start with `prefix`.
    ///
    /// `reversed=true` (default): yields matching keys in DESC order.
    /// `next()` is O(1). The first reverse seek is expected O(log n); an m-row reverse walk is
    /// expected/amortized O(log n + m).
    pub fn prefix_iter(&self, prefix: &[u8]) -> ConstIter<'_, K, V, D::Loc> {
        if prefix.len() > size_of::<K>() {
            let guard = self.index.collector().enter();
            return ConstIter {
                list: &self.index,
                front: std::ptr::null_mut(),
                back: Some(std::ptr::null_mut()),
                back_finger: BackFinger::empty(),
                end: Bound::Unbounded,
                start: Bound::Unbounded,
                reversed: self.reversed,
                done: true,
                _guard: guard,
            };
        }
        let guard = self.index.collector().enter();
        let (search_key, end) = self.prefix_bounds(prefix);
        let front = self.index.find_first_ge(search_key.as_bytes(), &guard);
        ConstIter {
            list: &self.index,
            front,
            back: None,
            back_finger: BackFinger::empty(),
            end,
            start: Bound::Included(search_key),
            reversed: self.reversed,
            done: false,
            _guard: guard,
        }
    }

    /// Iterate all entries in index order.
    ///
    /// `reversed=true` (default): DESC. `reversed=false`: ASC.
    /// `next()` is O(1). The first reverse seek is expected O(log n); an m-row reverse walk is
    /// expected/amortized O(log n + m).
    pub fn iter(&self) -> ConstIter<'_, K, V, D::Loc> {
        let guard = self.index.collector().enter();
        let front = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        ConstIter {
            list: &self.index,
            front,
            back: None,
            back_finger: BackFinger::empty(),
            end: Bound::Unbounded,
            start: Bound::Unbounded,
            reversed: self.reversed,
            done: false,
            _guard: guard,
        }
    }

    /// Iterate entries in `[start, end)` — start inclusive, end exclusive.
    ///
    /// `reversed=true` (default): DESC within range. `reversed=false`: ASC.
    /// `next()` is O(1). The first reverse seek is expected O(log n); an m-row reverse walk is
    /// expected/amortized O(log n + m).
    pub fn range(&self, start: &K, end: &K) -> ConstIter<'_, K, V, D::Loc> {
        self.range_bounds(Bound::Included(start), Bound::Excluded(end))
    }

    /// Iterate entries in range defined by `start` and `end` bounds.
    ///
    /// Unlike [`range()`](Self::range), allows `Included`, `Excluded`, or `Unbounded`
    /// for each bound independently.
    ///
    /// `reversed=true` (default): DESC within range. `reversed=false`: ASC.
    /// `next()` is O(1). The first reverse seek is expected O(log n); an m-row reverse walk is
    /// expected/amortized O(log n + m).
    pub fn range_bounds(&self, start: Bound<&K>, end: Bound<&K>) -> ConstIter<'_, K, V, D::Loc> {
        let guard = self.index.collector().enter();
        if self.reversed {
            // reversed SkipList: [max → min]. Iteration yields DESC.
            // front = position near upper bound (end), back lazily resolved.
            let front = self.resolve_front_rev(&end, &guard);
            ConstIter {
                list: &self.index,
                front,
                back: None,
                back_finger: BackFinger::empty(),
                end: bound_owned(&start),
                start: bound_owned(&end),
                reversed: true,
                done: false,
                _guard: guard,
            }
        } else {
            let front = self.resolve_front_asc(&start, &guard);
            ConstIter {
                list: &self.index,
                front,
                back: None,
                back_finger: BackFinger::empty(),
                end: bound_owned(&end),
                start: bound_owned(&start),
                reversed: false,
                done: false,
                _guard: guard,
            }
        }
    }

    /// Key-only range iterator with explicit direction override.
    pub(crate) fn keys_range_bounds_dir(
        &self,
        start: Bound<&K>,
        end: Bound<&K>,
        dir: Dir,
    ) -> SkipKeyKeys<'_, K, ConstNode<K, V, D::Loc>> {
        let guard = self.index.collector().enter();
        let against = dir_from(self.reversed) != dir;
        let (front, iter_end, iter_start, storage_reversed) = if against {
            if self.reversed {
                (
                    std::ptr::null_mut(),
                    bound_owned(&start),
                    bound_owned(&end),
                    true,
                )
            } else {
                (
                    std::ptr::null_mut(),
                    bound_owned(&end),
                    bound_owned(&start),
                    false,
                )
            }
        } else if self.reversed {
            let front = self.resolve_front_rev(&end, &guard);
            (front, bound_owned(&start), bound_owned(&end), true)
        } else {
            let front = self.resolve_front_asc(&start, &guard);
            (front, bound_owned(&end), bound_owned(&start), false)
        };
        SkipKeyKeys::new(SkipKeyIter::bounded_dir(
            &self.index,
            iter_start,
            iter_end,
            front,
            against,
            storage_reversed,
            guard,
        ))
    }

    /// Range iterator with explicit direction override.
    ///
    /// When `dir` matches storage order ([`dir_from`](crate::map_index::dir_from)(`self.reversed`)),
    /// behavior is identical to [`range_bounds`](Self::range_bounds). When opposite, the first
    /// opposing-direction seek is expected O(log n); an m-row opposing-direction scan is
    /// expected/amortized O(log n + m).
    pub(crate) fn range_bounds_dir(
        &self,
        start: Bound<&K>,
        end: Bound<&K>,
        dir: Dir,
    ) -> DirRangeIter<ConstIter<'_, K, V, D::Loc>> {
        let iter = self.range_bounds(start, end);
        DirRangeIter::from_storage_match(iter, dir == dir_from(self.reversed))
    }

    pub fn len(&self) -> usize {
        self.index.len()
    }

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

    /// Whether index iteration yields descending key order.
    pub(crate) fn reversed(&self) -> bool {
        self.reversed
    }

    /// Field-granular prefix / range / page builder over this tree's key space.
    pub fn scope<const N: usize>(
        &self,
        key: &K,
    ) -> crate::scope::Scope<'_, crate::scope::TreeScan<&Self>, N> {
        crate::scope::tree_scope::<_, _, N>(self, key, dir_from(self.reversed()))
    }

    /// Iterate all entries and optionally mutate them. Call once at startup.
    ///
    /// The callback receives each (key, value) and returns `MigrateAction`:
    /// - `Keep` — no change (fires `on_init` if `NEEDS_INIT`)
    /// - `Update(new_value)` — replace value (fires `on_init`, hook-free write)
    /// - `Delete` — remove entry (hook-free delete, no `on_init`)
    ///
    /// `on_write` is **never** fired during migration.
    /// Returns the number of mutated entries.
    pub fn migrate(
        &self,
        f: impl Fn(&K, &[u8; V]) -> crate::MigrateAction<[u8; V]>,
    ) -> DbResult<usize> {
        self.migrate_inner(f, true)
    }

    pub(crate) fn migrate_inner(
        &self,
        f: impl Fn(&K, &[u8; V]) -> crate::MigrateAction<[u8; V]>,
        fire_init: bool,
    ) -> DbResult<usize> {
        use crate::MigrateAction;

        let _guard = self.index.collector().enter();
        let mut current = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        let mut count = 0;
        while !current.is_null() {
            let node = unsafe { &*current };
            current = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
            if node.is_marked() {
                continue;
            }
            let value = node.read_value();
            match f(&node.key, &value) {
                MigrateAction::Keep => {
                    if fire_init && H::NEEDS_INIT {
                        self.hook.on_init(&node.key, &value[..]);
                    }
                }
                MigrateAction::Update(value) => {
                    self.put_no_hook(&node.key, &value)?;
                    if fire_init && H::NEEDS_INIT {
                        self.hook.on_init(&node.key, &value[..]);
                    }
                    count += 1;
                }
                MigrateAction::Delete => {
                    self.delete_no_hook(&node.key)?;
                    count += 1;
                }
            }
        }

        tracing::info!(mutations = count, "const_tree migration complete");
        Ok(count)
    }

    /// Replay `on_init` for every live entry. Used when no migration runs.
    pub(crate) fn replay_init(&self) {
        if !H::NEEDS_INIT {
            return;
        }
        let _guard = self.index.collector().enter();
        let mut current = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        while !current.is_null() {
            let node = unsafe { &*current };
            current = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
            if !node.is_marked() {
                self.hook.on_init(&node.key, &node.read_value()[..]);
            }
        }
    }

    /// The budget one `project` call runs under: `min(hook budget, config
    /// budget)`. Every projection site — facade write, `atomic()`/TTL replay,
    /// transaction commit, migration reconcile — and the schema validator read
    /// this one accessor, so `validate`/`repair` can never evaluate a different
    /// contract than writes do.
    pub(crate) fn effective_projection_limits(&self) -> (usize, usize) {
        let (hook_rows, hook_bytes) = self.hook.projection_limits();
        (
            hook_rows.min(self.durability.max_projected_rows()),
            hook_bytes.min(self.durability.max_projected_bytes()),
        )
    }

    /// T10 Task 5: refuse a non-transactional write that would change this
    /// source's claim set, before any mutation. `H` here is the byte-level
    /// `WriteHook<K>` (`ZeroHookAdapter` for `ZeroTree`), so this is the
    /// byte-snapshot twin of `TypedTree::claim_guard` — same placement rule,
    /// same `H::CLAIMS` compile-time short-circuit, same `collection` source
    /// (`counters_dirty` marker bound unconditionally at managed open).
    #[allow(clippy::collapsible_if)]
    #[inline]
    fn claim_guard_bytes(&self, key: &K, old: Option<&[u8]>, new: Option<&[u8]>) -> DbResult<()> {
        #[cfg(feature = "armour")]
        if H::EMITS && !crate::txn_depth::in_stream_apply() {
            if let Some(targets) = self.hook.stream_targets() {
                let collection = self
                    .counter_dirty
                    .get()
                    .map(crate::counter_dirty::CounterDirty::name)
                    .unwrap_or("");
                return Err(DbError::ChangeStreamRequiresTx {
                    collection,
                    queue: targets.first_name(),
                });
            }
        }
        if !H::CLAIMS {
            return Ok(());
        }
        // F1 fix (review): `H::CLAIMS` is `true` for every `Projected`
        // carrier, including plain index/counter sources that never called
        // `.and_unique(...)` — check the runtime declaration before doing
        // anything else. Callers with a genuinely extra pre-read
        // (`upsert_locked`) gate on the same condition before reading.
        if self.hook.claim_targets().is_none() {
            return Ok(());
        }
        let collection = self
            .counter_dirty
            .get()
            .map(crate::counter_dirty::CounterDirty::name)
            .unwrap_or("");
        crate::unique::guard_nontx_write_bytes(
            collection,
            &self.hook,
            self.effective_projection_limits(),
            key,
            old,
            new,
        )
    }

    pub(crate) fn reconcile_projections(&self) -> DbResult<()> {
        if !H::PROJECTS {
            return Ok(());
        }
        let Some(targets) = self.hook.projection_targets() else {
            return Ok(());
        };
        let (max_rows, max_bytes) = self.effective_projection_limits();
        crate::projection::reconcile_carrier_projections(
            targets,
            |emit| {
                let mut err: Option<DbError> = None;
                let _guard = self.index.collector().enter();
                let mut current = crate::skiplist::strip_mark(unsafe {
                    (*self.index.head_ptr())
                        .tower(0)
                        .load(std::sync::atomic::Ordering::Acquire)
                });
                while !current.is_null() {
                    let node = unsafe { &*current };
                    current = crate::skiplist::strip_mark(
                        node.tower(0).load(std::sync::atomic::Ordering::Acquire),
                    );
                    if node.is_marked() {
                        continue;
                    }
                    if err.is_some() {
                        break;
                    }
                    let value = node.read_value();
                    let mut sink =
                        crate::projection::ProjectionSink::new(targets, max_rows, max_bytes);
                    match self
                        .hook
                        .project(&node.key, None, Some(&value[..]), &mut sink)
                        .and_then(|()| {
                            for row in sink.rows() {
                                emit(row.target, &row.key, &row.value)?;
                            }
                            Ok(())
                        }) {
                        Ok(()) => {}
                        Err(e) => err = Some(e),
                    }
                }
                err.map_or(Ok(()), Err)
            },
            true,
        )
    }

    /// Open-time / on-demand recount (T9 Task 8) — see `counter::recount` for
    /// the shared two-pass walk. No-op (`Ok(0)`) when this collection is not
    /// itself a counted source.
    pub(crate) fn reconcile_counters(&self) -> DbResult<u64> {
        if !H::COUNTS {
            return Ok(0);
        }
        let Some(targets) = self.hook.counter_targets() else {
            return Ok(0);
        };
        let (max_rows, max_bytes) = self.effective_projection_limits();
        crate::counter::recount(
            |k: &K, v: &[u8], sink| self.hook.counter_project(k, None, Some(v), sink),
            targets,
            (max_rows, max_bytes),
            &mut |visit| {
                let mut err: Option<DbError> = None;
                let _guard = self.index.collector().enter();
                let mut current = crate::skiplist::strip_mark(unsafe {
                    (*self.index.head_ptr())
                        .tower(0)
                        .load(std::sync::atomic::Ordering::Acquire)
                });
                while !current.is_null() {
                    let node = unsafe { &*current };
                    current = crate::skiplist::strip_mark(
                        node.tower(0).load(std::sync::atomic::Ordering::Acquire),
                    );
                    if node.is_marked() {
                        continue;
                    }
                    if err.is_some() {
                        break;
                    }
                    let value = node.read_value();
                    if let Err(e) = visit(&node.key, &value[..]) {
                        err = Some(e);
                    }
                }
                err.map_or(Ok(()), Err)
            },
        )
    }

    /// The collections whose `counters_dirty` flag this collection must also
    /// consult at open time (T9 Task 8) — one entry per declared counter
    /// target. Empty when this collection is not a counted source.
    pub(crate) fn counter_target_names(&self) -> Vec<&'static str> {
        self.hook
            .counter_targets()
            .map(|t| t.all().map(|m| m.collection_name()).collect())
            .unwrap_or_default()
    }

    pub(crate) fn index(&self) -> &SkipList<ConstNode<K, V, D::Loc>> {
        &self.index
    }

    pub fn shard_for(&self, key: &K) -> usize {
        if self.shard_prefix_bits == 0 || self.shard_prefix_bits >= size_of::<K>() * 8 {
            let hash = xxhash_rust::xxh3::xxh3_64(key.as_bytes());
            return (hash as usize) % self.durability.shard_count();
        }

        let full_bytes = self.shard_prefix_bits / 8;
        let extra_bits = self.shard_prefix_bits % 8;

        let hash = if extra_bits == 0 {
            xxhash_rust::xxh3::xxh3_64(&key.as_bytes()[..full_bytes])
        } else {
            let mut buf = K::zeroed();
            buf.as_bytes_mut()[..full_bytes].copy_from_slice(&key.as_bytes()[..full_bytes]);
            let mask = !((1u8 << (8 - extra_bits)) - 1);
            buf.as_bytes_mut()[full_bytes] = key.as_bytes()[full_bytes] & mask;
            xxhash_rust::xxh3::xxh3_64(&buf.as_bytes()[..full_bytes + 1])
        };

        (hash as usize) % self.durability.shard_count()
    }

    /// Flush the durability backend.
    pub fn flush(&self) -> DbResult<()> {
        self.durability.flush()
    }

    /// Batch point-read by a list of keys (tier-3 finger-walk). Lock-free, zero
    /// disk I/O. Result is aligned to `keys`: position `i` is `Some(value)` if
    /// `keys[i]` exists, else `None`. Not snapshot — a concurrent writer may be
    /// observed mid-walk. Without concurrent writers, duplicate keys yield the
    /// same value at each position.
    pub fn get_many(&self, keys: &[K]) -> Vec<Option<[u8; V]>> {
        let n = keys.len();
        let mut out: Vec<Option<[u8; V]>> = (0..n).map(|_| None).collect();
        if n == 0 {
            return out;
        }
        // Visit keys in list order so the finger only moves forward.
        let mut order: Vec<usize> = (0..n).collect();
        order.sort_by(|&a, &b| {
            let ord = keys[a].as_bytes().cmp(keys[b].as_bytes());
            if self.reversed { ord.reverse() } else { ord }
        });
        let guard = self.index.collector().enter();
        let mut finger = self.index.finger();
        for &i in &order {
            let kb = keys[i].as_bytes();
            if let Some(node) = self.index.finger_seek(&mut finger, kb, &guard)
                && node.key_bytes() == kb
            {
                out[i] = Some(node.read_value());
            }
        }
        out
    }

    /// Batch read-modify-write (tier-2, shard-grouped). The closure decides per key
    /// from `Option<&value>` (`None` = absent). Keys are grouped by shard and each
    /// shard is processed under its own lock, one at a time. Within a shard, items
    /// are applied in input order, so a later duplicate of a key sees the prior
    /// in-batch result (natural fold). Per-shard isolation only — no rollback, no
    /// crash-atomicity. On a real I/O `Err` the batch stops; the applied prefix is
    /// committed and its hooks fire. Result is aligned to `items`.
    pub fn update_many<P>(
        &self,
        items: Vec<(K, P)>,
        f: impl Fn(&K, Option<&[u8; V]>, &P) -> Upsert<[u8; V]>,
    ) -> DbResult<Vec<(K, Applied<[u8; V]>)>> {
        let n = items.len();
        let mut by_shard: std::collections::BTreeMap<usize, Vec<usize>> =
            std::collections::BTreeMap::new();
        for (i, (k, _)) in items.iter().enumerate() {
            by_shard.entry(self.shard_for(k)).or_default().push(i);
        }

        let mut out: Vec<Option<(K, Applied<[u8; V]>)>> = (0..n).map(|_| None).collect();
        #[allow(clippy::type_complexity)]
        let mut events: Vec<(K, Option<[u8; V]>, Option<[u8; V]>)> = Vec::new();
        let mut err: Option<DbError> = None;

        for (shard_id, idxs) in by_shard {
            let events_start = events.len();
            let mut applied_mutation = false;
            let mut wakes = WatchWakeBatch::default();
            let mut inner = self.durability.lock_shard(shard_id);
            let producer = self.watch_enter();
            let active = producer.active();
            let context_active = H::WATCHES.then(|| crate::watch::ContextActiveGuard::set(active));
            let guard = self.index.collector().enter();
            for &i in &idxs {
                #[cfg(test)]
                if UPDATE_MANY_FAIL_AFTER.with(|remaining| match remaining.get() {
                    0 => true,
                    usize::MAX => false,
                    value => {
                        remaining.set(value - 1);
                        false
                    }
                }) {
                    err = Some(DbError::Internal("test: update_many injected failure"));
                    break;
                }
                let (k, p) = &items[i];
                let cur = self
                    .index
                    .get(k.as_bytes(), &guard)
                    .map(|node| node.read_value());
                let applied = match f(k, cur.as_ref(), p) {
                    Upsert::Put(v) => match self.put_locked(shard_id, &mut *inner, &guard, k, &v) {
                        Ok(old) => {
                            if let Some(event) = self.defer_write_event(k, old, Some(v)) {
                                self.publish_watch(&event, active, &mut wakes);
                                events.push(Self::own(event));
                            }
                            Applied::Written { old, new: v }
                        }
                        Err(e) => {
                            err = Some(e);
                            break;
                        }
                    },
                    Upsert::Keep => Applied::Kept,
                    Upsert::Delete => {
                        match self.delete_locked(shard_id, &mut *inner, &guard, k) {
                            Ok(Some(old)) => {
                                if let Some(event) = self.defer_write_event(k, Some(old), None) {
                                    self.publish_watch(&event, active, &mut wakes);
                                    events.push(Self::own(event));
                                }
                                Applied::Deleted(old)
                            }
                            Ok(None) => Applied::Kept, // delete-miss: no write, no hook
                            Err(e) => {
                                err = Some(e);
                                break;
                            }
                        }
                    }
                };
                applied_mutation |=
                    matches!(applied, Applied::Written { .. } | Applied::Deleted(_));
                out[i] = Some((*k, applied));
            }
            let needs_sync = applied_mutation && inner.should_sync();
            drop(inner);
            drop(guard);
            drop(context_active);
            if H::NEEDS_WRITE || H::WATCHES {
                for (k, old, new) in &events[events_start..] {
                    self.hook.on_write(
                        k,
                        old.as_ref().map(|v| &v[..]),
                        new.as_ref().map(|v| &v[..]),
                    );
                }
            }
            let sync_err = if needs_sync {
                // Sync after the shard lock is released.
                self.durability.sync_shard(shard_id).err()
            } else {
                None
            };
            wakes.drain();
            match (err.take(), sync_err) {
                (Some(operation), Some(sync)) => {
                    tracing::error!(
                        teardown = %sync,
                        operation = %operation,
                        "sync failed during update_many teardown but the mutation error is returned"
                    );
                    return Err(operation);
                }
                (Some(operation), None) => return Err(operation),
                (None, Some(sync)) => return Err(sync),
                (None, None) => {}
            }
        }

        Ok(out
            .into_iter()
            .map(|slot| slot.expect("every index assigned on the success path"))
            .collect())
    }

    #[cfg(test)]
    pub(crate) fn hook_ref(&self) -> &H {
        &self.hook
    }
}

// ==========================================================================
// Replication (Bitcask only — uses DiskLoc)
// ==========================================================================

#[cfg(feature = "replication")]
impl<K: Key, const V: usize, H: WriteHook<K>> crate::replication::ReplicationTarget
    for ConstTree<K, V, H, Bitcask>
{
    fn apply_entry(
        &self,
        _shard_inner: &mut crate::shard::ShardInner,
        _shard_id: u8,
        file_id: u32,
        entry_offset: u64,
        header: &crate::entry::EntryHeader,
        key: &[u8],
        value: &[u8],
    ) -> DbResult<crate::replication::ApplyOutcome> {
        use crate::replication::ApplyOutcome;

        let key: K = K::from_bytes(key);

        let value_offset =
            entry_offset + size_of::<crate::entry::EntryHeader>() as u64 + size_of::<K>() as u64;
        let disk = DiskLoc::new(file_id, value_offset as u32, header.value_len);

        if header.is_tombstone() {
            let guard = self.index.collector().enter();
            let removed = self.index.remove(key.as_bytes(), &guard);
            match removed {
                Some(node_ptr) => {
                    let old_disk = unsafe { &*node_ptr }.read_loc();
                    Ok(ApplyOutcome::TombstoneRemoved(old_disk))
                }
                None => Ok(ApplyOutcome::Inserted), // no-op tombstone — no dead bytes
            }
        } else {
            let value: [u8; V] = value.try_into().map_err(|_| DbError::CorruptedEntry {
                offset: entry_offset,
            })?;
            let guard = self.index.collector().enter();
            let height = random_height();
            let node_ptr = ConstNode::alloc(key, value, disk, height);
            match self.index.insert(node_ptr, &guard) {
                InsertResult::Inserted => Ok(ApplyOutcome::Inserted),
                InsertResult::Exists(existing) => {
                    let old_disk = existing.read_loc();
                    existing.write_data(disk, &value);
                    unsafe {
                        ConstNode::<K, V>::dealloc_node(node_ptr);
                    }
                    Ok(ApplyOutcome::Replaced(old_disk))
                }
            }
        }
    }

    fn try_apply_entry(
        &self,
        shard_inner: &mut crate::shard::ShardInner,
        shard_id: u8,
        file_id: u32,
        entry_offset: u64,
        header: &crate::entry::EntryHeader,
        raw_after_header: &[u8],
    ) -> DbResult<crate::replication::ApplyOutcome> {
        use crate::replication::ApplyOutcome;

        if raw_after_header.len() < size_of::<K>() + header.value_len as usize {
            return Ok(ApplyOutcome::NotMatched);
        }
        let key = &raw_after_header[..size_of::<K>()];
        let value = &raw_after_header[size_of::<K>()..size_of::<K>() + header.value_len as usize];
        let crc = crate::entry::compute_crc32(header.gsn, header.value_len, key, value);
        if crc != header.crc32 {
            return Ok(ApplyOutcome::NotMatched);
        }
        self.apply_entry(
            shard_inner,
            shard_id,
            file_id,
            entry_offset,
            header,
            key,
            value,
        )
    }

    fn key_len(&self) -> usize {
        size_of::<K>()
    }

    fn mark_replication_follower(&self) {
        ConstTree::mark_replication_follower(self);
    }
}

#[cfg(feature = "replication")]
impl<K: Key, const V: usize, H: WriteHook<K>> ConstTree<K, V, H, Bitcask> {
    /// Install SPSC replication producers into every shard and start a
    /// `ReplicationServer` bound to `bind_addr`.
    ///
    /// # Single-call contract
    ///
    /// Each call installs fresh SPSC producers, replacing any previously
    /// installed ones. Call this at most once per `ConstTree` instance — a
    /// second call will orphan the in-flight producer of any active streaming
    /// connection on the first server, which will then observe an empty ring
    /// buffer and silently stop forwarding entries.
    pub fn start_replication_server(
        &self,
        bind_addr: std::net::SocketAddr,
        signal: crate::shutdown::ShutdownSignal,
    ) -> crate::error::DbResult<crate::replication::ReplicationServer> {
        let consumers = self.install_replication_producers()?;
        crate::replication::ReplicationServer::start(
            bind_addr,
            self.durability.engine.shards().clone(),
            consumers,
            self.durability.engine.config().max_file_size,
            signal,
        )
    }

    fn install_replication_producers(
        &self,
    ) -> crate::error::DbResult<Vec<rtrb::Consumer<crate::replication::ReplicationEntry>>> {
        const SPSC_CAPACITY: usize = 4096;
        let shards = self.durability.engine.shards();
        let mut consumers = Vec::with_capacity(shards.len());
        for shard in shards.iter() {
            let (p, c) = rtrb::RingBuffer::new(SPSC_CAPACITY);
            shard.set_replication_producer(p);
            consumers.push(c);
        }
        Ok(consumers)
    }

    /// Start a `ReplicationClient` that streams entries from `leader_addr`
    /// into `registry`. Symmetric to [`Self::start_replication_server`].
    ///
    /// `key_len` is derived from `size_of::<K>()` (the `V` const-generic is
    /// the value byte length, not part of the key).
    pub fn start_replication_client(
        &self,
        leader_addr: std::net::SocketAddr,
        registry: std::sync::Arc<crate::replication::ReplicationRegistry>,
        signal: crate::shutdown::ShutdownSignal,
    ) -> crate::error::DbResult<crate::replication::ReplicationClient> {
        crate::replication::ReplicationClient::start(
            leader_addr,
            self.durability.engine.shards().clone(),
            registry,
            size_of::<K>() as u16,
            signal,
        )
    }
}

#[cfg(feature = "replication")]
impl<K, const V: usize, H> ConstTree<K, V, H, Bitcask>
where
    K: Key + Send + Sync + 'static,
    H: WriteHook<K> + Send + Sync + 'static,
{
    /// Wrap a shared handle to this tree as a `Box<dyn ReplicationTarget>`.
    ///
    /// The returned box holds an `Arc` clone — the caller retains full read
    /// access to the original tree through the `Arc` while the registry owns
    /// the box. This is the intended pattern for follower-side wiring:
    ///
    /// ```ignore
    /// let follower = Arc::new(ConstTree::<[u8; 8], 8>::open(path, cfg)?);
    /// let registry = ReplicationRegistry::new(follower.as_replication_target());
    /// // `follower` remains usable for .get() etc.
    /// ```
    pub fn as_replication_target(
        self: &std::sync::Arc<Self>,
    ) -> Box<dyn crate::replication::ReplicationTarget> {
        Box::new(std::sync::Arc::clone(self))
    }
}

// ==========================================================================
// ConstShard — generic over D: Durability
// ==========================================================================

/// Handle for atomic multi-key operations within a single shard.
/// Obtained via [`ConstTree::atomic`]. The shard lock is held for the
/// lifetime of this struct — keep the closure short.
pub struct ConstShard<'a, K: Key, const V: usize, H: WriteHook<K> = NoHook, D: Durability = Bitcask>
{
    tree: &'a ConstTree<K, V, H, D>,
    pub(crate) held: Held<'a, ConstTree<K, V, H, D>>,
    guard: seize::LocalGuard<'a>,
}

impl<K: Key, const V: usize, H: WriteHook<K>, D: Durability> ConstShard<'_, K, V, H, D> {
    pub fn put(&mut self, key: &K, value: &[u8; V]) -> DbResult<Option<[u8; V]>> {
        self.check_shard(key)?;
        let tree = self.tree;
        let sid = self.held.sid;
        let guard = &self.guard;
        Ctx::with_shard(&mut self.held, tree, sid, |locked, ev_out| {
            let old = tree.put_locked(sid, &mut *locked, guard, key, value)?;
            *ev_out = tree.defer_write_event(key, old, Some(*value));
            Ok(old)
        })
    }

    pub fn insert(&mut self, key: &K, value: &[u8; V]) -> DbResult<()> {
        self.check_shard(key)?;
        let tree = self.tree;
        let sid = self.held.sid;
        let guard = &self.guard;
        Ctx::with_shard(&mut self.held, tree, sid, |locked, ev_out| {
            tree.insert_locked(sid, &mut *locked, guard, key, value)?;
            *ev_out = tree.defer_write_event(key, None, Some(*value));
            Ok(())
        })
    }

    pub fn delete(&mut self, key: &K) -> DbResult<Option<[u8; V]>> {
        self.check_shard(key)?;
        let tree = self.tree;
        let sid = self.held.sid;
        let guard = &self.guard;
        Ctx::with_shard(&mut self.held, tree, sid, |locked, ev_out| {
            let old = tree.delete_locked(sid, &mut *locked, guard, key)?;
            let ev = match &old {
                Some(o) => tree.defer_write_event(key, Some(*o), None),
                None => None,
            };
            *ev_out = ev;
            Ok(old)
        })
    }

    /// Read-modify-write within this shard: applies `f` to the current value and
    /// stores the result. Returns `Some(new_value)` if the key existed, `None`
    /// otherwise (closure not invoked). The closure must not be heavy — the shard
    /// lock is held.
    pub fn update(
        &mut self,
        key: &K,
        f: impl FnOnce(&[u8; V]) -> [u8; V],
    ) -> DbResult<Option<[u8; V]>> {
        self.update_inner(key, f, false)
    }

    /// Like [`update()`](Self::update), but returns `Some(old_value)` instead of
    /// the new one.
    pub fn fetch_update(
        &mut self,
        key: &K,
        f: impl FnOnce(&[u8; V]) -> [u8; V],
    ) -> DbResult<Option<[u8; V]>> {
        self.update_inner(key, f, true)
    }

    /// Decide-and-write within this shard. The closure must be short — shard lock is held.
    pub fn upsert<U: crate::IntoUpsert<[u8; V]>>(
        &mut self,
        key: &K,
        f: impl FnOnce(Option<&[u8; V]>) -> U,
    ) -> DbResult<U::Out> {
        self.check_shard(key)?;
        let tree = self.tree;
        let sid = self.held.sid;
        let guard = &self.guard;
        Ctx::with_shard(&mut self.held, tree, sid, |locked, ev_out| {
            let existing = tree.index.get(key.as_bytes(), guard);
            let cur = existing.map(|n| n.read_value());
            let (decision, out) = f(cur.as_ref()).into_parts();
            if let Some((old, new)) =
                tree.upsert_locked(sid, &mut *locked, guard, key, decision, existing)?
            {
                *ev_out = tree.defer_write_event(key, old, new);
            }
            Ok(out)
        })
    }

    /// Generic [`upsert_via_held`] path — retained for agreement tests against the fused body.
    ///
    /// # Unstable
    /// Not part of the public API: it is `pub` only because `tests/upsert_agreement.rs`
    /// is an integration test. May change or disappear in any release — call
    /// [`upsert`](Self::upsert).
    #[doc(hidden)]
    pub fn upsert_generic<U: crate::IntoUpsert<[u8; V]>>(
        &mut self,
        key: &K,
        f: impl FnOnce(Option<&[u8; V]>) -> U,
    ) -> DbResult<U::Out> {
        crate::upsert::upsert_via_held(self, key, f)
    }

    pub fn try_get(&self, key: &K) -> DbResult<Option<[u8; V]>> {
        self.check_shard(key)?;
        match self.tree.index.get(key.as_bytes(), &self.guard) {
            Some(node) => Ok(Some(node.read_value())),
            None => Ok(None),
        }
    }

    fn update_inner(
        &mut self,
        key: &K,
        f: impl FnOnce(&[u8; V]) -> [u8; V],
        return_old: bool,
    ) -> DbResult<Option<[u8; V]>> {
        self.check_shard(key)?;
        let tree = self.tree;
        let sid = self.held.sid;
        let guard = &self.guard;
        Ctx::with_shard(&mut self.held, tree, sid, |locked, ev_out| {
            let (old_value, new_value) =
                match tree.update_locked(sid, &mut *locked, guard, key, f)? {
                    Some(pair) => pair,
                    None => return Ok(None),
                };
            let result = Some(if return_old { old_value } else { new_value });
            *ev_out = tree.defer_write_event(key, Some(old_value), Some(new_value));
            Ok(result)
        })
    }

    pub fn get_or_err(&self, key: &K) -> DbResult<[u8; V]> {
        self.try_get(key)?.ok_or(DbError::KeyNotFound)
    }

    pub fn try_contains(&self, key: &K) -> DbResult<bool> {
        self.check_shard(key)?;
        Ok(self.tree.index.get(key.as_bytes(), &self.guard).is_some())
    }

    fn check_shard(&self, key: &K) -> DbResult<()> {
        if self.tree.shard_for(key) != self.held.sid {
            return Err(DbError::ShardMismatch);
        }
        Ok(())
    }
}

impl<K: Key, const V: usize, H: WriteHook<K>> ConstShard<'_, K, V, H, Bitcask> {
    pub(crate) fn tx_arm_gsn(&mut self, gsn: u64) {
        self.held.locked.arm_gsn(gsn);
    }

    pub(crate) fn tx_clear_armed_gsn(&mut self) {
        self.held.locked.clear_armed_gsn();
    }

    /// Refuse further appends on this shard: a committed transaction operation
    /// reserved `gsn` and never wrote it, so only replay can complete it.
    /// Returns the effective fence (an earlier, lower one wins).
    pub(crate) fn tx_fence(&mut self, gsn: u64) -> u64 {
        self.held.locked.fence(gsn)
    }

    /// Hook events buffered so far. The transaction reads it around every
    /// mutation to learn how many events that mutation produced — a write that
    /// changed nothing produces none.
    pub(crate) fn tx_buffered_events(&self) -> usize {
        self.held.events.len()
    }
}

impl<K: Key, const V: usize, H: WriteHook<K>> ConstShard<'_, K, V, H, Fixed> {
    pub(crate) fn tx_buffered_events(&self) -> usize {
        self.held.events.len()
    }
}

// ==========================================================================
// Helper functions
// ==========================================================================

fn bound_owned<K: Copy>(b: &Bound<&K>) -> Bound<K> {
    match b {
        Bound::Included(k) => Bound::Included(**k),
        Bound::Excluded(k) => Bound::Excluded(**k),
        Bound::Unbounded => Bound::Unbounded,
    }
}

// ==========================================================================
// ConstIter — generic over L: Location
// ==========================================================================

/// Iterator over entries in a `ConstTree`. Returned by `iter()`, `range()`, and `prefix_iter()`.
///
/// Weakly-consistent: concurrent inserts/updates may be visible during iteration.
/// Deleted entries are skipped. The `seize` guard prevents use-after-free.
pub struct ConstIter<'a, K: Key, const V: usize, L: Location = DiskLoc> {
    list: &'a SkipList<ConstNode<K, V, L>>,
    front: *mut ConstNode<K, V, L>,
    /// `None` = not yet resolved (lazy). Computed on first `next_back()` call.
    back: Option<*mut ConstNode<K, V, L>>,
    back_finger: BackFinger<ConstNode<K, V, L>>,
    end: Bound<K>,
    start: Bound<K>,
    reversed: bool,
    done: bool,
    _guard: seize::LocalGuard<'a>,
}

impl<K: Key, const V: usize, L: Location> Iterator for ConstIter<'_, K, V, L> {
    type Item = (K, [u8; V]);

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            if self.done || self.front.is_null() {
                return None;
            }
            let node = unsafe { &*self.front };
            let converged = self.back.is_some_and(|back| std::ptr::eq(self.front, back));
            self.front = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
            if converged {
                self.done = true;
            }
            if node.is_marked() {
                if converged {
                    return None;
                }
                continue;
            }
            if !self.check_end(&node.key) {
                self.done = true;
                return None;
            }
            return Some((node.key, node.read_value()));
        }
    }
}

impl<K: Key, const V: usize, L: Location> DoubleEndedIterator for ConstIter<'_, K, V, L> {
    fn next_back(&mut self) -> Option<Self::Item> {
        if self.back.is_none() {
            self.back = Some(self.resolve_back());
            // If front already reached the end of the list, everything is consumed.
            if self.front.is_null() {
                self.done = true;
            }
        }
        loop {
            let back = self.back.unwrap_or(std::ptr::null_mut());
            if self.done || back.is_null() {
                return None;
            }
            let node = unsafe { &*back };
            let key = node.key;
            let converged = std::ptr::eq(self.front, back);
            self.back = Some(self.list.back_step(
                &mut self.back_finger,
                key.as_bytes(),
                &self._guard,
            ));
            if converged {
                self.done = true;
            }
            if node.is_marked() {
                if converged {
                    return None;
                }
                continue;
            }
            if !self.check_start(&key) {
                self.done = true;
                return None;
            }
            return Some((key, node.read_value()));
        }
    }
}

impl<K: Key, const V: usize, L: Location> ConstIter<'_, K, V, L> {
    /// Lazily resolve the back pointer for DoubleEndedIterator.
    /// Uses `end` bound — the limit for forward iteration — to find the
    /// starting position for backward iteration.
    fn resolve_back(&mut self) -> *mut ConstNode<K, V, L> {
        match &self.end {
            Bound::Unbounded => self
                .list
                .back_seek_last(&mut self.back_finger, &self._guard),
            Bound::Excluded(k) => {
                self.list
                    .back_step(&mut self.back_finger, k.as_bytes(), &self._guard)
            }
            Bound::Included(k) => {
                let ge = self.list.find_first_ge(k.as_bytes(), &self._guard);
                if !ge.is_null()
                    && !unsafe { &*ge }.is_marked()
                    && unsafe { &*ge }.key_bytes() == k.as_bytes()
                {
                    ge
                } else {
                    self.list
                        .back_step(&mut self.back_finger, k.as_bytes(), &self._guard)
                }
            }
        }
    }

    /// Check if key is within the end bound (forward direction).
    #[inline(always)]
    fn check_end(&self, key: &K) -> bool {
        match &self.end {
            Bound::Unbounded => true,
            Bound::Excluded(end) => {
                if self.reversed {
                    key.as_bytes() > end.as_bytes()
                } else {
                    key.as_bytes() < end.as_bytes()
                }
            }
            Bound::Included(end) => {
                if self.reversed {
                    key.as_bytes() >= end.as_bytes()
                } else {
                    key.as_bytes() <= end.as_bytes()
                }
            }
        }
    }

    /// Check if key is within the start bound (backward direction).
    #[inline(always)]
    fn check_start(&self, key: &K) -> bool {
        match &self.start {
            Bound::Unbounded => true,
            Bound::Excluded(s) => {
                if self.reversed {
                    key.as_bytes() < s.as_bytes()
                } else {
                    key.as_bytes() > s.as_bytes()
                }
            }
            Bound::Included(s) => {
                if self.reversed {
                    key.as_bytes() <= s.as_bytes()
                } else {
                    key.as_bytes() >= s.as_bytes()
                }
            }
        }
    }
}

impl<K: Key, const V: usize, L: Location> ConstIter<'_, K, V, L> {
    /// Collect all remaining entries. Convenience for backward compatibility.
    pub fn collect_vec(&mut self) -> Vec<(K, [u8; V])> {
        self.collect()
    }
}

// ==========================================================================
// Tests
// ==========================================================================

#[cfg(test)]
mod tests {
    use super::{ConstTree, UpdateManyFailAfter};
    use crate::durability::Durability;
    use crate::fixed::FixedTree;
    use crate::flush_test::{
        FAIL_NEXT_DATA_SYNC, data_attempts, data_successes, hold_data_sync, reset,
    };
    use crate::hook::WriteHook;
    use crate::map_index::{Dir, dir_from};
    use crate::{Config, DbError, DbResult, FixedConfig};
    use std::ops::Bound;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};
    use tempfile::tempdir;

    #[test]
    fn open_installs_file_min_after_recovering_rotated_hint() {
        let dir = tempdir().unwrap();
        let mut config = Config::test();
        config.shard_count = 1;
        config.max_file_size = 4096;
        config.write_buffer_size = 4096;
        {
            let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config.clone()).unwrap();
            for key in 0u64..200 {
                tree.put(&key.to_be_bytes(), &[7; 8]).unwrap();
            }
            tree.close().unwrap();
        }
        let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), config).unwrap();
        let inner = tree.durability.engine.shards()[0].lock();
        assert!(matches!(
            inner.active.seqs.min,
            crate::shard::MinSeq::Known(_)
        ));
        assert!(
            inner
                .immutable
                .iter()
                .all(|file| { matches!(file.seqs.min, crate::shard::MinSeq::Known(_)) })
        );
    }

    #[derive(Default)]
    struct RecHook {
        writes: AtomicUsize,
        #[allow(clippy::type_complexity)]
        seq: crate::sync::Mutex<Vec<(u64, Option<Vec<u8>>, Option<Vec<u8>>)>>,
    }
    impl WriteHook<[u8; 8]> for RecHook {
        const NEEDS_OLD_VALUE: bool = true;
        fn on_write(&self, key: &[u8; 8], old: Option<&[u8]>, new: Option<&[u8]>) {
            self.writes.fetch_add(1, AtomicOrdering::Relaxed);
            crate::sync::lock(&self.seq).push((
                u64::from_be_bytes(*key),
                old.map(<[u8]>::to_vec),
                new.map(<[u8]>::to_vec),
            ));
        }
    }

    fn open_const_hooked(dir: &std::path::Path, hook: RecHook) -> ConstTree<[u8; 8], 4, RecHook> {
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        ConstTree::open_hooked(dir, cfg, hook).expect("open hooked")
    }

    /// Regression (bug 26-06-18): Bitcask compaction must reclaim dead bytes
    /// under pure overwrite churn. Before the liveness-skip fix the scan copied
    /// every superseded entry into the new file verbatim, so on-disk size tracked
    /// total bytes written (~41x the live set here) and `compact()` reclaimed
    /// nothing no matter how many times it ran.
    #[test]
    fn compaction_reclaims_dead_bytes_under_overwrite_churn() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 64 * 1024; // small files → many rotations
        cfg.write_buffer_size = 16 * 1024;
        cfg.compaction_threshold = 0.30; // default
        let max_file_size = cfg.max_file_size;

        let tree = ConstTree::<[u8; 8], 64>::open(dir.path(), cfg).unwrap();

        const N: u64 = 500; // working set
        const M: u64 = 40; // overwrites per key
        let entry_sz: u64 = 16 + 8 + 64; // header + key + value
        let live_set = N * entry_sz;
        let total_written = N * (M + 1) * entry_sz;

        for k in 0..N {
            tree.put(&k.to_be_bytes(), &[0u8; 64]).unwrap();
        }
        for round in 1..=M {
            let mut v = [0u8; 64];
            v[0] = round as u8;
            for k in 0..N {
                tree.put(&k.to_be_bytes(), &v).unwrap();
            }
        }
        tree.flush_buffers().unwrap();

        let shard = &tree.durability.engine.shards()[0];
        let disk = || {
            let inner = shard.lock();
            inner.active.write_offset + inner.immutable.iter().map(|f| f.total_bytes).sum::<u64>()
        };

        // Churn really did inflate the file set far past the live set.
        assert!(
            disk() > total_written / 2,
            "expected dead-byte buildup before compaction: disk={} total_written={total_written}",
            disk()
        );

        // A single pass must reclaim something (the truncate(4) batch of dead files).
        let before = disk();
        tree.compact().unwrap();
        assert!(
            disk() < before,
            "one compaction pass reclaimed nothing: {before} -> {}",
            disk()
        );

        // Compact to convergence; on-disk size must collapse toward the live set,
        // not the total bytes written. (The per-call truncate(4) cap means several
        // passes are needed — that throttle is a separate, secondary concern.)
        for _ in 0..200 {
            let before = disk();
            tree.compact().unwrap();
            if disk() == before {
                break;
            }
        }
        assert!(
            disk() <= live_set + max_file_size,
            "compaction failed to reclaim dead bytes: final disk={} live_set={live_set} \
             total_written={total_written}",
            disk()
        );

        // Data must remain correct and complete after all the rewrites.
        let mut v = [0u8; 64];
        v[0] = M as u8;
        assert_eq!(tree.get(&0u64.to_be_bytes()), Some(v));
        assert_eq!(tree.get(&(N - 1).to_be_bytes()), Some(v));
        assert_eq!(tree.len(), N as usize);
    }

    /// Regression (bug 26-06-18, continuous-write gap): dead-byte accounting on
    /// IMMUTABLE files must reach the selection threshold WITHOUT a preceding
    /// flush. The quiesced reclaim test flushes before compacting; the benchmark
    /// never does, which seeded the (wrong) hypothesis that supersessions don't
    /// credit dead bytes until flush. They do — `write_update` calls
    /// `add_dead_bytes` immediately — so after pure overwrite churn every rotated
    /// immutable file is ~100% dead and would be selected even with no flush.
    #[test]
    fn dead_bytes_credited_on_immutable_files_without_flush() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 64 * 1024;
        cfg.write_buffer_size = 16 * 1024;
        cfg.compaction_threshold = 0.30;
        let threshold = cfg.compaction_threshold;

        let tree = ConstTree::<[u8; 8], 64>::open(dir.path(), cfg).unwrap();

        const N: u64 = 500;
        const M: u64 = 40;

        for k in 0..N {
            tree.put(&k.to_be_bytes(), &[0u8; 64]).unwrap();
        }
        for round in 1..=M {
            let mut v = [0u8; 64];
            v[0] = round as u8;
            for k in 0..N {
                tree.put(&k.to_be_bytes(), &v).unwrap();
            }
        }
        // Deliberately NO flush_buffers() here — mirror the benchmark's regime.

        let shard = &tree.durability.engine.shards()[0];
        let inner = shard.lock();
        assert!(
            !inner.immutable.is_empty(),
            "churn should have rotated immutable files"
        );
        let mut crossed = 0usize;
        for f in &inner.immutable {
            let dead = inner.dead_bytes.get(&f.file_id).copied().unwrap_or(0);
            if f.total_bytes > 0 && dead as f64 / f.total_bytes as f64 > threshold {
                crossed += 1;
            }
        }
        assert_eq!(
            crossed,
            inner.immutable.len(),
            "every rotated immutable file must cross the {threshold} dead-ratio \
             threshold WITHOUT a flush (dead bytes are credited at write_update \
             time, not at flush): {crossed}/{} crossed",
            inner.immutable.len(),
        );
    }

    /// Regression (bug 26-06-18, continuous-write gap): `compaction_max_files_per_pass`
    /// bounds how many immutable files a single compact() pass retires. The
    /// default cap is the rate limit that stops compaction from keeping up with a
    /// continuous writer in the benchmark; setting it to 0 (unlimited) lets ONE
    /// pass collapse a full backlog of dead files down to the live set. Note that
    /// compact()'s return value is the count of LIVE entries copied forward, NOT
    /// bytes reclaimed — under pure overwrite the oldest files are 100% dead and
    /// copy 0 live entries even as space is freed.
    #[test]
    fn compaction_max_files_per_pass_controls_backlog_drain() {
        const N: u64 = 500;
        const M: u64 = 200; // heavy churn → large backlog of dead files
        let entry_sz: u64 = 16 + 8 + 64;
        let live_set = N * entry_sz;

        let build = |cap: usize| {
            let dir = tempdir().unwrap();
            let mut cfg = Config::test();
            cfg.shard_count = 1;
            cfg.max_file_size = 64 * 1024;
            cfg.write_buffer_size = 16 * 1024;
            cfg.compaction_threshold = 0.30;
            cfg.compaction_max_files_per_pass = cap;
            let tree = ConstTree::<[u8; 8], 64>::open(dir.path(), cfg).unwrap();
            for k in 0..N {
                tree.put(&k.to_be_bytes(), &[0u8; 64]).unwrap();
            }
            for round in 1..=M {
                let mut v = [0u8; 64];
                v[0] = round as u8;
                for k in 0..N {
                    tree.put(&k.to_be_bytes(), &v).unwrap();
                }
            }
            tree.flush_buffers().unwrap();
            (dir, tree)
        };

        let disk = |tree: &ConstTree<[u8; 8], 64>| {
            let inner = tree.durability.engine.shards()[0].lock();
            inner.active.write_offset + inner.immutable.iter().map(|f| f.total_bytes).sum::<u64>()
        };

        // Default cap (4): one pass barely dents the backlog — disk stays bloated.
        let default_cap = Config::test().compaction_max_files_per_pass;
        assert!(default_cap > 0, "default cap must be a finite throttle");
        let (_dir_d, tree_d) = build(default_cap);
        let before_d = tree_d.durability.engine.shards()[0].lock().immutable.len();
        tree_d.compact().unwrap();
        let after_d = tree_d.durability.engine.shards()[0].lock().immutable.len();
        assert!(
            before_d.saturating_sub(after_d) <= default_cap,
            "default-capped pass retired more than {default_cap} files: {before_d} -> {after_d}"
        );
        assert!(
            disk(&tree_d) > 10 * live_set,
            "one default-capped pass should NOT collapse the backlog: disk={} live_set={live_set}",
            disk(&tree_d)
        );

        // Unlimited (0): a single pass collapses the whole backlog to the live set.
        let (_dir_u, tree_u) = build(0);
        tree_u.compact().unwrap();
        assert!(
            disk(&tree_u) <= live_set + 64 * 1024,
            "unlimited pass must drain the backlog in one call: disk={} live_set={live_set}",
            disk(&tree_u)
        );

        // Data stays correct under the unlimited drain.
        let mut v = [0u8; 64];
        v[0] = M as u8;
        assert_eq!(tree_u.get(&0u64.to_be_bytes()), Some(v));
        assert_eq!(tree_u.get(&(N - 1).to_be_bytes()), Some(v));
        assert_eq!(tree_u.len(), N as usize);
    }

    /// Regression (codex review P1): the compaction scan must CRC-verify each
    /// entry *before* using its key for the liveness skip. A live entry whose key
    /// bytes are corrupted on disk would otherwise be parsed into the wrong key,
    /// classified as superseded, skipped, and then silently lost when the source
    /// file is removed. It must instead abort with `CrcMismatch`, leaving the
    /// source file intact.
    #[test]
    fn compaction_aborts_on_corrupt_key_of_live_entry() {
        use std::os::unix::fs::FileExt;

        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.compaction_threshold = 0.0; // select any file with dead bytes
        let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), cfg).unwrap();

        // `victim` is the live, soon-to-be-corrupted entry (written first, so it
        // sits at offset 0). `other` is overwritten once to give the file a
        // non-zero dead-byte ratio so compaction selects it at threshold 0.0.
        let victim = 1u64.to_be_bytes();
        let victim_val = 7u64.to_be_bytes();
        let other = 2u64.to_be_bytes();
        tree.put(&victim, &victim_val).unwrap();
        tree.put(&other, &9u64.to_be_bytes()).unwrap();
        tree.put(&other, &10u64.to_be_bytes()).unwrap(); // dead bytes on file 1

        // Roll the active file (holding all three entries) to immutable.
        let shard = &tree.durability.engine.shards()[0];
        shard.rotate_active_for_test(8).unwrap();

        // Corrupt the first key byte of `victim` on disk. The in-memory index
        // still maps `victim` -> (file 1, offset) so the entry is live; only the
        // on-disk bytes are damaged. Layout: [header(16) | key(8) | value(8)].
        let data_path = dir.path().join("shard_000").join("000001.data");
        let f = std::fs::OpenOptions::new()
            .write(true)
            .open(&data_path)
            .unwrap();
        f.write_at(&[0xFF], 16).unwrap();
        f.sync_all().unwrap();
        drop(f);

        let res = tree.compact();
        assert!(
            matches!(res, Err(crate::DbError::CrcMismatch { .. })),
            "expected CrcMismatch, got {res:?}"
        );
        assert!(
            data_path.exists(),
            "source file must be preserved when compaction aborts on corruption"
        );
        // The index is intact, so the value is still served (zero-I/O read).
        assert_eq!(tree.get(&victim), Some(victim_val));
    }

    /// Scan the durable on-disk state of a shard (the `.data` files, excluding the
    /// in-memory write buffer) and return the last value stored for `key`, if any.
    /// Mirrors what a full-scan recovery after a process crash would observe.
    fn scan_disk_for_key(shard_dir: &std::path::Path, key: &[u8; 8]) -> Option<[u8; 8]> {
        use crate::entry::EntryHeader;
        use zerocopy::FromBytes;
        let hsz = std::mem::size_of::<EntryHeader>();
        let mut ids: Vec<u32> = std::fs::read_dir(shard_dir)
            .unwrap()
            .filter_map(|e| {
                let n = e.ok()?.file_name().to_string_lossy().into_owned();
                n.strip_suffix(".data")?.parse::<u32>().ok()
            })
            .collect();
        ids.sort_unstable();
        let mut found = None;
        for id in ids {
            let bytes = std::fs::read(shard_dir.join(format!("{id:06}.data"))).unwrap();
            let mut off = 0usize;
            while off + hsz <= bytes.len() {
                let header = match EntryHeader::read_from_bytes(&bytes[off..off + hsz]) {
                    Ok(h) => h,
                    Err(_) => break,
                };
                if header.gsn == 0 && header.crc32 == 0 && header.value_len == 0 {
                    break; // zero padding / EOF
                }
                let total = hsz + 8 + header.value_len as usize;
                if off + total > bytes.len() {
                    break;
                }
                if &bytes[off + hsz..off + hsz + 8] == key {
                    found = if header.is_tombstone() {
                        None
                    } else {
                        Some(bytes[off + hsz + 8..off + total].try_into().unwrap())
                    };
                }
                off += total;
            }
        }
        found
    }

    /// Regression (codex review P1): compaction must not delete the last durable
    /// copy of a key while the superseding overwrite is still in the (unflushed)
    /// write buffer. `ConstTree` defaults to `hints = false`, so recovery does a
    /// full `.data` scan — the live key must therefore have a durable on-disk
    /// entry after compaction, or a crash before the buffer flush loses it.
    #[test]
    fn compaction_preserves_durable_value_when_overwrite_unflushed() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.compaction_threshold = 0.0;
        let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), cfg).unwrap();

        let key = 1u64.to_be_bytes();
        let v1 = 11u64.to_be_bytes();
        let v2 = 22u64.to_be_bytes();

        tree.put(&key, &v1).unwrap();
        let shard = &tree.durability.engine.shards()[0];
        shard.rotate_active_for_test(8).unwrap(); // v1 now durable in immutable file

        // Overwrite: v2 lands in the in-memory write buffer (no flush). The old
        // v1 entry's file gains dead bytes, so compaction selects it.
        tree.put(&key, &v2).unwrap();

        tree.compact().unwrap();

        // Simulate a crash before the buffer is flushed: only the on-disk .data
        // files survive. Some durable version of the key MUST remain.
        let shard_dir = dir.path().join("shard_000");
        let durable = scan_disk_for_key(&shard_dir, &key);
        assert!(
            durable.is_some(),
            "compaction dropped the only durable copy of a live key (overwrite was unflushed)"
        );
    }

    #[test]
    fn open_resolves_hints_false_for_const() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = crate::Config::test_no_hints();
        let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), cfg).unwrap();
        assert_eq!(tree.config().hints, Some(false));
    }

    #[test]
    fn const_tree_atomic_fires_hooks_in_order() {
        let dir = tempfile::tempdir().unwrap();
        let tree = open_const_hooked(dir.path(), RecHook::default());
        let k = 7u64.to_be_bytes();
        tree.atomic(&k, |s| {
            s.put(&k, &[1, 1, 1, 1])?; // new key: old=None
            s.put(&k, &[2, 2, 2, 2])?; // update: old=[1;4]
            s.delete(&k)?; // delete: old=[2;4], new=None
            Ok(())
        })
        .expect("atomic");

        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 3);
        let seq = crate::sync::lock(&tree.hook.seq).clone();
        assert_eq!(seq[0], (7, None, Some(vec![1, 1, 1, 1])));
        assert_eq!(seq[1], (7, Some(vec![1, 1, 1, 1]), Some(vec![2, 2, 2, 2])));
        assert_eq!(seq[2], (7, Some(vec![2, 2, 2, 2]), None));
    }

    fn peer_key_on_same_shard(key: &[u8; 8], shard_for: impl Fn(&[u8; 8]) -> usize) -> [u8; 8] {
        let shard = shard_for(key);
        (0u64..10_000)
            .map(|i| i.to_be_bytes())
            .find(|k| k != key && shard_for(k) == shard)
            .expect("second key on same shard")
    }

    fn fixed_sync_fail_cfg() -> FixedConfig {
        FixedConfig {
            shard_count: 1,
            grow_step: 64,
            sync_batch_size: 1,
            ..FixedConfig::test()
        }
    }

    /// F2: `sync()`-Err after an applied mutation must not swallow `on_write` —
    /// the mutation is already visible to readers, so the hook observes the
    /// in-memory commit.
    #[test]
    fn hook_fires_even_when_sync_fails() {
        use crate::durability::Fixed;
        use crate::fixed::shard::{SyncSite, fail_next_sync};

        let dir = tempdir().unwrap();
        let tree = ConstTree::<[u8; 8], 8, RecHook, Fixed>::open_with_hook(
            dir.path(),
            fixed_sync_fail_cfg(),
            RecHook::default(),
        )
        .unwrap();

        let failure = fail_next_sync(SyncSite::Write);
        let res = tree.put(&1u64.to_be_bytes(), &2u64.to_be_bytes());
        drop(failure);
        assert!(
            res.is_err(),
            "injected sync failure must propagate to caller"
        );
        assert_eq!(
            tree.hook.writes.load(AtomicOrdering::Relaxed),
            1,
            "on_write must fire despite sync error"
        );
        assert_eq!(
            tree.get(&1u64.to_be_bytes()),
            Some(2u64.to_be_bytes()),
            "mutation is visible — that's why the hook must have observed it"
        );
    }

    /// F2: `sync()`-Err inside `atomic()` must not drop replay of the event buffer.
    #[test]
    fn atomic_replay_fires_even_when_sync_fails() {
        use crate::durability::Fixed;
        use crate::fixed::shard::{SyncSite, fail_next_sync};

        let dir = tempdir().unwrap();
        let tree = ConstTree::<[u8; 8], 8, RecHook, Fixed>::open_with_hook(
            dir.path(),
            fixed_sync_fail_cfg(),
            RecHook::default(),
        )
        .unwrap();

        let key1 = 1u64.to_be_bytes();
        let peer_key = peer_key_on_same_shard(&key1, |k| tree.shard_for(k));

        let failure = fail_next_sync(SyncSite::Write);
        let res = tree.atomic(&key1, |shard| {
            shard.put(&key1, &2u64.to_be_bytes())?;
            shard.put(&peer_key, &3u64.to_be_bytes())?;
            Ok(())
        });
        drop(failure);
        assert!(res.is_err(), "injected sync failure must propagate");
        assert_eq!(
            tree.hook.writes.load(AtomicOrdering::Relaxed),
            2,
            "both buffered events must replay despite sync error"
        );
    }

    #[test]
    fn const_tree_atomic_fires_for_applied_on_err() {
        let dir = tempfile::tempdir().unwrap();
        let tree = open_const_hooked(dir.path(), RecHook::default());
        let k = 1u64.to_be_bytes();
        let r: DbResult<()> = tree.atomic(&k, |s| {
            s.put(&k, &[9, 9, 9, 9])?;
            Err(DbError::KeyNotFound) // app-level error after a successful mutation
        });
        assert!(r.is_err());
        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 1); // fired for the applied put
    }

    #[test]
    fn const_tree_atomic_nohook_applies_mutations() {
        // NoHook path: atomic still works, data persists, no hook machinery runs.
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        let tree = ConstTree::<[u8; 8], 4>::open(dir.path(), cfg).unwrap();
        let k = 3u64.to_be_bytes();
        tree.atomic(&k, |s| {
            s.put(&k, &[5, 5, 5, 5])?;
            Ok(())
        })
        .unwrap();
        assert_eq!(tree.get(&k), Some([5, 5, 5, 5]));
    }

    #[test]
    fn const_tree_atomic_hook_reentrancy_no_deadlock() {
        // A hook that reads the same tree from on_write must not deadlock
        // (replay happens after the shard lock is released).
        use std::sync::{Arc, OnceLock, Weak};
        #[derive(Default)]
        struct ReentrantHook {
            tree: OnceLock<Weak<ConstTree<[u8; 8], 4, ReentrantHook>>>,
            seen: AtomicUsize,
        }
        impl WriteHook<[u8; 8]> for ReentrantHook {
            fn on_write(&self, key: &[u8; 8], _old: Option<&[u8]>, _new: Option<&[u8]>) {
                if let Some(t) = self.tree.get().and_then(Weak::upgrade) {
                    let _ = t.get(key); // re-enters shard lock path
                    self.seen.fetch_add(1, AtomicOrdering::Relaxed);
                }
            }
        }
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        let tree =
            Arc::new(ConstTree::open_hooked(dir.path(), cfg, ReentrantHook::default()).unwrap());
        tree.hook.tree.set(Arc::downgrade(&tree)).ok();
        let k = 2u64.to_be_bytes();
        tree.atomic(&k, |s| {
            s.put(&k, &[1, 2, 3, 4])?;
            Ok(())
        })
        .unwrap();
        assert_eq!(tree.hook.seen.load(AtomicOrdering::Relaxed), 1);
    }

    #[test]
    fn const_tree_atomic_fires_hooks_fixedstore_sync_seam() {
        // A write threshold can make `should_sync()` true, so
        // `atomic()` runs `sync()?` BEFORE replaying hooks. Force the seam with
        // `sync_batch_size = 1` (every write trips it) and assert hooks still
        // fire, in application order, after the sync.
        let dir = tempdir().unwrap();
        let fixed_cfg = FixedConfig {
            shard_count: 1,
            grow_step: 64,
            sync_batch_size: 1,
            ..FixedConfig::test()
        };
        let tree = FixedTree::<[u8; 8], 4, RecHook>::open_with_hook(
            dir.path(),
            fixed_cfg,
            RecHook::default(),
        )
        .expect("open fixed hooked");

        let k = 7u64.to_be_bytes();
        tree.atomic(&k, |s| {
            s.put(&k, &[1, 1, 1, 1])?;
            s.put(&k, &[2, 2, 2, 2])?;
            s.delete(&k)?;
            Ok(())
        })
        .expect("atomic");

        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 3);
        let seq = crate::sync::lock(&tree.hook.seq).clone();
        assert_eq!(seq[0], (7, None, Some(vec![1, 1, 1, 1])));
        assert_eq!(seq[1], (7, Some(vec![1, 1, 1, 1]), Some(vec![2, 2, 2, 2])));
        assert_eq!(seq[2], (7, Some(vec![2, 2, 2, 2]), None));
    }

    fn cfg() -> FixedConfig {
        FixedConfig {
            shard_count: 2,
            grow_step: 64,
            ..FixedConfig::test()
        }
    }

    #[test]
    fn compare_delete_fixed_store_match_mismatch_absent() {
        let dir = tempdir().unwrap();
        let tree = FixedTree::<[u8; 8], 8>::open(dir.path(), cfg()).unwrap();

        let key = 1u64.to_be_bytes();
        let val = 42u64.to_be_bytes();
        let other = 99u64.to_be_bytes();
        tree.put(&key, &val).unwrap();

        // mismatch: value preserved
        assert!(matches!(
            tree.compare_delete(&key, &other),
            Err(DbError::CasMismatch)
        ));
        assert_eq!(tree.get(&key), Some(val));

        // match: deleted
        assert!(tree.compare_delete(&key, &val).is_ok());
        assert_eq!(tree.get(&key), None);

        // absent: KeyNotFound
        assert!(matches!(
            tree.compare_delete(&key, &val),
            Err(DbError::KeyNotFound)
        ));
    }

    #[test]
    fn compare_delete_bitcask_match_mismatch_absent() {
        let dir = tempdir().unwrap();
        let tree = crate::ConstTree::<[u8; 8], 8>::open(dir.path(), Config::test()).unwrap();

        let key = 7u64.to_be_bytes();
        let val = 7u64.to_be_bytes();
        let other = 8u64.to_be_bytes();
        tree.put(&key, &val).unwrap();

        assert!(matches!(
            tree.compare_delete(&key, &other),
            Err(DbError::CasMismatch)
        ));
        assert_eq!(tree.get(&key), Some(val));
        assert!(tree.compare_delete(&key, &val).is_ok());
        assert_eq!(tree.get(&key), None);
        assert!(matches!(
            tree.compare_delete(&key, &val),
            Err(DbError::KeyNotFound)
        ));
    }

    #[test]
    fn fixed_tree_default_desc() {
        let dir = tempdir().unwrap();
        let tree = FixedTree::<[u8; 8], 8>::open(dir.path(), FixedConfig::test()).unwrap();
        for i in 1u64..=3 {
            tree.put(&i.to_be_bytes(), &i.to_be_bytes()).unwrap();
        }
        let keys: Vec<u64> = tree.iter().map(|(k, _)| u64::from_be_bytes(k)).collect();
        assert_eq!(keys, [3, 2, 1], "default is DESC (newest first)");
        assert_eq!(tree.first().map(|(k, _)| u64::from_be_bytes(k)), Some(3));
        assert_eq!(tree.last().map(|(k, _)| u64::from_be_bytes(k)), Some(1));
    }

    #[test]
    fn fixed_tree_ascending_when_reversed_false() {
        let dir = tempdir().unwrap();
        let cfg = FixedConfig {
            reversed: false,
            ..FixedConfig::test()
        };
        let tree = FixedTree::<[u8; 8], 8>::open(dir.path(), cfg).unwrap();
        for i in 1u64..=3 {
            tree.put(&i.to_be_bytes(), &i.to_be_bytes()).unwrap();
        }
        let keys: Vec<u64> = tree.iter().map(|(k, _)| u64::from_be_bytes(k)).collect();
        assert_eq!(keys, [1, 2, 3], "reversed=false yields ASC");
        assert_eq!(tree.first().map(|(k, _)| u64::from_be_bytes(k)), Some(1));
        assert_eq!(tree.last().map(|(k, _)| u64::from_be_bytes(k)), Some(3));
    }

    #[test]
    fn const_tree_shard_update_and_fetch_update() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        let tree = ConstTree::<[u8; 8], 4>::open(dir.path(), cfg).unwrap();
        let key = 1u64.to_be_bytes();
        tree.put(&key, &[10u8; 4]).unwrap();

        let (upd, fetched, missing) = tree
            .atomic(&key, |shard| {
                // update returns the NEW value
                let upd = shard.update(&key, |old| {
                    let mut v = *old;
                    v[0] = v[0].wrapping_add(1);
                    v
                })?;
                // fetch_update returns the OLD value
                let fetched = shard.fetch_update(&key, |old| {
                    let mut v = *old;
                    v[0] = v[0].wrapping_add(1);
                    v
                })?;
                // missing key -> None, closure not called
                let missing = shard.update(&2u64.to_be_bytes(), |_| [0u8; 4])?;
                Ok((upd, fetched, missing))
            })
            .expect("atomic");

        assert_eq!(upd, Some([11, 10, 10, 10])); // new value
        assert_eq!(fetched, Some([11, 10, 10, 10])); // old value (before the +1)
        assert_eq!(missing, None);
        assert_eq!(tree.get(&key), Some([12, 10, 10, 10])); // both updates landed
    }

    #[test]
    fn const_tree_atomic_update_fires_hook_with_old_new() {
        let dir = tempdir().unwrap();
        let tree = open_const_hooked(dir.path(), RecHook::default());
        let key = 1u64.to_be_bytes();
        tree.put(&key, &[5u8; 4]).unwrap();
        let writes_before = tree.hook.writes.load(AtomicOrdering::Relaxed);
        tree.atomic(&key, |shard| {
            shard.update(&key, |_| [7u8; 4])?;
            Ok(())
        })
        .expect("atomic");
        // hook fired exactly once for the update, AFTER the lock was released
        assert_eq!(
            tree.hook.writes.load(AtomicOrdering::Relaxed),
            writes_before + 1
        );
        assert_eq!(tree.get(&key), Some([7u8; 4]));
    }

    #[test]
    fn fixed_tree_reversed_is_tunable_across_reopen() {
        // Spec acceptance criterion 4: `reversed` is in-memory only, so it can be
        // flipped between reopens with no on-disk migration. Open DESC, persist,
        // reopen ASC over the SAME data, and assert the order flipped while the
        // values stayed intact.
        let dir = tempdir().unwrap();
        {
            let tree = FixedTree::<[u8; 8], 8>::open(dir.path(), FixedConfig::test()).unwrap();
            for i in 1u64..=3 {
                tree.put(&i.to_be_bytes(), &i.to_be_bytes()).unwrap();
            }
            let keys: Vec<u64> = tree.iter().map(|(k, _)| u64::from_be_bytes(k)).collect();
            assert_eq!(keys, [3, 2, 1], "default DESC before reopen");
            tree.close().unwrap();
        }
        let cfg = FixedConfig {
            reversed: false,
            ..FixedConfig::test()
        };
        let tree = FixedTree::<[u8; 8], 8>::open(dir.path(), cfg).unwrap();
        let keys: Vec<u64> = tree.iter().map(|(k, _)| u64::from_be_bytes(k)).collect();
        assert_eq!(
            keys,
            [1, 2, 3],
            "ASC after reopen — direction is tunable, no migration"
        );
        // Data survived the reopen unchanged.
        assert_eq!(tree.get(&2u64.to_be_bytes()), Some(2u64.to_be_bytes()));
    }

    // ==========================================================================
    // Tests — get_many / update_many (Task 3)
    // ==========================================================================

    #[test]
    fn get_many_inline_aligned_dups_and_missing() {
        use crate::{Applied, Upsert};
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 4;
        let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), cfg).unwrap();
        for i in 1u64..=5 {
            tree.put(&i.to_be_bytes(), &(i * 10).to_be_bytes()).unwrap();
        }
        // input order with a duplicate (key 2) and a missing key (99)
        let keys: Vec<[u8; 8]> = [3u64, 1, 99, 2, 2]
            .iter()
            .map(|k| k.to_be_bytes())
            .collect();
        let got = tree.get_many(&keys);
        assert_eq!(got.len(), keys.len());
        assert_eq!(got[0], Some(30u64.to_be_bytes()));
        assert_eq!(got[1], Some(10u64.to_be_bytes()));
        assert_eq!(got[2], None);
        assert_eq!(got[3], Some(20u64.to_be_bytes()));
        assert_eq!(got[4], Some(20u64.to_be_bytes()));
        let _ = (Applied::<[u8; 8]>::Kept, Upsert::<[u8; 8]>::Keep); // suppress unused import warning
    }

    #[test]
    fn update_many_set_keep_delete_and_fold() {
        use crate::{Applied, Upsert};
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 4;
        let tree = ConstTree::<[u8; 8], 8>::open(dir.path(), cfg).unwrap();
        tree.put(&1u64.to_be_bytes(), &100u64.to_be_bytes())
            .unwrap();
        tree.put(&2u64.to_be_bytes(), &200u64.to_be_bytes())
            .unwrap();

        // items: key1 += delta twice (fold), key2 delete, key3 upsert-init, key4 delete-miss
        let items: Vec<([u8; 8], u64)> = vec![
            (1u64.to_be_bytes(), 5),
            (1u64.to_be_bytes(), 7),
            (2u64.to_be_bytes(), 0),
            (3u64.to_be_bytes(), 42),
            (4u64.to_be_bytes(), 0),
        ];
        let out = tree
            .update_many(items, |k, cur, delta| {
                let k2 = 2u64.to_be_bytes();
                let k4 = 4u64.to_be_bytes();
                if *k == k2 || *k == k4 {
                    Upsert::Delete
                } else {
                    let base = cur.map(|b| u64::from_be_bytes(*b)).unwrap_or(0);
                    Upsert::Put((base + *delta).to_be_bytes())
                }
            })
            .unwrap();

        assert_eq!(out.len(), 5);
        // pos0: key1 100->105 (insert? no, overwrite) old=Some(100) new=105
        assert_eq!(
            out[0].1,
            Applied::Written {
                old: Some(100u64.to_be_bytes()),
                new: 105u64.to_be_bytes()
            }
        );
        // pos1: key1 105->112 fold
        assert_eq!(
            out[1].1,
            Applied::Written {
                old: Some(105u64.to_be_bytes()),
                new: 112u64.to_be_bytes()
            }
        );
        // pos2: key2 deleted
        assert_eq!(out[2].1, Applied::Deleted(200u64.to_be_bytes()));
        // pos3: key3 inserted old=None
        assert_eq!(
            out[3].1,
            Applied::Written {
                old: None,
                new: 42u64.to_be_bytes()
            }
        );
        // pos4: key4 delete-miss -> Kept, no value
        assert_eq!(out[4].1, Applied::Kept);

        assert_eq!(tree.get(&1u64.to_be_bytes()), Some(112u64.to_be_bytes()));
        assert_eq!(tree.get(&2u64.to_be_bytes()), None);
        assert_eq!(tree.get(&3u64.to_be_bytes()), Some(42u64.to_be_bytes()));
    }

    #[test]
    fn update_many_fires_hooks_after_unlock_in_order() {
        use crate::{Applied, Upsert};
        use std::sync::Mutex;
        type VecItem = (u64, Option<u64>, Option<u64>);
        struct RecHook2(Mutex<Vec<VecItem>>);
        impl crate::WriteHook<[u8; 8]> for RecHook2 {
            fn on_write(&self, key: &[u8; 8], old: Option<&[u8]>, new: Option<&[u8]>) {
                let k = u64::from_be_bytes(*key);
                let o = old.map(|b| u64::from_be_bytes(b.try_into().unwrap()));
                let n = new.map(|b| u64::from_be_bytes(b.try_into().unwrap()));
                self.0.lock().unwrap().push((k, o, n));
            }
        }
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1; // single shard => deterministic application order
        let tree = ConstTree::<[u8; 8], 8, RecHook2>::open_hooked(
            dir.path(),
            cfg,
            RecHook2(Mutex::new(Vec::new())),
        )
        .unwrap();
        let items: Vec<([u8; 8], u64)> = vec![
            (1u64.to_be_bytes(), 10),
            (2u64.to_be_bytes(), 20),
            (1u64.to_be_bytes(), 11),
        ];
        tree.update_many(items, |_k, _cur, v| Upsert::Put(v.to_be_bytes()))
            .unwrap();
        let rec = tree.hook_ref().0.lock().unwrap().clone();
        assert_eq!(
            rec,
            vec![
                (1, None, Some(10)),
                (2, None, Some(20)),
                (1, Some(10), Some(11))
            ]
        );
        let _ = Applied::<[u8; 8]>::Kept; // suppress unused import warning
    }

    #[test]
    fn update_many_partial_error_barriers_applied_prefix_after_unlock() {
        reset();
        let gate = hold_data_sync();
        let dir = tempdir().unwrap();
        let mut config = Config::test();
        config.shard_count = 1;
        config.sync_batch_size = 3;
        let tree =
            Arc::new(ConstTree::<[u8; 8], 8>::open(dir.path(), config).expect("open const tree"));
        FAIL_NEXT_DATA_SYNC.store(true, AtomicOrdering::SeqCst);
        let worker_tree = Arc::clone(&tree);
        let worker = std::thread::spawn(move || {
            let _failure = UpdateManyFailAfter::after(3);
            worker_tree.update_many(
                (0..4_u64)
                    .map(|value| (value.to_be_bytes(), value))
                    .collect(),
                |_key, _old, value| crate::Upsert::Put(value.to_be_bytes()),
            )
        });

        gate.wait_until_blocked();
        drop(tree.durability.lock_shard(0));
        gate.release();

        let result = worker.join().expect("update_many worker panicked");
        assert!(
            matches!(
                result,
                Err(DbError::Internal("test: update_many injected failure"))
            ),
            "the later mutation error must remain the operation result"
        );
        assert_eq!(
            data_attempts(),
            1,
            "the applied prefix must attempt one barrier"
        );
        assert_eq!(
            data_successes(),
            0,
            "the injected barrier failure must not replace the mutation error"
        );
    }

    #[test]
    fn update_many_fixed_backend_persists() {
        use crate::{Applied, Upsert};
        let dir = tempdir().unwrap();
        let cfg = FixedConfig::test();
        let tree =
            ConstTree::<[u8; 8], 8, crate::NoHook, crate::durability::Fixed>::open(dir.path(), cfg)
                .unwrap();
        let items: Vec<([u8; 8], u64)> = vec![(1u64.to_be_bytes(), 111), (2u64.to_be_bytes(), 222)];
        tree.update_many(items, |_k, _cur, v| Upsert::Put(v.to_be_bytes()))
            .unwrap();
        assert_eq!(tree.get(&1u64.to_be_bytes()), Some(111u64.to_be_bytes()));
        assert_eq!(tree.get(&2u64.to_be_bytes()), Some(222u64.to_be_bytes()));
        let _ = Applied::<[u8; 8]>::Kept; // suppress unused import warning
    }

    /// Regression (WP-C, fixed-store F1/F2): a crash with `enable_fsync=false`
    /// can leave the same key durably present in two OCCUPIED, CRC-valid slots.
    /// Recovery must keep exactly one node (no node leak on the `Exists` branch),
    /// free the losing slot in `versions`+`bitmap`, leave `bitmap.occupied() ==
    /// len()`, and pick the same deterministic winner as ConstMap
    /// (last-scanned-wins).
    #[test]
    fn fixed_tree_recovery_dedups_duplicate_key() {
        use crate::durability::{Durability, Fixed};

        let key = [7u8; 8];
        let v1 = [0xAAu8; 8];
        let v2 = [0xBBu8; 8];
        let cfg = FixedConfig {
            shard_count: 1,
            grow_step: 64,
            ..FixedConfig::test()
        };

        // Forge helper: same key in two distinct slots, no clean shutdown (drop)
        // → dirty full-scan recovery on reopen.
        let forge = |dir: &std::path::Path| {
            let fixed = Fixed::open(dir, cfg.clone(), 8, 8).unwrap();
            let mut inner = fixed.lock_shard(0);
            let a = inner.alloc_slot().unwrap();
            inner.write_slot(a, &key, &v1).unwrap();
            let b = inner.alloc_slot().unwrap();
            inner.write_slot(b, &key, &v2).unwrap();
            assert_ne!(a, b, "two distinct slots hold the same key");
            inner.sync(crate::fixed::shard::SyncSite::Write).unwrap();
        };

        let dir = tempdir().unwrap();
        forge(dir.path());

        let tree = FixedTree::<[u8; 8], 8>::open(dir.path(), cfg.clone()).unwrap();
        assert_eq!(tree.len(), 1, "exactly one node after dedup (no node leak)");
        assert_eq!(
            tree.get(&key),
            Some(v2),
            "last-scanned slot (higher id) wins"
        );

        {
            let inner = tree.durability.lock_shard(0);
            assert_eq!(
                inner.bitmap.occupied() as usize,
                tree.len(),
                "bitmap.occupied() == len() (no leaked slot)"
            );
            assert_eq!(inner.bitmap.occupied(), 1);
            assert_eq!(
                crate::fixed::slot::status_of(inner.versions[0]),
                crate::fixed::slot::STATUS_FREE,
                "losing slot 0 freed in versions"
            );
            assert!(!inner.bitmap.is_set(0), "losing slot 0 cleared in bitmap");
            assert_eq!(
                crate::fixed::slot::status_of(inner.versions[1]),
                crate::fixed::slot::STATUS_OCCUPIED,
                "winning slot 1 kept OCCUPIED"
            );
        }

        // ConstMap must resolve the identical on-disk state to the SAME value —
        // the recovery winner policy is unified across both collections.
        let dir_map = tempdir().unwrap();
        forge(dir_map.path());
        let map = crate::FixedMap::<[u8; 8], 8>::open(dir_map.path(), cfg).unwrap();
        assert_eq!(map.len(), 1);
        assert_eq!(
            map.get(&key),
            tree.get(&key),
            "ConstMap and ConstTree recover the same value"
        );
    }

    #[test]
    fn range_bounds_dir_asc_matches_desc_rev() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        let tree = ConstTree::<[u8; 8], 4>::open(dir.path(), cfg).unwrap();
        for i in [1u8, 3, 5] {
            let mut k = [0u8; 8];
            k[7] = i;
            tree.put(&k, &[i; 4]).unwrap();
        }

        let asc: Vec<_> = tree
            .range_bounds_dir(Bound::Unbounded, Bound::Unbounded, Dir::Asc)
            .map(|(k, _)| k[7])
            .collect();
        let desc_rev: Vec<_> = tree
            .range_bounds_dir(Bound::Unbounded, Bound::Unbounded, Dir::Desc)
            .map(|(k, _)| k[7])
            .collect::<Vec<_>>()
            .into_iter()
            .rev()
            .collect();
        assert_eq!(asc, vec![1, 3, 5]);
        assert_eq!(asc, desc_rev);

        let storage: Vec<_> = tree
            .range_bounds(Bound::Unbounded, Bound::Unbounded)
            .map(|(k, _)| k[7])
            .collect();
        let with_storage: Vec<_> = tree
            .range_bounds_dir(
                Bound::Unbounded,
                Bound::Unbounded,
                dir_from(tree.reversed()),
            )
            .map(|(k, _)| k[7])
            .collect();
        assert_eq!(storage, with_storage);
    }

    type CompactionTestTree = ConstTree<[u8; 8], 8>;

    fn compaction_hint_config() -> Config {
        let mut config = Config::test();
        config.shard_count = 1;
        config.hints = Some(true);
        config.compaction_threshold = 0.0;
        config.compaction_max_files_per_pass = 0;
        config
    }

    fn tombstone_gc_config() -> Config {
        let mut config = compaction_hint_config();
        config.compaction_max_files_per_pass = 0;
        config
    }

    fn tombstone_gc_budget_one_config() -> Config {
        let mut config = tombstone_gc_config();
        config.compaction_threshold = 0.20;
        config.compaction_max_files_per_pass = 1;
        config
    }

    fn data_file_exists(root: &std::path::Path, file_id: u32) -> bool {
        root.join("shard_000")
            .join(format!("{file_id:06}.data"))
            .exists()
    }

    fn compact_to_stable(tree: &CompactionTestTree, passes: usize) {
        for _ in 0..passes {
            tree.compact().unwrap();
        }
    }

    fn count_physical_tombstones(root: &std::path::Path, key: [u8; 8]) -> usize {
        use crate::entry::EntryHeader;
        use zerocopy::FromBytes;

        let shard_dir = root.join("shard_000");
        let mut ids: Vec<u32> = std::fs::read_dir(&shard_dir)
            .unwrap()
            .filter_map(|entry| {
                entry
                    .ok()?
                    .file_name()
                    .to_string_lossy()
                    .strip_suffix(".data")?
                    .parse()
                    .ok()
            })
            .collect();
        ids.sort_unstable();
        let mut tombstones = 0;
        let header_len = std::mem::size_of::<EntryHeader>();
        for id in ids {
            let bytes = std::fs::read(shard_dir.join(format!("{id:06}.data"))).unwrap();
            let mut offset = 0;
            while offset + header_len <= bytes.len() {
                let Ok(header) = EntryHeader::read_from_bytes(&bytes[offset..offset + header_len])
                else {
                    break;
                };
                if header.gsn == 0 && header.crc32 == 0 && header.value_len == 0 {
                    break;
                }
                let total = header_len + 8 + header.value_len as usize;
                if offset + total > bytes.len() {
                    break;
                }
                if header.is_tombstone()
                    && bytes[offset + header_len..offset + header_len + 8] == key
                {
                    tombstones += 1;
                }
                offset += total;
            }
        }
        tombstones
    }

    fn tombstone_gc_tree_with_coverable_files() -> (tempfile::TempDir, CompactionTestTree, [u8; 8])
    {
        let dir = tempfile::tempdir().unwrap();
        let tree = CompactionTestTree::open(dir.path(), tombstone_gc_config()).unwrap();
        let deleted_key = 1u64.to_be_bytes();
        let peer = 2u64.to_be_bytes();
        tree.put(&deleted_key, &10u64.to_be_bytes()).unwrap();
        tree.durability.engine.shards()[0]
            .rotate_active_for_test(8)
            .unwrap();
        tree.delete(&deleted_key).unwrap();
        tree.put(&peer, &1u64.to_be_bytes()).unwrap();
        tree.put(&peer, &2u64.to_be_bytes()).unwrap();
        tree.durability.engine.shards()[0]
            .rotate_active_for_test(8)
            .unwrap();
        (dir, tree, deleted_key)
    }

    fn tombstone_gc_tree_with_budget_one() -> (tempfile::TempDir, CompactionTestTree, [u8; 8], u32)
    {
        let dir = tempfile::tempdir().unwrap();
        let tree = CompactionTestTree::open(dir.path(), tombstone_gc_budget_one_config()).unwrap();
        let deleted_key = 1u64.to_be_bytes();
        tree.put(&deleted_key, &10u64.to_be_bytes()).unwrap();
        for filler in 10u64..74 {
            tree.put(&filler.to_be_bytes(), &filler.to_be_bytes())
                .unwrap();
        }
        tree.durability.engine.shards()[0]
            .rotate_active_for_test(8)
            .unwrap();
        let put_file_id = 1;
        let peer = 2u64.to_be_bytes();
        tree.delete(&deleted_key).unwrap();
        tree.put(&peer, &1u64.to_be_bytes()).unwrap();
        tree.put(&peer, &2u64.to_be_bytes()).unwrap();
        tree.durability.engine.shards()[0]
            .rotate_active_for_test(8)
            .unwrap();
        (dir, tree, deleted_key, put_file_id)
    }

    fn tombstone_sequence_in_file(root: &std::path::Path, file_id: u32, key: [u8; 8]) -> u64 {
        use zerocopy::FromBytes;

        let bytes =
            std::fs::read(root.join("shard_000").join(format!("{file_id:06}.data"))).unwrap();
        let header_len = std::mem::size_of::<crate::entry::EntryHeader>();
        let mut offset = 0;
        while offset + header_len <= bytes.len() {
            let header =
                crate::entry::EntryHeader::read_from_bytes(&bytes[offset..offset + header_len])
                    .unwrap();
            if header.gsn == 0 && header.crc32 == 0 && header.value_len == 0 {
                break;
            }
            let total = header_len + 8 + header.value_len as usize;
            if header.is_tombstone() && bytes[offset + header_len..offset + header_len + 8] == key {
                return header.sequence();
            }
            offset += total;
        }
        panic!("expected tombstone for key in file {file_id}");
    }

    fn sequence_for_key_in_file(root: &std::path::Path, file_id: u32, key: [u8; 8]) -> u64 {
        use zerocopy::FromBytes;

        let bytes =
            std::fs::read(root.join("shard_000").join(format!("{file_id:06}.data"))).unwrap();
        let header_len = std::mem::size_of::<crate::entry::EntryHeader>();
        let mut offset = 0;
        while offset + header_len <= bytes.len() {
            let header =
                crate::entry::EntryHeader::read_from_bytes(&bytes[offset..offset + header_len])
                    .unwrap();
            if header.gsn == 0 && header.crc32 == 0 && header.value_len == 0 {
                break;
            }
            let total = header_len + 8 + header.value_len as usize;
            if bytes[offset + header_len..offset + header_len + 8] == key {
                return header.sequence();
            }
            offset += total;
        }
        panic!("expected key in file {file_id}");
    }

    #[test]
    fn compaction_drops_tombstone_after_all_older_files_are_covered() {
        let (dir, tree, deleted_key) = tombstone_gc_tree_with_coverable_files();
        compact_to_stable(&tree, 16);
        assert_eq!(tree.get(&deleted_key), None);
        assert_eq!(count_physical_tombstones(dir.path(), deleted_key), 0);
        drop(tree);
        let reopened = CompactionTestTree::open(dir.path(), tombstone_gc_config()).unwrap();
        assert_eq!(reopened.get(&deleted_key), None);
    }

    #[test]
    fn compaction_keeps_tombstone_when_budget_leaves_older_put_uncovered() {
        let (dir, tree, deleted_key, put_file_id) = tombstone_gc_tree_with_budget_one();
        tree.compact().unwrap();
        assert!(data_file_exists(dir.path(), put_file_id));
        assert_eq!(count_physical_tombstones(dir.path(), deleted_key), 1);
        drop(tree);
        let reopened =
            CompactionTestTree::open(dir.path(), tombstone_gc_budget_one_config()).unwrap();
        assert_eq!(reopened.get(&deleted_key), None);
    }

    #[test]
    fn unknown_active_file_is_a_widening_hard_stop() {
        let (dir, tree, deleted_key, put_file_id) = tombstone_gc_tree_with_budget_one();
        let shard = &tree.durability.engine.shards()[0];
        // Model a legacy partial active hint recovered by Task 2: the active
        // physical floor is unknown, so it must prevent widening even though it
        // itself is never selected.
        shard.lock().active.seqs = crate::shard::FileSeqs::UNKNOWN;
        tree.compact().unwrap();
        assert!(data_file_exists(dir.path(), put_file_id));
        assert_eq!(count_physical_tombstones(dir.path(), deleted_key), 1);
    }

    #[test]
    fn compaction_widens_into_below_threshold_tombstone_file() {
        let (dir, tree, deleted_key, put_file_id) = tombstone_gc_tree_with_budget_one();
        let initial_candidate_id = {
            let inner = tree.durability.engine.shards()[0].lock();
            let source = inner
                .immutable
                .iter()
                .find(|file| file.file_id == put_file_id)
                .unwrap();
            let source_dead = inner.dead_bytes.get(&put_file_id).copied().unwrap_or(0);
            assert!(
                (source_dead as f64 / source.total_bytes as f64)
                    <= tree.durability.compaction_threshold,
                "the tombstone source must be below threshold before widening"
            );
            let candidate = inner
                .immutable
                .iter()
                .find(|file| file.file_id != put_file_id)
                .unwrap();
            let candidate_dead = inner
                .dead_bytes
                .get(&candidate.file_id)
                .copied()
                .unwrap_or(0);
            assert!(
                (candidate_dead as f64 / candidate.total_bytes as f64)
                    > tree.durability.compaction_threshold,
                "a distinct initial candidate must trigger widening"
            );
            candidate.file_id
        };
        tree.durability.engine.shards()[0]
            .lock()
            .compaction_max_files_per_pass = 0;
        tree.compact().unwrap();
        assert_ne!(initial_candidate_id, put_file_id);
        assert!(!data_file_exists(dir.path(), put_file_id));
        assert_eq!(count_physical_tombstones(dir.path(), deleted_key), 0);
    }

    #[test]
    fn cooldown_file_is_a_widening_hard_stop() {
        let (dir, tree, deleted_key, put_file_id) = tombstone_gc_tree_with_budget_one();
        tree.durability.engine.shards()[0]
            .lock()
            .last_compaction_output_ids = vec![put_file_id];
        tree.compact().unwrap();
        assert!(data_file_exists(dir.path(), put_file_id));
        assert_eq!(count_physical_tombstones(dir.path(), deleted_key), 1);
    }

    #[test]
    fn unknown_file_is_a_widening_hard_stop() {
        let (dir, tree, deleted_key, put_file_id) = tombstone_gc_tree_with_budget_one();
        // A queued hint job holds a second `Arc` to the file it retired (F8),
        // so drain before reaching for exclusive access to one.
        tree.durability.engine.drain_hints_for_test();
        let mut inner = tree.durability.engine.shards()[0].lock();
        std::sync::Arc::get_mut(&mut inner.immutable[0])
            .unwrap()
            .seqs = crate::shard::FileSeqs::UNKNOWN;
        drop(inner);
        tree.compact().unwrap();
        assert!(data_file_exists(dir.path(), put_file_id));
        assert_eq!(count_physical_tombstones(dir.path(), deleted_key), 1);
    }

    #[test]
    fn tombstone_first_file_minimum_uses_sequence_not_flagged_gsn() {
        use zerocopy::FromBytes;
        let (dir, tree, deleted_key) = tombstone_gc_tree_with_coverable_files();
        let path = dir.path().join("shard_000/000002.data");
        let bytes = std::fs::read(path).unwrap();
        let header = crate::entry::EntryHeader::read_from_bytes(
            &bytes[..std::mem::size_of::<crate::entry::EntryHeader>()],
        )
        .unwrap();
        let tombstone_sequence = header.sequence();
        let state = tree.durability.engine.shards()[0].lock().immutable[1]
            .seqs
            .min;
        assert_eq!(state, crate::shard::MinSeq::Known(tombstone_sequence));
        assert_ne!(
            state,
            crate::shard::MinSeq::Known(tombstone_sequence | crate::entry::TOMBSTONE_BIT)
        );
        assert_eq!(tree.get(&deleted_key), None);
    }

    #[cfg(test)]
    #[cfg(feature = "replication")]
    fn tombstone_gc_tree_with_guard_hole() -> (
        tempfile::TempDir,
        CompactionTestTree,
        [u8; 8],
        u64,
        u32,
        u32,
    ) {
        let dir = tempfile::tempdir().unwrap();
        let tree = CompactionTestTree::open(dir.path(), tombstone_gc_config()).unwrap();
        let target = 1u64.to_be_bytes();
        let filler = 2u64.to_be_bytes();
        let peer = 3u64.to_be_bytes();
        let high = 4u64.to_be_bytes();

        // File 1 supplies a live sequence below the target tombstone.  File 2
        // is the source which the guarded pass must retire.
        tree.put(&filler, &10u64.to_be_bytes()).unwrap();
        tree.put(&target, &11u64.to_be_bytes()).unwrap();
        tree.durability.engine.shards()[0]
            .rotate_active_for_test(8)
            .unwrap();
        tree.delete(&target).unwrap();
        tree.put(&peer, &1u64.to_be_bytes()).unwrap();
        tree.put(&peer, &2u64.to_be_bytes()).unwrap();
        tree.durability.engine.shards()[0]
            .rotate_active_for_test(8)
            .unwrap();
        let tombstone_source_id = 2;
        let tombstone_seq = tombstone_sequence_in_file(dir.path(), tombstone_source_id, target);

        // File 3 has a live sequence above the tombstone and one dead entry,
        // letting the first pass make a high-id output whose range straddles it.
        tree.put(&high, &1u64.to_be_bytes()).unwrap();
        tree.put(&high, &2u64.to_be_bytes()).unwrap();
        tree.durability.engine.shards()[0]
            .rotate_active_for_test(8)
            .unwrap();
        {
            let mut inner = tree.durability.engine.shards()[0].lock();
            inner.last_compaction_output_ids = vec![tombstone_source_id];
        }
        tree.compact().unwrap();
        let mixed_output_id = newest_immutable_file_id(&tree);
        assert!(mixed_output_id > tombstone_source_id);
        let mixed = tree.durability.engine.shards()[0]
            .lock()
            .immutable
            .iter()
            .find(|file| file.file_id == mixed_output_id)
            .unwrap()
            .seqs
            .min;
        assert!(matches!(mixed, crate::shard::MinSeq::Known(min) if min < tombstone_seq));
        // Rotation hands hint generation to a background worker (F8), so wait
        // for the queue before reading a retired file's hint — and before the
        // guarded pass below, whose watermark filter reads one too.
        tree.durability.engine.drain_hints_for_test();
        // The live high entry is last in the output, so obtain its maximum from
        // the physical hint rather than relying on its first record.
        let hint = std::fs::read(
            dir.path()
                .join("shard_000")
                .join(format!("{mixed_output_id:06}.hint")),
        )
        .unwrap();
        let max_seq = crate::hint::parse_hint_entries::<[u8; 8]>(&hint)
            .map(|entry| entry.gsn & crate::entry::SEQUENCE_MASK)
            .max()
            .unwrap();
        assert!(max_seq > tombstone_seq);
        let source_hint = std::fs::read(
            dir.path()
                .join("shard_000")
                .join(format!("{tombstone_source_id:06}.hint")),
        )
        .unwrap();
        let source_max = crate::hint::parse_hint_entries::<[u8; 8]>(&source_hint)
            .map(|entry| entry.gsn & crate::entry::SEQUENCE_MASK)
            .max()
            .unwrap();
        assert!(tombstone_seq <= source_max && source_max < max_seq);
        let watermark = source_max + (max_seq - source_max) / 2;
        assert!(tombstone_seq < watermark && watermark < max_seq);
        (
            dir,
            tree,
            target,
            watermark,
            tombstone_source_id,
            mixed_output_id,
        )
    }

    #[cfg(test)]
    #[cfg(feature = "replication")]
    #[test]
    fn guard_withheld_file_keeps_drop_floor_below_tombstone() {
        struct FixedWatermark(u64);
        impl crate::compaction::CompactionGuard for FixedWatermark {
            fn min_replicated_gsn(&self, _: u8) -> u64 {
                self.0
            }
        }
        let (dir, tree, deleted_key, watermark, tombstone_source_id, mixed_output_id) =
            tombstone_gc_tree_with_guard_hole();
        crate::compaction::compact_shard_guarded(
            &tree.durability.engine.shards()[0],
            &tree,
            tree.durability.compaction_threshold,
            &FixedWatermark(watermark),
        )
        .unwrap();
        assert!(
            !data_file_exists(dir.path(), tombstone_source_id),
            "guard-eligible tombstone source must be retired"
        );
        assert!(
            data_file_exists(dir.path(), mixed_output_id),
            "mixed-range output above the watermark must stay outside the pass"
        );
        assert_eq!(count_physical_tombstones(dir.path(), deleted_key), 1);
        drop(tree);
        assert_eq!(
            CompactionTestTree::open(dir.path(), tombstone_gc_config())
                .unwrap()
                .get(&deleted_key),
            None
        );
    }

    /// A shard where widening can never reach the tombstone: a guard-withheld
    /// file sits between the clean low-floor file and the tombstone sequence.
    ///
    /// Returns the tree, the deleted key, a replication watermark that blocks
    /// only the straddling output, the clean file's id and that output's id.
    #[cfg(test)]
    #[cfg(feature = "replication")]
    fn tombstone_gc_tree_with_unreachable_floor() -> (
        tempfile::TempDir,
        CompactionTestTree,
        [u8; 8],
        u64,
        u32,
        u32,
    ) {
        let dir = tempfile::tempdir().unwrap();
        let tree = CompactionTestTree::open(dir.path(), tombstone_gc_config()).unwrap();
        let clean_a = 1u64.to_be_bytes();
        let clean_b = 2u64.to_be_bytes();
        let target = 3u64.to_be_bytes();
        let straddle = 4u64.to_be_bytes();
        let after = 5u64.to_be_bytes();
        let later = 6u64.to_be_bytes();
        let rotate = || {
            tree.durability.engine.shards()[0]
                .rotate_active_for_test(8)
                .unwrap();
        };

        // File 1 — never overwritten, so it holds the shard's lowest sequences
        // and no dead bytes. This is the file a widening pass would rewrite.
        tree.put(&clean_a, &1u64.to_be_bytes()).unwrap();
        tree.put(&clean_b, &2u64.to_be_bytes()).unwrap();
        rotate();
        let clean_id = 1;

        // File 2 — the deleted key's put plus an overwrite, so it has dead
        // bytes and one survivor whose sequence is below the tombstone.
        tree.put(&target, &10u64.to_be_bytes()).unwrap();
        tree.put(&straddle, &1u64.to_be_bytes()).unwrap();
        tree.put(&straddle, &2u64.to_be_bytes()).unwrap();
        rotate();

        // File 3 — the tombstone. No dead bytes of its own, so it stays below
        // the threshold and only widening could ever select it.
        tree.delete(&target).unwrap();
        tree.put(&after, &1u64.to_be_bytes()).unwrap();
        rotate();
        let tombstone_source_id = 3;
        let tombstone_seq = tombstone_sequence_in_file(dir.path(), tombstone_source_id, target);

        // File 4 — entries above the tombstone, one of them dead.
        tree.put(&later, &1u64.to_be_bytes()).unwrap();
        tree.put(&later, &2u64.to_be_bytes()).unwrap();
        rotate();

        // Hold files 1 and 3 out of this pass (cooldown is a hard stop for
        // both selection and widening) so it merges 2 and 4 into one output
        // that straddles the tombstone sequence.
        {
            let mut inner = tree.durability.engine.shards()[0].lock();
            inner.last_compaction_output_ids = vec![clean_id, tombstone_source_id];
        }
        tree.compact().unwrap();
        let straddling_id = newest_immutable_file_id(&tree);
        assert!(straddling_id > tombstone_source_id);
        assert!(data_file_exists(dir.path(), clean_id));
        assert!(data_file_exists(dir.path(), tombstone_source_id));

        let straddling_min = tree.durability.engine.shards()[0]
            .lock()
            .immutable
            .iter()
            .find(|file| file.file_id == straddling_id)
            .unwrap()
            .seqs
            .min;
        assert!(
            matches!(straddling_min, crate::shard::MinSeq::Known(min) if min < tombstone_seq),
            "the straddling output must cap the floor below the tombstone"
        );

        // Rotation hint generation is asynchronous (F8): wait for it before
        // reading a retired file's hint, and before the guarded pass whose
        // watermark filter reads one.
        tree.durability.engine.drain_hints_for_test();

        // Watermark = the straddling output's maximum: it withholds that file
        // and nothing else, so the tombstone's own file stays selectable.
        let watermark = max_hint_sequence(dir.path(), straddling_id);
        assert!(tombstone_seq < watermark);
        assert!(max_hint_sequence(dir.path(), tombstone_source_id) < watermark);
        (dir, tree, target, watermark, clean_id, straddling_id)
    }

    fn max_hint_sequence(root: &std::path::Path, file_id: u32) -> u64 {
        let hint =
            std::fs::read(root.join("shard_000").join(format!("{file_id:06}.hint"))).unwrap();
        crate::hint::parse_hint_entries::<[u8; 8]>(&hint)
            .map(|entry| entry.gsn & crate::entry::SEQUENCE_MASK)
            .max()
            .unwrap()
    }

    #[cfg(test)]
    #[cfg(feature = "replication")]
    #[test]
    fn widening_rolls_back_when_no_tombstone_can_be_freed() {
        struct FixedWatermark(u64);
        impl crate::compaction::CompactionGuard for FixedWatermark {
            fn min_replicated_gsn(&self, _: u8) -> u64 {
                self.0
            }
        }
        let (dir, tree, deleted_key, watermark, clean_id, straddling_id) =
            tombstone_gc_tree_with_unreachable_floor();

        // Two passes: the first proves the clean file is not rewritten, the
        // second that the cooldown clearing in between does not restart the
        // cycle. Without the rollback each pass copies the clean file into a
        // fresh output whose low sequences make it a candidate again — a
        // self-sustaining rewrite that reclaims nothing.
        for pass in 1..=2 {
            let compacted = crate::compaction::compact_shard_guarded(
                &tree.durability.engine.shards()[0],
                &tree,
                tree.durability.compaction_threshold,
                &FixedWatermark(watermark),
            )
            .unwrap();
            assert_eq!(compacted, 0, "pass {pass} must not rewrite anything");
            assert!(
                data_file_exists(dir.path(), clean_id),
                "pass {pass} must leave the clean file in place"
            );
            assert_eq!(
                newest_immutable_file_id(&tree),
                straddling_id,
                "pass {pass} must not produce an output"
            );
        }
        assert_eq!(count_physical_tombstones(dir.path(), deleted_key), 1);
        assert_eq!(tree.get(&deleted_key), None);
    }

    fn selectable_compaction_tree(hints: bool) -> (tempfile::TempDir, CompactionTestTree) {
        let dir = tempfile::tempdir().unwrap();
        let mut config = compaction_hint_config();
        config.hints = Some(hints);
        let tree = CompactionTestTree::open(dir.path(), config).unwrap();
        let key = 1u64.to_be_bytes();
        tree.put(&key, &10u64.to_be_bytes()).unwrap();
        tree.put(&key, &11u64.to_be_bytes()).unwrap();
        tree.put(&2u64.to_be_bytes(), &20u64.to_be_bytes()).unwrap();
        tree.durability.engine.shards()[0]
            .rotate_active_for_test(8)
            .unwrap();
        (dir, tree)
    }

    #[test]
    fn compaction_second_pass_returns_zero_while_first_holds_lease() {
        let (_dir, tree) = selectable_compaction_tree(true);
        let tree = std::sync::Arc::new(tree);
        let (entered_tx, entered_rx) = std::sync::mpsc::sync_channel(0);
        let (resume_tx, resume_rx) = std::sync::mpsc::sync_channel(0);
        let first_tree = tree.clone();
        let first = std::thread::spawn(move || {
            crate::compaction::test_pause_after_pass_acquire(entered_tx, resume_rx);
            first_tree.compact()
        });
        entered_rx.recv().unwrap();
        assert_eq!(tree.compact().unwrap(), 0);
        resume_tx.send(()).unwrap();
        first.join().unwrap().unwrap();
    }

    #[cfg(test)]
    #[cfg(feature = "replication")]
    #[test]
    fn guarded_compaction_without_hints_returns_configuration_error() {
        let (_dir, tree) = selectable_compaction_tree(false);
        let shard = &tree.durability.engine.shards()[0];
        let error = crate::compaction::compact_shard_guarded(
            shard,
            &tree,
            tree.durability.compaction_threshold,
            &crate::compaction::NoReplicationGuard,
        )
        .unwrap_err();
        assert!(matches!(
            error,
            DbError::GuardedCompactionRequiresHints { shard: 0 }
        ));
    }

    #[cfg(test)]
    #[cfg(feature = "replication")]
    #[test]
    fn guarded_no_hints_error_precedes_busy_lease() {
        let (_dir, tree) = selectable_compaction_tree(false);
        let tree = std::sync::Arc::new(tree);
        let (entered_tx, entered_rx) = std::sync::mpsc::sync_channel(0);
        let (resume_tx, resume_rx) = std::sync::mpsc::sync_channel(0);
        let worker_tree = tree.clone();
        let worker = std::thread::spawn(move || {
            crate::compaction::test_pause_after_pass_acquire(entered_tx, resume_rx);
            worker_tree.compact()
        });
        entered_rx.recv().unwrap();
        let guarded = crate::compaction::compact_shard_guarded(
            &tree.durability.engine.shards()[0],
            tree.as_ref(),
            tree.durability.compaction_threshold,
            &crate::compaction::NoReplicationGuard,
        );
        assert!(matches!(
            guarded,
            Err(DbError::GuardedCompactionRequiresHints { shard: 0 })
        ));
        resume_tx.send(()).unwrap();
        worker.join().unwrap().unwrap();
    }

    #[cfg(test)]
    #[cfg(feature = "replication")]
    #[test]
    fn guarded_hints_with_unreplicated_data_still_returns_zero() {
        struct NothingReplicated;
        impl crate::compaction::CompactionGuard for NothingReplicated {
            fn min_replicated_gsn(&self, _shard_id: u8) -> u64 {
                0
            }
        }
        let (_dir, tree) = selectable_compaction_tree(true);
        let result = crate::compaction::compact_shard_guarded(
            &tree.durability.engine.shards()[0],
            &tree,
            tree.durability.compaction_threshold,
            &NothingReplicated,
        );
        assert_eq!(result.unwrap(), 0);
    }

    #[test]
    fn compaction_lease_resets_after_error() {
        let (_dir, tree) = selectable_compaction_tree(true);
        {
            // Scoped so the failpoint is disarmed before the retry below: the
            // second pass must succeed because the lease was released, not
            // because the injected fault happened to target only call one.
            let _fault = crate::compaction::test_fail_next_output_open();
            assert!(tree.compact().is_err());
        }
        assert!(
            !tree.durability.engine.shards()[0]
                .compacting
                .load(std::sync::atomic::Ordering::Acquire)
        );
        tree.compact().unwrap();
    }

    #[test]
    fn compaction_lease_resets_after_panic() {
        let (_dir, tree) = selectable_compaction_tree(true);
        crate::compaction::test_panic_after_pass_acquire();
        let panic = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| tree.compact()));
        assert!(panic.is_err());
        assert!(
            !tree.durability.engine.shards()[0]
                .compacting
                .load(std::sync::atomic::Ordering::Acquire)
        );
        tree.compact().unwrap();
    }

    fn compaction_tree_with_selectable_lost_cas_put()
    -> (tempfile::TempDir, CompactionTestTree, [u8; 8]) {
        let dir = tempfile::tempdir().unwrap();
        let tree = CompactionTestTree::open(dir.path(), compaction_hint_config()).unwrap();
        let dead_key = 1u64.to_be_bytes();
        let lost_key = 2u64.to_be_bytes();
        tree.put(&dead_key, &10u64.to_be_bytes()).unwrap();
        tree.put(&dead_key, &11u64.to_be_bytes()).unwrap();
        tree.put(&lost_key, &20u64.to_be_bytes()).unwrap();
        tree.durability.engine.shards()[0]
            .rotate_active_for_test(8)
            .unwrap();
        (dir, tree, lost_key)
    }

    fn newest_immutable_file_id(tree: &CompactionTestTree) -> u32 {
        tree.durability.engine.shards()[0]
            .lock()
            .immutable
            .iter()
            .map(|file| file.file_id)
            .max()
            .unwrap()
    }

    fn logical_data_len(shard_dir: &std::path::Path, file_id: u32) -> u64 {
        std::fs::metadata(shard_dir.join(format!("{file_id:06}.data")))
            .unwrap()
            .len()
    }

    fn assert_hint_entries_tile_plain_file(
        entries: &[crate::hint::HintEntry<[u8; 8]>],
        logical_end: u64,
    ) {
        assert!(
            crate::hint::entries_tile_logical_file(entries, logical_end, |_start, _end| Ok(false),)
                .unwrap()
        );
    }

    fn force_lost_cas_copy(tree: &std::sync::Arc<CompactionTestTree>, lost_key: [u8; 8]) -> u32 {
        let (entered_tx, entered_rx) = std::sync::mpsc::sync_channel(0);
        let (resume_tx, resume_rx) = std::sync::mpsc::sync_channel(0);
        let worker_tree = tree.clone();
        let worker = std::thread::spawn(move || {
            crate::compaction::test_pause_before_deferred_cas(entered_tx, resume_rx);
            worker_tree.compact()
        });
        entered_rx.recv().unwrap();
        assert!(tree.delete(&lost_key).unwrap().is_some());
        resume_tx.send(()).unwrap();
        worker.join().unwrap().unwrap();
        newest_immutable_file_id(tree)
    }

    fn rotate_tombstone_source_and_make_it_selectable(
        tree: &std::sync::Arc<CompactionTestTree>,
        key: [u8; 8],
        high_output_id: u32,
    ) -> (u32, u64) {
        let peer = 3u64.to_be_bytes();
        tree.put(&peer, &1u64.to_be_bytes()).unwrap();
        tree.put(&peer, &2u64.to_be_bytes()).unwrap();
        tree.durability.engine.shards()[0]
            .rotate_active_for_test(8)
            .unwrap();
        let source_id = tree.durability.engine.shards()[0]
            .lock()
            .immutable
            .iter()
            .filter(|file| file.file_id < high_output_id)
            .map(|file| file.file_id)
            .max()
            .unwrap();
        let tombstone_seq = tombstone_sequence_in_file(
            tree.durability.engine.shards()[0].dir().parent().unwrap(),
            source_id,
            key,
        );
        (source_id, tombstone_seq)
    }

    fn configure_second_pass_to_leave_file_uncovered(
        tree: &std::sync::Arc<CompactionTestTree>,
        high_output_id: u32,
    ) {
        let mut inner = tree.durability.engine.shards()[0].lock();
        inner.compaction_max_files_per_pass = 1;
        assert!(
            inner
                .immutable
                .iter()
                .any(|file| file.file_id == high_output_id),
            "high failed-CAS output must be immutable before the second pass"
        );
    }

    #[test]
    fn physical_hint_after_lost_cas_tiles_output_and_lists_copy() {
        let (dir, tree, lost_key) = compaction_tree_with_selectable_lost_cas_put();
        let tree = std::sync::Arc::new(tree);
        let (entered_tx, entered_rx) = std::sync::mpsc::sync_channel(0);
        let (resume_tx, resume_rx) = std::sync::mpsc::sync_channel(0);
        let worker_tree = tree.clone();
        let worker = std::thread::spawn(move || {
            crate::compaction::test_pause_before_deferred_cas(entered_tx, resume_rx);
            worker_tree.compact()
        });

        entered_rx.recv().unwrap();
        assert!(tree.delete(&lost_key).unwrap().is_some());
        resume_tx.send(()).unwrap();
        worker.join().unwrap().unwrap();

        let shard_dir = dir.path().join("shard_000");
        let output_id = newest_immutable_file_id(&tree);
        let hint = std::fs::read(shard_dir.join(format!("{output_id:06}.hint"))).unwrap();
        let (header, _) = crate::hint::validate_hint_file(&hint, 8).unwrap();
        let entries: Vec<_> = crate::hint::parse_hint_entries::<[u8; 8]>(&hint).collect();
        assert!(entries.iter().any(|entry| entry.key == lost_key));
        assert_eq!(header.logical_end, logical_data_len(&shard_dir, output_id));
        assert_hint_entries_tile_plain_file(&entries, header.logical_end);
    }

    #[test]
    fn physical_hint_after_lost_cas_survives_reopen_accounting() {
        let (dir, tree, lost_key) = compaction_tree_with_selectable_lost_cas_put();
        let tree = std::sync::Arc::new(tree);
        let output_id = force_lost_cas_copy(&tree, lost_key);
        drop(tree);

        let reopened = CompactionTestTree::open(dir.path(), compaction_hint_config()).unwrap();
        assert_eq!(reopened.get(&lost_key), None);
        let dead = reopened.durability.engine.shards()[0]
            .lock()
            .dead_bytes
            .get(&output_id)
            .copied()
            .unwrap_or(0);
        assert_eq!(dead, crate::entry::entry_size(8, 8));
    }

    #[test]
    fn tombstone_survives_high_id_failed_cas_copy_with_hint_and_full_scan_reopen() {
        let (dir, tree, key) = compaction_tree_with_selectable_lost_cas_put();
        let tree = std::sync::Arc::new(tree);
        let original_source_id = newest_immutable_file_id(&tree);
        let g1 = sequence_for_key_in_file(dir.path(), original_source_id, key);
        let high_output_id = force_lost_cas_copy(&tree, key);
        let (tombstone_source_id, g2) =
            rotate_tombstone_source_and_make_it_selectable(&tree, key, high_output_id);
        assert!(tombstone_source_id < high_output_id);
        assert!(g1 < g2, "stale copied put must precede the tombstone");
        configure_second_pass_to_leave_file_uncovered(&tree, high_output_id);
        tree.compact().unwrap();
        assert!(
            !data_file_exists(dir.path(), tombstone_source_id),
            "the lower-id tombstone source must be selected and retired"
        );
        assert!(
            data_file_exists(dir.path(), high_output_id),
            "the high failed-CAS output must remain outside the budget-one pass"
        );
        let high_min = tree.durability.engine.shards()[0]
            .lock()
            .immutable
            .iter()
            .find(|file| file.file_id == high_output_id)
            .unwrap()
            .seqs
            .min;
        assert!(
            matches!(high_min, crate::shard::MinSeq::Known(sequence) if sequence <= g1 && g1 < g2)
        );
        assert_eq!(tree.get(&key), None);
        assert_eq!(count_physical_tombstones(dir.path(), key), 1);
        drop(tree);
        let hinted = CompactionTestTree::open(dir.path(), tombstone_gc_config()).unwrap();
        assert_eq!(hinted.get(&key), None);
        drop(hinted);
        for entry in std::fs::read_dir(dir.path().join("shard_000")).unwrap() {
            let path = entry.unwrap().path();
            if path
                .extension()
                .is_some_and(|extension| extension == "hint")
            {
                std::fs::remove_file(path).unwrap();
            }
        }
        let scanned = CompactionTestTree::open(dir.path(), tombstone_gc_config()).unwrap();
        assert_eq!(scanned.get(&key), None);
    }

    fn run_compaction_model(
        ops: Vec<(u8, u8, u16)>,
    ) -> Result<(), proptest::test_runner::TestCaseError> {
        use proptest::prelude::*;
        let dir = tempfile::tempdir().unwrap();
        let config = tombstone_gc_config();
        let mut tree = Some(CompactionTestTree::open(dir.path(), config.clone()).unwrap());
        let mut model = std::collections::BTreeMap::<[u8; 8], [u8; 8]>::new();
        for (action, raw_key, value) in ops {
            let key = (raw_key as u64).to_be_bytes();
            match action {
                0 => {
                    tree.as_ref()
                        .unwrap()
                        .put(&key, &(value as u64).to_be_bytes())
                        .unwrap();
                    model.insert(key, (value as u64).to_be_bytes());
                }
                1 => {
                    tree.as_ref().unwrap().delete(&key).unwrap();
                    model.remove(&key);
                }
                2 => {
                    tree.as_ref().unwrap().compact().unwrap();
                }
                _ => {
                    drop(tree.take());
                    tree = Some(CompactionTestTree::open(dir.path(), config.clone()).unwrap());
                }
            }
            for key in 0u64..8 {
                prop_assert_eq!(
                    tree.as_ref().unwrap().get(&key.to_be_bytes()),
                    model.get(&key.to_be_bytes()).copied()
                );
            }
        }
        drop(tree.take());
        let reopened = CompactionTestTree::open(dir.path(), config).unwrap();
        for key in 0u64..8 {
            prop_assert_eq!(
                reopened.get(&key.to_be_bytes()),
                model.get(&key.to_be_bytes()).copied()
            );
        }
        Ok(())
    }

    fn active_file_with_legacy_incomplete_hint() -> (tempfile::TempDir, Config) {
        let dir = tempfile::tempdir().unwrap();
        let config = tombstone_gc_config();
        let tree = CompactionTestTree::open(dir.path(), config.clone()).unwrap();
        for key in 1u64..=3 {
            tree.put(&key.to_be_bytes(), &key.to_be_bytes()).unwrap();
        }
        let active_id = tree.durability.engine.shards()[0].lock().active.file_id;
        tree.close().unwrap();
        let hint_path = dir
            .path()
            .join("shard_000")
            .join(format!("{active_id:06}.hint"));
        let hint = std::fs::read(&hint_path).unwrap();
        let (header, payload) = crate::hint::validate_hint_file(&hint, 8).unwrap();
        let entry_size = crate::hint::hint_entry_size(8);
        let mut partial = Vec::with_capacity(entry_size * 2);
        partial.extend_from_slice(&payload[..entry_size]);
        partial.extend_from_slice(&payload[entry_size * 2..entry_size * 3]);
        let partial = crate::hint::hint_file_bytes(&partial, header.logical_end, 8).unwrap();
        crate::hint::write_hint_file(&hint_path, &partial).unwrap();
        (dir, config)
    }

    /// A legacy survivor-only hint on an *immutable* file, reached through a
    /// real open rather than by poking shard state: the upgrade path for a
    /// database whose compaction outputs were written before hints became
    /// physical.
    fn immutable_file_with_legacy_incomplete_hint() -> (tempfile::TempDir, Config, [u8; 8], u32) {
        let dir = tempfile::tempdir().unwrap();
        let config = tombstone_gc_config();
        let deleted_key = 9u64.to_be_bytes();
        {
            let tree = CompactionTestTree::open(dir.path(), config.clone()).unwrap();
            for key in 1u64..=3 {
                tree.put(&key.to_be_bytes(), &key.to_be_bytes()).unwrap();
            }
            tree.durability.engine.shards()[0]
                .rotate_active_for_test(8)
                .unwrap();
            tree.put(&deleted_key, &10u64.to_be_bytes()).unwrap();
            tree.delete(&deleted_key).unwrap();
            tree.durability.engine.shards()[0]
                .rotate_active_for_test(8)
                .unwrap();
            tree.close().unwrap();
        }
        let legacy_id = 1;
        let hint_path = dir
            .path()
            .join("shard_000")
            .join(format!("{legacy_id:06}.hint"));
        let hint = std::fs::read(&hint_path).unwrap();
        let (header, payload) = crate::hint::validate_hint_file(&hint, 8).unwrap();
        let entry_size = crate::hint::hint_entry_size(8);
        // Drop the middle entry: the hint still replays and still ends at
        // `logical_end`, so only the structural tiling check can tell that it
        // no longer accounts for the whole file.
        let mut partial = Vec::with_capacity(entry_size * 2);
        partial.extend_from_slice(&payload[..entry_size]);
        partial.extend_from_slice(&payload[entry_size * 2..entry_size * 3]);
        let partial = crate::hint::hint_file_bytes(&partial, header.logical_end, 8).unwrap();
        crate::hint::write_hint_file(&hint_path, &partial).unwrap();
        (dir, config, deleted_key, legacy_id)
    }

    #[test]
    fn legacy_immutable_hint_keeps_tombstones_across_open() {
        let (dir, config, deleted_key, legacy_id) = immutable_file_with_legacy_incomplete_hint();
        let tree = CompactionTestTree::open(dir.path(), config).unwrap();
        let seqs = tree.durability.engine.shards()[0]
            .lock()
            .immutable
            .iter()
            .find(|file| file.file_id == legacy_id)
            .expect("legacy file is immutable after reopen")
            .seqs;
        assert_eq!(
            seqs,
            crate::shard::FileSeqs::UNKNOWN,
            "an incomplete hint must not establish physical provenance"
        );
        assert_eq!(tree.get(&1u64.to_be_bytes()), Some(1u64.to_be_bytes()));
        assert_eq!(tree.get(&3u64.to_be_bytes()), Some(3u64.to_be_bytes()));
        assert_eq!(
            tree.dead_bytes_for_test(legacy_id),
            0,
            "an omitted physical entry is unproven, not recovered padding"
        );

        // Unknown provenance keeps the tombstone: the omitted entry could be
        // an older copy that compaction must not discard.
        tree.compact().unwrap();
        assert_eq!(count_physical_tombstones(dir.path(), deleted_key), 1);
        assert_eq!(tree.get(&deleted_key), None);
    }

    #[test]
    fn append_after_reopen_keeps_unknown_active_minimum() {
        let (dir, config) = active_file_with_legacy_incomplete_hint();
        let tree = CompactionTestTree::open(dir.path(), config).unwrap();
        assert_eq!(
            tree.durability.engine.shards()[0].lock().active.seqs,
            crate::shard::FileSeqs::UNKNOWN
        );
        tree.put(&9u64.to_be_bytes(), &9u64.to_be_bytes()).unwrap();
        assert_eq!(
            tree.durability.engine.shards()[0].lock().active.seqs,
            crate::shard::FileSeqs::UNKNOWN
        );
    }

    proptest::proptest! {
        #![proptest_config(proptest::test_runner::Config::with_cases(32))]
        #[test]
        fn compaction_tombstone_gc_matches_model(
            ops in proptest::collection::vec((0u8..4, 0u8..8, proptest::prelude::any::<u16>()), 1..80)
        ) { run_compaction_model(ops)?; }
    }
    /// F8: hint generation is asynchronous, so a *clean* shutdown is what
    /// re-establishes the guarantee it relaxes — after `clean_shutdown` returns,
    /// every retired file's hint is on disk. Only a crash leaves the window open,
    /// and a crash already implies a recovery scan.
    #[test]
    fn clean_shutdown_waits_for_queued_hint_generation() {
        use crate::shard::HintGate;
        use std::sync::Arc;
        use std::time::Duration;

        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 64 * 1024;
        cfg.write_buffer_size = 16 * 1024;
        let tree = ConstTree::<[u8; 8], 512>::open(dir.path(), cfg).unwrap();

        let gate = Arc::new(HintGate::closed());
        for shard in tree.durability.engine.shards().iter() {
            shard.lock().test_hint_gate = Some(gate.clone());
        }

        for i in 0..400u64 {
            tree.put(&i.to_be_bytes(), &[0xAB; 512]).expect("put");
        }

        // Release the parked worker shortly after the shutdown starts: a
        // shutdown that does not wait returns before the hint is written.
        let opener = {
            let gate = gate.clone();
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(300));
                gate.open();
            })
        };

        // The *retired* files are the ones whose hints the worker owns;
        // `sync_hints` covers only the active file, so assert on these.
        let retired: Vec<u32> = {
            let inner = tree.durability.engine.shards()[0].lock();
            inner.immutable.iter().map(|f| f.file_id).collect()
        };
        assert!(!retired.is_empty(), "test wrote too little to rotate");

        tree.clean_shutdown().expect("clean shutdown");

        // Checked *before* joining the opener: a shutdown that waits has
        // already outlived the gate by the time it returns, while one that does
        // not returns with the worker still parked.
        let missing: Vec<u32> = retired
            .iter()
            .copied()
            .filter(|id| !dir.path().join(format!("shard_000/{id:06}.hint")).exists())
            .collect();
        assert!(
            missing.is_empty(),
            "clean shutdown returned before the queued hints were written: {missing:?} missing"
        );
        opener.join().expect("gate opener");
    }

    /// F8: compaction deletes `{id}.data` together with `{id}.hint`. A hint
    /// generated in the background for a file compaction has meanwhile retired
    /// must not be published — a `{id}.hint` with no `{id}.data` is permanent
    /// garbage no later pass looks at. The worker takes the shard lock (the one
    /// compaction holds while deleting) to check the file is still an immutable
    /// member before renaming.
    #[test]
    fn compacted_away_file_gets_no_orphan_hint() {
        use crate::shard::HintGate;
        use std::sync::Arc;

        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 64 * 1024;
        cfg.write_buffer_size = 16 * 1024;
        cfg.compaction_threshold = 0.30;
        let tree = ConstTree::<[u8; 8], 512>::open(dir.path(), cfg).unwrap();

        // Hold every generation back so the jobs are still queued when
        // compaction runs.
        let gate = Arc::new(HintGate::closed());
        for shard in tree.durability.engine.shards().iter() {
            shard.lock().test_hint_gate = Some(gate.clone());
        }

        // Churn: rewrite a small key set until the retired files are mostly dead
        // bytes, which is what makes compaction retire them.
        for _ in 0..6 {
            for i in 0..60u64 {
                tree.put(&i.to_be_bytes(), &[0xCD; 512]).expect("put");
            }
        }
        let retired: Vec<u32> = {
            let inner = tree.durability.engine.shards()[0].lock();
            inner.immutable.iter().map(|f| f.file_id).collect()
        };
        assert!(!retired.is_empty(), "test wrote too little to rotate");

        tree.compact().expect("compact");
        let surviving: Vec<u32> = {
            let inner = tree.durability.engine.shards()[0].lock();
            inner.immutable.iter().map(|f| f.file_id).collect()
        };
        let compacted_away: Vec<u32> = retired
            .iter()
            .copied()
            .filter(|id| !surviving.contains(id))
            .collect();
        assert!(
            !compacted_away.is_empty(),
            "compaction retired nothing — the test proves nothing"
        );

        gate.open();
        tree.clean_shutdown().expect("clean shutdown");

        let shard_dir = dir.path().join("shard_000");
        for id in compacted_away {
            let hint = shard_dir.join(format!("{id:06}.hint"));
            assert!(
                !hint.exists(),
                "orphan hint {} published for a file compaction already deleted",
                hint.display()
            );
        }
        let tmps: Vec<_> = std::fs::read_dir(&shard_dir)
            .expect("shard dir")
            .filter_map(|e| e.ok())
            .map(|e| e.path())
            .filter(|p| p.to_string_lossy().ends_with(".hint.tmp"))
            .collect();
        assert!(tmps.is_empty(), "temporary hints left behind: {tmps:?}");
    }

    /// F8: rotations can outrun the worker. When they do, the job is dropped —
    /// never blocked on — because a missing hint costs one file's scan at the
    /// next open, while blocking would reintroduce the very stall this design
    /// removes. The data itself must still come back intact.
    #[test]
    fn a_full_hint_queue_drops_jobs_instead_of_stalling_writes() {
        use crate::shard::HintGate;
        use std::sync::Arc;

        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 32 * 1024; // tiny files → a rotation every few writes
        cfg.write_buffer_size = 16 * 1024;
        let tree = ConstTree::<[u8; 8], 512>::open(dir.path(), cfg.clone()).unwrap();

        // Park the worker so nothing drains: every rotation past the queue's
        // capacity has to be dropped.
        let gate = Arc::new(HintGate::closed());
        for shard in tree.durability.engine.shards().iter() {
            shard.lock().test_hint_gate = Some(gate.clone());
        }

        const N: u64 = 1200; // ≫ QUEUE_CAPACITY rotations at 32 KiB per file
        for i in 0..N {
            tree.put(&i.to_be_bytes(), &[0xEF; 512]).expect("put");
        }

        assert!(
            tree.durability.engine.hint_jobs_dropped() > 0,
            "a queue this far behind must drop jobs, not grow without bound"
        );

        gate.open();
        tree.close().expect("close");

        // A dropped hint is a recovery cost, not a data loss: reopening rebuilds
        // the index by scanning the files that have no hint.
        let reopened = ConstTree::<[u8; 8], 512>::open(dir.path(), cfg).unwrap();
        for i in 0..N {
            assert_eq!(
                reopened.get(&i.to_be_bytes()),
                Some([0xEF; 512]),
                "key {i} lost after reopening without some hint files"
            );
        }
    }

    /// F8: dropping a collection must take its hint worker with it. The worker
    /// originally held an `Arc` back to the service that owns its sender, which
    /// made the service's refcount never reach zero — `Drop` was unreachable and
    /// the thread outlived the collection, silently, for the life of the process.
    #[test]
    fn dropping_a_collection_stops_its_hint_worker() {
        use crate::hint_worker::LIVE_WORKERS;
        use std::sync::atomic::Ordering;
        use std::time::{Duration, Instant};

        let before = LIVE_WORKERS.load(Ordering::Relaxed);

        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 64 * 1024;
        cfg.write_buffer_size = 16 * 1024;

        {
            let tree = ConstTree::<[u8; 8], 512>::open(dir.path(), cfg).unwrap();
            for i in 0..400u64 {
                tree.put(&i.to_be_bytes(), &[0xAB; 512]).expect("put");
            }
            // Spawning does not wait for the worker to enter its run loop.
            let deadline = Instant::now() + Duration::from_secs(5);
            while LIVE_WORKERS.load(Ordering::Relaxed) == before && Instant::now() < deadline {
                std::thread::yield_now();
            }
            assert_eq!(
                LIVE_WORKERS.load(Ordering::Relaxed),
                before + 1,
                "rotation should have started exactly one worker"
            );
        } // dropped without close(): Drop alone must stop the thread

        let deadline = Instant::now() + Duration::from_secs(5);
        while LIVE_WORKERS.load(Ordering::Relaxed) > before && Instant::now() < deadline {
            std::thread::yield_now();
        }
        assert_eq!(
            LIVE_WORKERS.load(Ordering::Relaxed),
            before,
            "hint worker outlived the collection that owns it"
        );
    }

    /// F8 (libfunc/armour#83): rotation retires a data file and then generates
    /// its hint — a full scan of up to `max_file_size` bytes. Doing that inside
    /// `rotate()` holds the shard lock for the whole scan, so the writer that
    /// happened to trigger the rotation (and every other writer to that shard)
    /// stalls on an optimization it does not need. The write must not wait for
    /// it.
    #[test]
    fn rotation_does_not_block_the_writer_on_hint_generation() {
        use crate::shard::HintGate;
        use std::sync::Arc;
        use std::sync::mpsc;
        use std::time::Duration;

        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 64 * 1024; // small files → rotation within a few writes
        cfg.write_buffer_size = 16 * 1024;
        let tree = Arc::new(ConstTree::<[u8; 8], 512>::open(dir.path(), cfg).unwrap());

        // Park hint generation: from here on, whoever runs it blocks until we
        // open the gate.
        let gate = Arc::new(HintGate::closed());
        for shard in tree.durability.engine.shards().iter() {
            shard.lock().test_hint_gate = Some(gate.clone());
        }

        // Write enough to force at least one rotation, from another thread so a
        // stall shows up as a timeout rather than hanging the test process.
        let (tx, rx) = mpsc::channel();
        let writer = {
            let tree = tree.clone();
            std::thread::spawn(move || {
                for i in 0..400u64 {
                    tree.put(&i.to_be_bytes(), &[0xAB; 512]).expect("put");
                }
                let _ = tx.send(());
            })
        };

        let finished = rx.recv_timeout(Duration::from_secs(10)).is_ok();
        // Let the parked generation through before asserting, so a failure ends
        // the test instead of leaving the writer wedged on the gate.
        gate.open();
        writer.join().expect("writer thread");

        assert!(
            finished,
            "writes stalled until hint generation completed — rotation is \
             generating the hint under the shard lock"
        );
    }
}

// ==========================================================================
// Tests — Fixed replication apply helpers
// ==========================================================================

#[cfg(all(test, feature = "replication"))]
mod replication_helper_tests {
    use crate::FixedConfig;
    use crate::fixed::FixedTree;
    use tempfile::tempdir;

    fn cfg() -> FixedConfig {
        FixedConfig {
            shard_count: 2,
            grow_step: 64,
            ..FixedConfig::test()
        }
    }

    #[test]
    fn get_slot_id_present_and_absent() {
        let dir = tempdir().unwrap();
        let tree = FixedTree::<[u8; 8], 8>::open(dir.path(), cfg()).unwrap();

        let key = 1u64.to_be_bytes();
        let value = 42u64.to_be_bytes();
        tree.put(&key, &value).unwrap();

        let slot = tree.get_slot_id(&key).expect("present key must resolve");
        // Slot id must be stable across a read-only lookup.
        assert_eq!(tree.get_slot_id(&key), Some(slot));

        let missing = 9999u64.to_be_bytes();
        assert_eq!(tree.get_slot_id(&missing), None);
    }

    #[test]
    fn remove_key_if_slot_matches_matching_and_nonmatching() {
        let dir = tempdir().unwrap();
        let tree = FixedTree::<[u8; 8], 8>::open(dir.path(), cfg()).unwrap();

        let key = 7u64.to_be_bytes();
        let value = 7u64.to_be_bytes();
        tree.put(&key, &value).unwrap();
        let slot = tree.get_slot_id(&key).unwrap();

        // Non-matching slot: removal must be refused, entry preserved.
        assert!(!tree.remove_key_if_slot_matches(&key, slot.wrapping_add(1)));
        assert!(tree.contains(&key));

        // Matching slot: removal succeeds.
        assert!(tree.remove_key_if_slot_matches(&key, slot));
        assert!(!tree.contains(&key));

        // Absent key: always false.
        assert!(!tree.remove_key_if_slot_matches(&key, slot));
    }

    #[test]
    fn upsert_replicated_insert_and_update() {
        let dir = tempdir().unwrap();
        let tree = FixedTree::<[u8; 8], 8>::open(dir.path(), cfg()).unwrap();

        let key = 3u64.to_be_bytes();
        let value_a = 100u64.to_be_bytes();
        let value_b = 200u64.to_be_bytes();

        // Insert path (absent key).
        tree.upsert_replicated(&key, value_a, 77);
        assert_eq!(tree.get(&key), Some(value_a));
        assert_eq!(tree.get_slot_id(&key), Some(77));

        // Update path at same slot (SeqLock write).
        tree.upsert_replicated(&key, value_b, 77);
        assert_eq!(tree.get(&key), Some(value_b));
        assert_eq!(tree.get_slot_id(&key), Some(77));

        // Update path with a new slot id (remove + insert fresh node).
        let value_c = 300u64.to_be_bytes();
        tree.upsert_replicated(&key, value_c, 123);
        assert_eq!(tree.get(&key), Some(value_c));
        assert_eq!(tree.get_slot_id(&key), Some(123));
    }
}