armdb 0.5.3

use crate::Key;
use crate::compaction::{CompactionIndex, compact_shard};
use crate::config::Config;
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::recovery::recover_const_tree;
use crate::skiplist::node::{ConstNode, SkipNode, random_height};
use crate::skiplist::{InsertResult, SkipList};
use crate::sync::MutexGuard;
use std::mem::size_of;
use std::ops::Bound;

/// 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>`).
///
/// 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`,
/// 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,
    hook: H,
}

// ==========================================================================
// 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.
    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> {
    /// Open or create a `ConstTree` with a write hook for secondary index maintenance.
    pub fn open_hooked(
        path: impl AsRef<std::path::Path>,
        config: Config,
        hook: H,
    ) -> DbResult<Self> {
        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,
        };

        // 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.cipher(),
        )?;
        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);
        }
        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)
    }

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

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

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

    /// 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.
    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.
    pub fn open_with_hook(
        path: impl AsRef<std::path::Path>,
        config: FixedConfig,
        hook: H,
    ) -> DbResult<Self> {
        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 total_recovered =
            dur.recover_entries(|_shard_id, key_bytes, value_bytes, slot_id| {
                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);
                let guard = index.collector().enter();
                let _ = index.insert(node, &guard);
            })?;

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

        Ok(Self {
            index,
            durability: dur,
            shard_prefix_bits,
            reversed,
            hook,
        })
    }

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

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

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

impl<K: Key, const V: usize, H: WriteHook<K>, D: Durability> ConstTree<K, V, H, D> {
    /// 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)
    }

    /// 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);
        let mut inner = self.durability.lock_shard(shard_id);
        let guard = self.index.collector().enter();
        let old = self.put_locked(shard_id, &mut *inner, &guard, key, value)?;
        let needs_sync = inner.should_sync();
        drop(inner);
        if needs_sync {
            self.durability.lock_shard(shard_id).sync()?;
        }
        self.hook
            .on_write(key, old.as_ref().map(|v| &v[..]), 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 inner = self.durability.lock_shard(shard_id);
        let guard = self.index.collector().enter();
        self.insert_locked(shard_id, &mut *inner, &guard, key, value)?;
        let needs_sync = inner.should_sync();
        drop(inner);
        if needs_sync {
            self.durability.lock_shard(shard_id).sync()?;
        }
        self.hook.on_write(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 inner = self.durability.lock_shard(shard_id);
        let guard = self.index.collector().enter();
        let old = self.delete_locked(shard_id, &mut *inner, &guard, key)?;
        let needs_sync = inner.should_sync();
        drop(inner);
        if needs_sync {
            self.durability.lock_shard(shard_id).sync()?;
        }
        if let Some(ref old_val) = old {
            self.hook.on_write(key, Some(&old_val[..]), None);
        }
        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 shard_id = self.shard_for(shard_key);
        let inner = self.durability.lock_shard(shard_id);
        let guard = self.index.collector().enter();
        let mut shard = ConstShard {
            tree: self,
            inner,
            shard_id,
            guard,
            events: Vec::new(),
        };
        let result = f(&mut shard);
        let ConstShard {
            inner,
            guard,
            events,
            ..
        } = shard;
        let needs_sync = inner.should_sync();
        drop(inner);
        if needs_sync {
            self.durability.lock_shard(shard_id).sync()?;
        }
        if H::NEEDS_WRITE {
            for (k, old, new) in &events {
                self.hook.on_write(
                    k,
                    old.as_ref().map(|v| &v[..]),
                    new.as_ref().map(|v| &v[..]),
                );
            }
        }
        drop(guard);
        result
    }

    fn put_locked(
        &self,
        shard_id: usize,
        inner: &mut D::Inner,
        guard: &seize::LocalGuard<'_>,
        key: &K,
        value: &[u8; V],
    ) -> DbResult<Option<[u8; V]>> {
        // 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();
            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);
            return Ok(Some(old_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 => 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);
                }
                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);
        }

        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 => 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();
            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);
            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 inner = self.durability.lock_shard(shard_id);
        let guard = self.index.collector().enter();
        let result = self.put_locked(shard_id, &mut *inner, &guard, key, value);
        let needs_sync = result.is_ok() && inner.should_sync();
        drop(inner);
        drop(guard);
        if needs_sync {
            self.durability.lock_shard(shard_id).sync()?;
        }
        result
    }

    fn delete_no_hook(&self, key: &K) -> DbResult<Option<[u8; V]>> {
        let shard_id = self.shard_for(key);
        let mut inner = self.durability.lock_shard(shard_id);
        let guard = self.index.collector().enter();
        let result = self.delete_locked(shard_id, &mut *inner, &guard, key);
        let needs_sync = result.is_ok() && inner.should_sync();
        drop(inner);
        drop(guard);
        if needs_sync {
            self.durability.lock_shard(shard_id).sync()?;
        }
        result
    }

    /// 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");
        let shard_id = self.shard_for(key);
        let mut inner = self.durability.lock_shard(shard_id);

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

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

        let needs_sync = inner.should_sync();
        drop(inner);
        if needs_sync {
            self.durability.lock_shard(shard_id).sync()?;
        }

        self.hook
            .on_write(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 inner = self.durability.lock_shard(shard_id);

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

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

        let needs_sync = inner.should_sync();
        drop(inner);
        if needs_sync {
            self.durability.lock_shard(shard_id).sync()?;
        }

        self.hook.on_write(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_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 inner = self.durability.lock_shard(shard_id);

        let guard = self.index.collector().enter();
        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);

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

        let needs_sync = inner.should_sync();
        drop(inner);
        if needs_sync {
            self.durability.lock_shard(shard_id).sync()?;
        }

        self.hook
            .on_write(key, Some(&old_value[..]), Some(&new_value[..]));
        Ok(Some(if return_old { old_value } else { new_value }))
    }

    /// 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 = 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), `next_back()` is O(log n).
    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()),
                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,
            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), `next_back()` is O(log n).
    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,
            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), `next_back()` is O(log n).
    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), `next_back()` is O(log n).
    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,
                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,
                end: bound_owned(&end),
                start: bound_owned(&start),
                reversed: false,
                done: false,
                _guard: guard,
            }
        }
    }

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

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

    /// 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> {
        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 H::NEEDS_INIT {
                        self.hook.on_init(&node.key, &value[..]);
                    }
                }
                MigrateAction::Update(value) => {
                    self.put_no_hook(&node.key, &value)?;
                    if 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()[..]);
            }
        }
    }

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

// ==========================================================================
// 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>()
    }
}

#[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
// ==========================================================================

type ConstShardEvent<K, const V: usize> = (K, Option<[u8; V]>, Option<[u8; V]>);

/// 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>,
    inner: MutexGuard<'a, D::Inner>,
    shard_id: usize,
    guard: seize::LocalGuard<'a>,
    events: Vec<ConstShardEvent<K, V>>,
}

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 old = self
            .tree
            .put_locked(self.shard_id, &mut *self.inner, &self.guard, key, value)?;
        if H::NEEDS_WRITE {
            self.events.push((*key, old, Some(*value)));
        }
        Ok(old)
    }

    pub fn insert(&mut self, key: &K, value: &[u8; V]) -> DbResult<()> {
        self.check_shard(key)?;
        self.tree
            .insert_locked(self.shard_id, &mut *self.inner, &self.guard, key, value)?;
        if H::NEEDS_WRITE {
            self.events.push((*key, None, Some(*value)));
        }
        Ok(())
    }

    pub fn delete(&mut self, key: &K) -> DbResult<Option<[u8; V]>> {
        self.check_shard(key)?;
        let old = self
            .tree
            .delete_locked(self.shard_id, &mut *self.inner, &self.guard, key)?;
        if H::NEEDS_WRITE
            && let Some(ref old_val) = old
        {
            self.events.push((*key, Some(*old_val), None));
        }
        Ok(old)
    }

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

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

    pub fn contains(&self, key: &K) -> bool {
        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.shard_id {
            return Err(DbError::ShardMismatch);
        }
        Ok(())
    }
}

// ==========================================================================
// MultiTx — cross-collection transaction support (feature `armour`)
// ==========================================================================

/// Multi-shard transaction handle for [`ConstTree`] inside `Db::atomicN`.
/// Holds one durability guard per locked shard, a single epoch guard, and one
/// collection-wide event log appended in closure order.
#[cfg(feature = "armour")]
pub struct ConstTx<'a, K: Key, const V: usize, H: WriteHook<K> = NoHook, D: Durability = Bitcask> {
    tree: &'a ConstTree<K, V, H, D>,
    inners: Vec<(usize, MutexGuard<'a, D::Inner>)>,
    seize: seize::LocalGuard<'a>,
    log: Vec<ConstShardEvent<K, V>>,
}

#[cfg(feature = "armour")]
impl<'a, K: Key, const V: usize, H: WriteHook<K>, D: Durability> ConstTx<'a, K, V, H, D> {
    fn position(&self, key: &K) -> DbResult<usize> {
        let sid = self.tree.shard_for(key);
        self.inners
            .iter()
            .position(|(s, _)| *s == sid)
            .ok_or(DbError::ShardMismatch)
    }

    pub fn try_get(&self, key: &K) -> DbResult<Option<[u8; V]>> {
        self.position(key)?;
        Ok(self
            .tree
            .index
            .get(key.as_bytes(), &self.seize)
            .map(|n| n.read_value()))
    }

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

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

    pub fn put(&mut self, key: &K, value: &[u8; V]) -> DbResult<Option<[u8; V]>> {
        let i = self.position(key)?;
        let (sid, inner) = &mut self.inners[i];
        let old = self
            .tree
            .put_locked(*sid, &mut **inner, &self.seize, key, value)?;
        if H::NEEDS_WRITE {
            self.log.push((*key, old, Some(*value)));
        }
        Ok(old)
    }

    pub fn insert(&mut self, key: &K, value: &[u8; V]) -> DbResult<()> {
        let i = self.position(key)?;
        let (sid, inner) = &mut self.inners[i];
        self.tree
            .insert_locked(*sid, &mut **inner, &self.seize, key, value)?;
        if H::NEEDS_WRITE {
            self.log.push((*key, None, Some(*value)));
        }
        Ok(())
    }

    pub fn delete(&mut self, key: &K) -> DbResult<Option<[u8; V]>> {
        let i = self.position(key)?;
        let (sid, inner) = &mut self.inners[i];
        let old = self
            .tree
            .delete_locked(*sid, &mut **inner, &self.seize, key)?;
        if H::NEEDS_WRITE
            && let Some(ref old_val) = old
        {
            self.log.push((*key, Some(*old_val), None));
        }
        Ok(old)
    }
}

#[cfg(feature = "armour")]
impl<K: Key, const V: usize, H: WriteHook<K>, D: Durability> crate::armour::multi_tx::MultiTx
    for ConstTree<K, V, H, D>
{
    type Key = K;
    type Tx<'a>
        = ConstTx<'a, K, V, H, D>
    where
        Self: 'a;

    fn shard_for_key(&self, key: &K) -> usize {
        self.shard_for(key)
    }

    fn begin_tx(&self) -> ConstTx<'_, K, V, H, D> {
        ConstTx {
            tree: self,
            inners: Vec::new(),
            seize: self.index.collector().enter(),
            log: Vec::new(),
        }
    }

    fn lock_shard_into<'a>(&'a self, shard_id: usize, tx: &mut ConstTx<'a, K, V, H, D>) {
        tx.inners
            .push((shard_id, self.durability.lock_shard(shard_id)));
    }

    fn release_locks(
        &self,
        tx: &mut ConstTx<'_, K, V, H, D>,
    ) -> crate::armour::multi_tx::SyncNeeds {
        let mut needs = crate::armour::multi_tx::SyncNeeds::none();
        for (sid, inner) in &tx.inners {
            if inner.should_sync() {
                needs.push(*sid);
            }
        }
        tx.inners.clear(); // drops MutexGuards → releases shard locks
        needs
    }

    fn run_sync(&self, needs: crate::armour::multi_tx::SyncNeeds) -> DbResult<()> {
        for &sid in needs.shards() {
            self.durability.lock_shard(sid).sync()?;
        }
        Ok(())
    }

    fn replay_hooks(&self, tx: ConstTx<'_, K, V, H, D>) {
        if H::NEEDS_WRITE {
            for (k, old, new) in &tx.log {
                self.hook.on_write(
                    k,
                    old.as_ref().map(|v| &v[..]),
                    new.as_ref().map(|v| &v[..]),
                );
            }
        }
    }
}

// ==========================================================================
// 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,
    }
}

fn prefix_to_end_bound<K: Key>(prefix: &[u8]) -> Bound<K> {
    let mut incremented = prefix.to_vec();
    let mut carry = true;
    for byte in incremented.iter_mut().rev() {
        if carry {
            if *byte == 0xFF {
                *byte = 0x00;
            } else {
                *byte += 1;
                carry = false;
                break;
            }
        }
    }
    if carry {
        Bound::Unbounded
    } else {
        let mut end = K::zeroed();
        end.as_bytes_mut()[..incremented.len()].copy_from_slice(&incremented);
        Bound::Excluded(end)
    }
}

// ==========================================================================
// 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>>,
    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.find_last_lt(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(&self) -> *mut ConstNode<K, V, L> {
        match &self.end {
            Bound::Unbounded => self.list.find_last(&self._guard),
            Bound::Excluded(k) => self.list.find_last_lt(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.find_last_lt(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;
    use crate::fixed::FixedTree;
    use crate::hook::WriteHook;
    use crate::{Config, DbError, DbResult, FixedConfig};
    use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};
    use tempfile::tempdir;

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

    #[cfg(feature = "armour")]
    #[test]
    fn const_tx_routes_get_put_and_rejects_out_of_scope() {
        use crate::armour::MultiTx;
        let dir = tempfile::tempdir().unwrap();
        // open_const_hooked forces shard_count=1; the out-of-scope branch below is
        // guarded so it simply does not run here (Task 14 exercises multi-shard).
        let tree = open_const_hooked(dir.path(), RecHook::default());
        let k1 = [1u8; 8];
        let k2 = [2u8; 8];
        tree.put(&k1, &[10u8; 4]).unwrap();
        // The seeding put above fires the hook once; reset so we can assert the
        // replay fires exactly once for the in-tx put of k2.
        tree.hook.writes.store(0, AtomicOrdering::Relaxed);

        let s1 = tree.shard_for(&k1);
        let s2 = tree.shard_for(&k2);
        let mut tx = tree.begin_tx();
        tree.lock_shard_into(s1, &mut tx);
        if s2 != s1 {
            tree.lock_shard_into(s2, &mut tx);
        }

        assert_eq!(tx.try_get(&k1).unwrap(), Some([10u8; 4]));
        assert!(!tx.try_contains(&k2).unwrap());
        tx.put(&k2, &[20u8; 4]).unwrap();
        assert_eq!(tx.try_get(&k2).unwrap(), Some([20u8; 4]));

        // A key whose shard was not locked → ShardMismatch on every accessor.
        let mut unlocked_key = [0u8; 8];
        for b in 3u8..=255 {
            unlocked_key[0] = b;
            let s = tree.shard_for(&unlocked_key);
            if s != s1 && s != s2 {
                break;
            }
        }
        let s_un = tree.shard_for(&unlocked_key);
        if s_un != s1 && s_un != s2 {
            assert!(matches!(
                tx.try_get(&unlocked_key),
                Err(DbError::ShardMismatch)
            ));
            assert!(matches!(
                tx.try_contains(&unlocked_key),
                Err(DbError::ShardMismatch)
            ));
            assert!(matches!(
                tx.put(&unlocked_key, &[0u8; 4]),
                Err(DbError::ShardMismatch)
            ));
        }

        let needs = tree.release_locks(&mut tx);
        tree.run_sync(needs).unwrap();
        tree.replay_hooks(tx);

        // Hook fired once (the put of k2); k2 persisted.
        assert_eq!(tree.get(&k2), Some([20u8; 4]));
        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 1);
    }

    // R4: a single collection touched on >=2 shards in one transaction replays
    // hooks in CLOSURE order, not shard order — proving the collection-wide log
    // (not per-shard buffers) drives ordering.
    #[cfg(feature = "armour")]
    #[test]
    fn cross_shard_hook_order_is_closure_order() {
        use crate::armour::Db;
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();

        // Hooked tree with 4 shards so two keys can land on different shards.
        let mut cfg = Config::test();
        cfg.shard_count = 4;
        let tree: ConstTree<[u8; 8], 4, RecHook> =
            ConstTree::open_hooked(dir.path().join("hooked"), cfg, RecHook::default())
                .expect("open hooked");
        // A second, distinct collection to satisfy atomic2's two-collection rule.
        let mut cfg2 = Config::test();
        cfg2.shard_count = 4;
        let other: ConstTree<[u8; 8], 4> =
            ConstTree::<[u8; 8], 4>::open(dir.path().join("other"), cfg2).expect("open other");

        // Find two keys routing to different shards.
        let k_lo = [1u8; 8];
        let mut k_hi = [2u8; 8];
        for b in 2u8..=255 {
            k_hi[0] = b;
            if tree.shard_for(&k_hi) != tree.shard_for(&k_lo) {
                break;
            }
        }
        assert_ne!(
            tree.shard_for(&k_hi),
            tree.shard_for(&k_lo),
            "need two shards"
        );
        let ko = [7u8; 8];

        // Write k_hi THEN k_lo (closure order); they live on different shards.
        db.atomic2(&tree, &[k_lo, k_hi], &other, &[ko], |t, _o| {
            t.put(&k_hi, &[1u8; 4])?;
            t.put(&k_lo, &[2u8; 4])?;
            Ok(())
        })
        .unwrap();

        let seq = crate::sync::lock(&tree.hook.seq).clone();
        let order: Vec<u64> = seq.iter().map(|(k, _, _)| *k).collect();
        assert_eq!(
            order,
            vec![u64::from_be_bytes(k_hi), u64::from_be_bytes(k_lo)],
            "hooks must replay in closure order, not shard order"
        );
    }

    #[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() {
        // FixedStore is the one family where `should_sync()` can be 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 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 — 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));
    }
}