armdb 0.6.0

use std::hash::Hash;
use std::mem::size_of;
use std::ops::Bound;

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::map_index::{
    Dir, KeyPage, MapIterView, Page, ShardIndex, ShardSource, dir_from, key_index_batch,
};
use crate::recovery::recover_const_map;
use crate::sync::{self, Mutex, MutexGuard};

pub(crate) struct MapEntry<const V: usize, L: Location> {
    pub(crate) loc: L,
    pub(crate) value: [u8; V],
}

/// A map with fixed-size keys and values. All values are stored inline in a per-shard HashMap.
/// Reads never touch disk — zero I/O reads. O(1) lookup instead of O(log n) SkipList.
/// Ordered iteration is available when opened with `Config::iterable(true)` via
/// [`iter_view`](Self::iter_view); otherwise use `ConstTree` for prefix/range scans.
///
/// Generic over `D: Durability` (default: [`Bitcask`]). Use [`Fixed`] backend for
/// frequent updates without compaction (`FixedMap` is a type alias for `ConstMap<K, V, Fixed>`).
///
/// Each `ConstMap` owns its storage engine — one map = one database directory.
///
/// # Write hooks
///
/// Uses [`WriteHook<K>`]. `on_write` fires on `put`/`insert`/`delete`/`cas`/`compare_delete`/`update`/`fetch_update`,
/// and on each mutation inside `atomic()` (replayed after the lock is released, in application order).
/// `on_init` fires once per live entry
/// during `migrate()` or `replay_init()`. Old value is always provided in
/// `on_write` — `NEEDS_OLD_VALUE` is ignored (value is inline, zero cost).
pub struct ConstMap<
    K: Key + Send + Sync + Hash + Eq,
    const V: usize,
    H: WriteHook<K> = NoHook,
    D: Durability = Bitcask,
> {
    indexes: Vec<Mutex<ShardIndex<K, MapEntry<V, D::Loc>>>>,
    durability: D,
    shard_prefix_bits: usize,
    hook: H,
    iterable: bool,
    reversed: bool,
}

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

impl<K: Key + Send + Sync + Hash + Eq, const V: usize> ConstMap<K, V, NoHook, Bitcask> {
    /// Open or create a `ConstMap` 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 + Send + Sync + Hash + Eq, const V: usize, H: WriteHook<K>> ConstMap<K, V, H, Bitcask> {
    /// Open or create a `ConstMap` 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 iterable = config.iterable;
        let reversed = config.reversed;
        let engine = Engine::open(path, config)?;

        let shard_count = engine.shards().len();
        let mut indexes = Vec::with_capacity(shard_count);
        for _ in 0..shard_count {
            // Recover without the companion even when `iterable`: it is built once
            // via `set_iterable_and_rebuild` after recovery (below), avoiding a
            // redundant incremental build during the per-entry `upsert` pass.
            indexes.push(Mutex::new(ShardIndex::new(false)));
        }

        let durability = Bitcask {
            engine,
            compaction_threshold,
        };

        let map = Self {
            indexes,
            durability,
            shard_prefix_bits,
            hook,
            iterable,
            reversed,
        };

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

        let hints = map.durability.engine.hints();
        let outcome = recover_const_map::<K, V>(
            &shard_dir_refs,
            &shard_ids,
            map.indexes(),
            hints,
            #[cfg(feature = "encryption")]
            map.durability.engine.cipher(),
        )?;
        for tail in &outcome.active_tails {
            map.durability.engine.shards()[tail.shard_idx].apply_recovery_tail(tail)?;
        }
        for (shard_idx, dead) in outcome.shard_dead_bytes {
            map.durability.engine.shards()[shard_idx].install_dead_bytes(dead);
        }
        let max_gsn = outcome.max_gsn;

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

        Ok(map)
    }

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

    pub(crate) fn indexes(&self) -> &[Mutex<ShardIndex<K, MapEntry<V, DiskLoc>>>] {
        &self.indexes
    }
}

impl<K: Key + Send + Sync + Hash + Eq, const V: usize, H: WriteHook<K>> CompactionIndex<K>
    for ConstMap<K, V, H, Bitcask>
{
    fn update_if_match(&self, key: &K, old_loc: DiskLoc, new_loc: DiskLoc) -> bool {
        let mut index = sync::lock(&self.indexes[self.shard_for(key)]);
        if let Some(entry) = index.get_mut(key)
            && entry.loc == old_loc
        {
            entry.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 index = sync::lock(&self.indexes[shard_id as usize]);
        index.get(key).is_some_and(|e| e.loc == loc)
    }
}

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

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

impl<K: Key + Send + Sync + Hash + Eq, const V: usize> ConstMap<K, V, NoHook, Fixed> {
    /// Open or create a `ConstMap` 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 + Send + Sync + Hash + Eq, const V: usize, H: WriteHook<K>> ConstMap<K, V, H, Fixed> {
    /// Open or create a `ConstMap` 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 iterable = config.iterable;
        let reversed = config.reversed;
        let dur = Fixed::open(path, config, std::mem::size_of::<K>(), V)?;

        let shard_count = dur.shard_count();
        let mut indexes = Vec::with_capacity(shard_count);
        for _ in 0..shard_count {
            // Built once after recovery via `set_iterable_and_rebuild` — see the
            // Bitcask `open_inner` for the rationale.
            indexes.push(Mutex::new(ShardIndex::new(false)));
        }

        let total_recovered = dur.recover_entries(|shard_idx| {
            // Hold this shard's index lock for the whole shard recovery.
            let mut guard = sync::lock(&indexes[shard_idx]);
            move |key_bytes: &[u8], value_bytes: &[u8], slot_id: u32| {
                let key = K::from_bytes(key_bytes);
                let mut value = [0u8; V];
                value.copy_from_slice(value_bytes);
                guard.upsert(
                    key,
                    MapEntry {
                        loc: slot_id,
                        value,
                    },
                );
            }
        })?;

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

        if iterable {
            for idx in &indexes {
                sync::lock(idx).set_iterable_and_rebuild();
            }
        }

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

    /// 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 + Send + Sync + Hash + Eq, const V: usize, H: WriteHook<K>> ConstMap<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 map'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(1).
    ///
    /// 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 index = sync::lock(&self.indexes[self.shard_for(key)]);
        index.get(key).map(|e| e.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.
    pub(crate) fn remove_key_if_slot_matches(&self, key: &K, slot_id: u32) -> bool {
        let mut index = sync::lock(&self.indexes[self.shard_for(key)]);
        match index.get(key) {
            Some(entry) if entry.loc == slot_id => {
                index.remove(key);
                true
            }
            _ => false,
        }
    }

    /// Upsert `key → (value, slot_id)` in the in-memory index. Bypasses
    /// durability — replication has already rewritten the slot on disk.
    pub(crate) fn upsert_replicated(&self, key: &K, value: [u8; V], slot_id: u32) {
        let mut index = sync::lock(&self.indexes[self.shard_for(key)]);
        index.upsert(
            *key,
            MapEntry {
                loc: slot_id,
                value,
            },
        );
    }
}

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

impl<K: Key + Send + Sync + Hash + Eq, const V: usize, H: WriteHook<K>, D: Durability>
    ConstMap<K, V, H, D>
{
    /// Configured iteration direction (`true` = DESC / newest-first).
    /// Backend-agnostic — `config()` exposes `reversed` only on Bitcask, so
    /// wrappers (`ZeroMap`) read direction through this on either backend.
    pub(crate) fn reversed(&self) -> bool {
        self.reversed
    }

    /// Get a value by key. O(1) HashMap lookup, zero disk I/O.
    /// Takes a brief per-shard mutex to read the index (value is inline).
    pub fn get(&self, key: &K) -> Option<[u8; V]> {
        metrics::counter!("armdb.ops", "op" => "get", "tree" => "const_map").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_map.get");
        let index = sync::lock(&self.indexes[self.shard_for(key)]);
        index.get(key).map(|e| e.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_map").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_map.put");
        let shard_id = self.shard_for(key);
        let mut inner = self.durability.lock_shard(shard_id);
        let mut index = sync::lock(&self.indexes[shard_id]);
        let old = self.put_locked(shard_id, &mut *inner, &mut index, key, value)?;
        let needs_sync = inner.should_sync();
        drop(index);
        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_map").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_map.insert");
        let shard_id = self.shard_for(key);
        let mut inner = self.durability.lock_shard(shard_id);
        let mut index = sync::lock(&self.indexes[shard_id]);
        self.insert_locked(shard_id, &mut *inner, &mut index, key, value)?;
        let needs_sync = inner.should_sync();
        drop(index);
        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_map").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_map.delete");
        let shard_id = self.shard_for(key);
        let mut inner = self.durability.lock_shard(shard_id);
        let mut index = sync::lock(&self.indexes[shard_id]);
        let old = self.delete_locked(shard_id, &mut *inner, &mut index, key)?;
        let needs_sync = inner.should_sync();
        drop(index);
        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)
    }

    /// 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.
    pub fn cas(&self, key: &K, expected: &[u8; V], new_value: &[u8; V]) -> DbResult<()> {
        metrics::counter!("armdb.ops", "op" => "cas", "tree" => "const_map").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_map.cas");
        let shard_id = self.shard_for(key);
        let mut inner = self.durability.lock_shard(shard_id);
        let mut index = sync::lock(&self.indexes[shard_id]);

        let entry = index.get(key).ok_or(DbError::KeyNotFound)?;
        if entry.value != *expected {
            return Err(DbError::CasMismatch);
        }

        let old_loc = entry.loc;
        let new_loc = inner.write_update(shard_id as u8, old_loc, key.as_bytes(), new_value)?;

        index.upsert(
            *key,
            MapEntry {
                loc: new_loc,
                value: *new_value,
            },
        );

        let needs_sync = inner.should_sync();
        drop(index);
        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.
    /// 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_map")
            .increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_map.compare_delete");
        let shard_id = self.shard_for(key);
        let mut inner = self.durability.lock_shard(shard_id);
        let mut index = sync::lock(&self.indexes[shard_id]);

        let entry = index.get(key).ok_or(DbError::KeyNotFound)?;
        if entry.value != *expected {
            return Err(DbError::CasMismatch);
        }
        let old_value = entry.value;
        let old_loc = entry.loc;

        inner.write_tombstone(shard_id as u8, old_loc, key.as_bytes())?;
        index.remove(key);

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

        self.hook.on_write(key, Some(&old_value[..]), None);
        Ok(())
    }

    /// Atomically read-modify-write. Returns `Some(new_value)` if key existed, `None` otherwise.
    /// 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_map").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_map.update");
        let shard_id = self.shard_for(key);
        let mut inner = self.durability.lock_shard(shard_id);
        let mut index = sync::lock(&self.indexes[shard_id]);

        let (old_value, new_value) =
            match self.update_locked(shard_id, &mut *inner, &mut index, key, f)? {
                Some(pair) => pair,
                None => return Ok(None),
            };

        let needs_sync = inner.should_sync();
        drop(index);
        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 {
        let index = sync::lock(&self.indexes[self.shard_for(key)]);
        index.contains_key(key)
    }

    /// Total number of entries across all shards.
    pub fn len(&self) -> usize {
        self.indexes.iter().map(|m| sync::lock(m).len()).sum()
    }

    pub fn is_empty(&self) -> bool {
        self.indexes.iter().all(|m| sync::lock(m).is_empty())
    }

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

    /// 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 ConstMapShard<'_, 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 index = sync::lock(&self.indexes[shard_id]);
        let mut shard = ConstMapShard {
            tree: self,
            inner,
            index,
            shard_id,
            events: Vec::new(),
        };
        let result = f(&mut shard);
        let ConstMapShard {
            inner,
            index,
            events,
            ..
        } = shard;
        let needs_sync = inner.should_sync();
        drop(index);
        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[..]),
                );
            }
        }
        result
    }

    fn put_locked(
        &self,
        shard_id: usize,
        inner: &mut D::Inner,
        index: &mut ShardIndex<K, MapEntry<V, D::Loc>>,
        key: &K,
        value: &[u8; V],
    ) -> DbResult<Option<[u8; V]>> {
        // Fast path: key exists — write_update (dead-bytes tracking / in-place overwrite)
        if let Some(existing) = index.get(key) {
            let old_value = existing.value;
            let old_loc = existing.loc;
            let new_loc = inner.write_update(shard_id as u8, old_loc, key.as_bytes(), value)?;
            index.upsert(
                *key,
                MapEntry {
                    loc: new_loc,
                    value: *value,
                },
            );
            return Ok(Some(old_value));
        }

        // Slow path: new key — write_new
        let loc = inner.write_new(shard_id as u8, key.as_bytes(), value)?;
        index.upsert(*key, MapEntry { loc, value: *value });
        Ok(None)
    }

    fn update_locked(
        &self,
        shard_id: usize,
        inner: &mut D::Inner,
        index: &mut ShardIndex<K, MapEntry<V, D::Loc>>,
        key: &K,
        f: impl FnOnce(&[u8; V]) -> [u8; V],
    ) -> DbResult<Option<([u8; V], [u8; V])>> {
        let entry = match index.get(key) {
            Some(e) => e,
            None => return Ok(None),
        };
        let old_value = entry.value;
        let old_loc = entry.loc;
        let new_value = f(&old_value);
        let new_loc = inner.write_update(shard_id as u8, old_loc, key.as_bytes(), &new_value)?;
        index.upsert(
            *key,
            MapEntry {
                loc: new_loc,
                value: new_value,
            },
        );
        Ok(Some((old_value, new_value)))
    }

    fn insert_locked(
        &self,
        shard_id: usize,
        inner: &mut D::Inner,
        index: &mut ShardIndex<K, MapEntry<V, D::Loc>>,
        key: &K,
        value: &[u8; V],
    ) -> DbResult<()> {
        if index.contains_key(key) {
            return Err(DbError::KeyExists);
        }

        let loc = inner.write_new(shard_id as u8, key.as_bytes(), value)?;
        index.upsert(*key, MapEntry { loc, value: *value });
        Ok(())
    }

    fn delete_locked(
        &self,
        shard_id: usize,
        inner: &mut D::Inner,
        index: &mut ShardIndex<K, MapEntry<V, D::Loc>>,
        key: &K,
    ) -> DbResult<Option<[u8; V]>> {
        let old = match index.get(key) {
            Some(old) => old,
            None => return Ok(None),
        };

        inner.write_tombstone(shard_id as u8, old.loc, key.as_bytes())?;
        let value = old.value;
        index.remove(key);
        Ok(Some(value))
    }

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

    // -- Ordered iteration ----------------------------------------------------

    /// Ordered scan view. `None` unless the collection was opened with
    /// `iterable(true)`. Direction follows `Config::reversed` (default DESC).
    ///
    /// **Occasional admin/maintenance API, not a hot path.** Use it to browse a
    /// collection or bulk-clean stale rows (`retain`); for scan-heavy access reach
    /// for the `*Tree` sibling — a Map's strength is O(1) point lookup, and the
    /// companion key index costs extra memory plus O(log N) per write. Unlike a
    /// `Tree` scan it is **not a point-in-time snapshot**: the k-way merge drops
    /// each shard lock between batches, so concurrent writes may be only partially
    /// reflected (run on a quiet DB if you need a coherent view).
    pub fn iter_view(&self) -> Option<ConstMapIterView<'_, K, V, H, D>> {
        self.iterable.then(|| ConstMapIterView {
            inner: MapIterView::new(self, dir_from(self.reversed())),
        })
    }

    // -- Read-only scan -------------------------------------------------------

    /// Read-only pass over all live entries. Order is unspecified; consistency
    /// is per-shard (snapshot under the shard index lock). Intended for schema
    /// validation / maintenance on a quiet database.
    ///
    /// Memory: each shard is snapshotted as a `Vec<(K, [u8; V])>` before the
    /// callback runs, so peak usage is one full shard of keys+values. For
    /// large `V` on big collections prefer trees (streaming `iter()`).
    pub fn for_each(&self, mut f: impl FnMut(&K, &[u8; V])) {
        for i in 0..self.durability.shard_count() {
            let entries: Vec<(K, [u8; V])> = {
                let index = sync::lock(&self.indexes[i]);
                index.iter().map(|(k, e)| (*k, e.value)).collect()
            };
            for (key, value) in entries {
                f(&key, &value);
            }
        }
    }

    // -- Migration ------------------------------------------------------------

    /// Iterate all entries and optionally mutate them.
    ///
    /// Must be called at startup before concurrent writers are active.
    /// Concurrent writes during migration are unsupported and may cause
    /// stale reads.
    ///
    /// Fires `on_init` for Keep/Update (if `NEEDS_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 mut count = 0;
        for i in 0..self.durability.shard_count() {
            let entries: Vec<(K, [u8; V])> = {
                let index = sync::lock(&self.indexes[i]);
                index.iter().map(|(k, e)| (*k, e.value)).collect()
            };
            for (key, value) in entries {
                match f(&key, &value) {
                    MigrateAction::Keep => {
                        if H::NEEDS_INIT {
                            self.hook.on_init(&key, &value[..]);
                        }
                    }
                    MigrateAction::Update(new_value) => {
                        self.put_no_hook(&key, &new_value)?;
                        if H::NEEDS_INIT {
                            self.hook.on_init(&key, &new_value[..]);
                        }
                        count += 1;
                    }
                    MigrateAction::Delete => {
                        self.delete_no_hook(&key)?;
                        count += 1;
                    }
                }
            }
        }

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

    /// Replay `on_init` for every live entry. Used when no migration runs.
    ///
    /// Lock order: snapshot entries under per-shard index lock, release the
    /// lock, then call `on_init` outside the lock. This allows hooks to
    /// re-enter the same collection and shard without deadlocking.
    pub(crate) fn replay_init(&self) {
        if !H::NEEDS_INIT {
            return;
        }
        for shard in &self.indexes {
            let entries: Vec<(K, [u8; V])> = {
                let index = sync::lock(shard);
                index.iter().map(|(k, e)| (*k, e.value)).collect()
            };
            for (key, value) in entries {
                self.hook.on_init(&key, &value[..]);
            }
        }
    }
}

impl<K: Key + Send + Sync + Hash + Eq, const V: usize, H: WriteHook<K>, D: Durability> ShardSource
    for ConstMap<K, V, H, D>
{
    type Key = K;
    type Val = [u8; V];

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

    fn next_batch(
        &self,
        shard: usize,
        after: Option<&K>,
        dir: Dir,
        n: usize,
        range_start: Bound<crate::map_index::OrdKey<K>>,
        range_end: Bound<crate::map_index::OrdKey<K>>,
    ) -> Vec<(K, [u8; V])> {
        let index = sync::lock(&self.indexes[shard]);
        let Some(ki) = index.key_index() else {
            return Vec::new();
        };
        let keys = key_index_batch(ki, after, dir, n, range_start, range_end);
        let mut out = Vec::with_capacity(keys.len());
        for k in keys {
            if let Some(e) = index.get(&k) {
                out.push((k, e.value));
            }
        }
        out
    }

    fn next_keys_batch(
        &self,
        shard: usize,
        after: Option<&K>,
        dir: Dir,
        n: usize,
        range_start: Bound<crate::map_index::OrdKey<K>>,
        range_end: Bound<crate::map_index::OrdKey<K>>,
    ) -> Vec<K> {
        let index = sync::lock(&self.indexes[shard]);
        let Some(ki) = index.key_index() else {
            return Vec::new();
        };
        key_index_batch(ki, after, dir, n, range_start, range_end)
    }
}

/// View over an iterable [`ConstMap`]; exposes ordered iteration over
/// `(K, [u8; V])` entries and cheap keys-only iteration. Obtained via
/// [`ConstMap::iter_view`].
pub struct ConstMapIterView<'a, K, const V: usize, H = NoHook, D = Bitcask>
where
    K: Key + Send + Sync + Hash + Eq,
    H: WriteHook<K>,
    D: Durability,
{
    inner: MapIterView<'a, ConstMap<K, V, H, D>>,
}

impl<K, const V: usize, H, D> ConstMapIterView<'_, K, V, H, D>
where
    K: Key + Send + Sync + Hash + Eq,
    H: WriteHook<K>,
    D: Durability,
{
    /// Full ordered scan in the collection's configured direction.
    pub fn iter(&self) -> impl Iterator<Item = (K, [u8; V])> + '_ {
        self.inner.iter()
    }

    /// Iterate entries in `[start, end)` — start inclusive, end exclusive.
    pub fn range(&self, start: &K, end: &K) -> impl Iterator<Item = (K, [u8; V])> + '_ {
        self.inner.range(start, end)
    }

    /// Iterate entries in range defined by `start` and `end` bounds.
    pub fn range_bounds(
        &self,
        start: Bound<&K>,
        end: Bound<&K>,
    ) -> impl Iterator<Item = (K, [u8; V])> + '_ {
        self.inner.range_bounds(start, end)
    }

    /// Iterate entries whose keys start with `prefix`.
    pub fn prefix_iter(&self, prefix: &[u8]) -> impl Iterator<Item = (K, [u8; V])> + '_ {
        self.inner.prefix_iter(prefix)
    }

    /// Keyset pagination in the collection's configured direction.
    pub fn paginate(&self, after: Option<&K>, limit: usize) -> Page<K, [u8; V]> {
        self.inner.paginate(after, limit)
    }

    /// Full ordered **keys-only** scan — no value materialized.
    pub fn keys(&self) -> impl Iterator<Item = K> + '_ {
        self.inner.keys()
    }

    /// Keys-only iteration over `[start, end)`.
    pub fn keys_range(&self, start: &K, end: &K) -> impl Iterator<Item = K> + '_ {
        self.inner.keys_range(start, end)
    }

    /// Keys-only iteration over a bounded range.
    pub fn keys_range_bounds(
        &self,
        start: Bound<&K>,
        end: Bound<&K>,
    ) -> impl Iterator<Item = K> + '_ {
        self.inner.keys_range_bounds(start, end)
    }

    /// Keys-only iteration over keys starting with `prefix`.
    pub fn keys_prefix(&self, prefix: &[u8]) -> impl Iterator<Item = K> + '_ {
        self.inner.keys_prefix(prefix)
    }

    /// Keys-only pagination — cheap, materializes no values.
    pub fn keys_paginate(&self, after: Option<&K>, limit: usize) -> KeyPage<K> {
        self.inner.keys_paginate(after, limit)
    }

    /// Keep entries where `f` returns `true`; delete the rest via
    /// [`ConstMap::compare_delete`] using the value observed during iteration.
    ///
    /// If the value changed between iteration and delete (`CasMismatch`) or the
    /// key was removed concurrently (`KeyNotFound`), the delete is skipped (R5).
    /// Not atomic across the collection; a genuine storage error aborts the scan
    /// and is returned.
    pub fn retain(&self, mut f: impl FnMut(&K, &[u8; V]) -> bool) -> DbResult<()> {
        for (key, value) in self.inner.iter() {
            if f(&key, &value) {
                continue;
            }
            match self.inner.src().compare_delete(&key, &value) {
                Ok(()) => {}
                Err(DbError::CasMismatch) | Err(DbError::KeyNotFound) => {
                    tracing::debug!("retain: compare_delete skipped (concurrent change)");
                }
                Err(e) => return Err(e),
            }
        }
        Ok(())
    }
}

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

#[cfg(feature = "replication")]
impl<K: Key + Send + Sync + Hash + Eq, const V: usize, H: WriteHook<K>>
    crate::replication::ReplicationTarget for ConstMap<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::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 old = sync::lock(&self.indexes[self.shard_for(&key)]).remove(&key);
            match old {
                Some(old_entry) => Ok(ApplyOutcome::TombstoneRemoved(old_entry.loc)),
                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 old = sync::lock(&self.indexes[self.shard_for(&key)])
                .upsert(key, MapEntry { loc: disk, value });
            match old {
                Some(old_entry) => Ok(ApplyOutcome::Replaced(old_entry.loc)),
                None => Ok(ApplyOutcome::Inserted),
            }
        }
    }

    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;

        let value_len = header.value_len as usize;
        if raw_after_header.len() < size_of::<K>() + value_len {
            return Ok(ApplyOutcome::NotMatched);
        }
        let key = &raw_after_header[..size_of::<K>()];
        let value = &raw_after_header[size_of::<K>()..size_of::<K>() + value_len];
        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>()
    }
}

// ==========================================================================
// ConstMapShard — 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 [`ConstMap::atomic`]. The shard + index locks are held for the
/// lifetime of this struct — keep the closure short.
pub struct ConstMapShard<
    'a,
    K: Key + Send + Sync + Hash + Eq,
    const V: usize,
    H: WriteHook<K> = NoHook,
    D: Durability = Bitcask,
> {
    tree: &'a ConstMap<K, V, H, D>,
    inner: MutexGuard<'a, D::Inner>,
    index: MutexGuard<'a, ShardIndex<K, MapEntry<V, D::Loc>>>,
    shard_id: usize,
    events: Vec<ConstShardEvent<K, V>>,
}

impl<K: Key + Send + Sync + Hash + Eq, const V: usize, H: WriteHook<K>, D: Durability>
    ConstMapShard<'_, 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, &mut self.index, 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, &mut self.index, 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, &mut self.index, key)?;
        if H::NEEDS_WRITE
            && let Some(ref old_val) = old
        {
            self.events.push((*key, Some(*old_val), None));
        }
        Ok(old)
    }

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

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

    fn update_inner(
        &mut self,
        key: &K,
        f: impl FnOnce(&[u8; V]) -> [u8; V],
        return_old: bool,
    ) -> DbResult<Option<[u8; V]>> {
        self.check_shard(key)?;
        let (old_value, new_value) = match self.tree.update_locked(
            self.shard_id,
            &mut *self.inner,
            &mut self.index,
            key,
            f,
        )? {
            Some(pair) => pair,
            None => return Ok(None),
        };
        if H::NEEDS_WRITE {
            self.events.push((*key, Some(old_value), Some(new_value)));
        }
        Ok(Some(if return_old { old_value } else { new_value }))
    }

    pub fn get(&self, key: &K) -> Option<[u8; V]> {
        self.index.get(key).map(|e| e.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.index.contains_key(key)
    }

    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`)
// ==========================================================================

#[cfg(feature = "armour")]
type ConstMapGuard<'a, K, const V: usize, D> = (
    usize,
    MutexGuard<'a, <D as Durability>::Inner>,
    MutexGuard<'a, ShardIndex<K, MapEntry<V, <D as Durability>::Loc>>>,
);

/// Multi-shard transaction handle for [`ConstMap`] inside `Db::atomicN`. Unlike
/// the tree handle it stores a `(durability inner, index HashMap)` guard pair per
/// shard and has no epoch guard.
#[cfg(feature = "armour")]
pub struct ConstMapTx<'a, K, const V: usize, H = NoHook, D = Bitcask>
where
    K: Key + Send + Sync + Hash + Eq,
    H: WriteHook<K>,
    D: Durability,
{
    tree: &'a ConstMap<K, V, H, D>,
    shards: Vec<ConstMapGuard<'a, K, V, D>>,
    log: Vec<ConstShardEvent<K, V>>,
}

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

    pub fn try_get(&self, key: &K) -> DbResult<Option<[u8; V]>> {
        let i = self.position(key)?;
        Ok(self.shards[i].2.get(key).map(|e| e.value))
    }

    pub fn try_contains(&self, key: &K) -> DbResult<bool> {
        let i = self.position(key)?;
        Ok(self.shards[i].2.contains_key(key))
    }

    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, index) = &mut self.shards[i];
        let old = self
            .tree
            .put_locked(*sid, &mut **inner, &mut **index, 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, index) = &mut self.shards[i];
        self.tree
            .insert_locked(*sid, &mut **inner, &mut **index, 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, index) = &mut self.shards[i];
        let old = self
            .tree
            .delete_locked(*sid, &mut **inner, &mut **index, 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, const V: usize, H, D> crate::armour::multi_tx::MultiTx for ConstMap<K, V, H, D>
where
    K: Key + Send + Sync + Hash + Eq,
    H: WriteHook<K>,
    D: Durability,
{
    type Key = K;
    type Tx<'a>
        = ConstMapTx<'a, K, V, H, D>
    where
        Self: 'a;

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

    fn begin_tx(&self) -> ConstMapTx<'_, K, V, H, D> {
        ConstMapTx {
            tree: self,
            shards: Vec::new(),
            log: Vec::new(),
        }
    }

    fn lock_shard_into<'a>(&'a self, shard_id: usize, tx: &mut ConstMapTx<'a, K, V, H, D>) {
        // Lock durability inner first, then the per-shard index — SAME order the
        // single-shard atomic() uses, so a concurrent atomic() on this shard cannot
        // invert against us.
        let inner = self.durability.lock_shard(shard_id);
        let index = sync::lock(&self.indexes[shard_id]);
        tx.shards.push((shard_id, inner, index));
    }

    fn release_locks(
        &self,
        tx: &mut ConstMapTx<'_, K, V, H, D>,
    ) -> crate::armour::multi_tx::SyncNeeds {
        let mut needs = crate::armour::multi_tx::SyncNeeds::none();
        for (sid, inner, _) in &tx.shards {
            if inner.should_sync() {
                needs.push(*sid);
            }
        }
        tx.shards.clear(); // drops index + inner guards
        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: ConstMapTx<'_, 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[..]),
                );
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::Config;
    use crate::hook::WriteHook;
    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering, Ordering as AtomicOrdering};
    use std::sync::mpsc;
    use std::time::Duration;
    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_map_hooked(dir: &std::path::Path, hook: RecHook) -> ConstMap<[u8; 8], 4, RecHook> {
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        ConstMap::open_hooked(dir, cfg, hook).expect("open hooked")
    }

    #[test]
    fn iter_view_none_unless_iterable() {
        let dir = tempfile::tempdir().unwrap();
        let m = ConstMap::<[u8; 2], 1>::open(dir.path(), Config::test()).unwrap();
        assert!(m.iter_view().is_none());
    }

    #[test]
    fn iter_desc_by_default_across_shards() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Config::balanced()
            .shard_count(4)
            .hints(true)
            .iterable(true)
            .build();
        let m = ConstMap::<[u8; 2], 1>::open(dir.path(), cfg).unwrap();
        for k in [[1u8, 0], [3, 0], [2, 0], [5, 0], [4, 0]] {
            m.insert(&k, &[7]).unwrap();
        }
        let got: Vec<[u8; 2]> = m.iter_view().unwrap().iter().map(|(k, _)| k).collect();
        assert_eq!(got, [[5, 0], [4, 0], [3, 0], [2, 0], [1, 0]]);
    }

    #[test]
    fn paginate_covers_keyset_once_desc() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Config::balanced()
            .shard_count(3)
            .hints(true)
            .iterable(true)
            .build();
        let m = ConstMap::<[u8; 2], 1>::open(dir.path(), cfg).unwrap();
        for i in 1u8..=10 {
            m.insert(&[i, 0], &[i]).unwrap();
        }
        let view = m.iter_view().unwrap();
        let mut seen = Vec::new();
        let mut after: Option<[u8; 2]> = None;
        loop {
            let page = view.paginate(after.as_ref(), 3);
            if page.items.is_empty() {
                break;
            }
            seen.extend(page.items.iter().map(|(k, _)| *k));
            after = page.next;
            if after.is_none() {
                break;
            }
        }
        let mut sorted = seen.clone();
        sorted.sort();
        sorted.dedup();
        assert_eq!(sorted.len(), 10, "every key exactly once");
        assert_eq!(seen.first().unwrap(), &[10, 0]);
    }

    #[test]
    fn range_and_prefix_bounds() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Config::balanced()
            .shard_count(2)
            .hints(true)
            .iterable(true)
            .reversed(false)
            .build();
        let m = ConstMap::<[u8; 2], 1>::open(dir.path(), cfg).unwrap();
        for i in 1u8..=5 {
            m.insert(&[i, 0], &[i]).unwrap();
        }
        let v = m.iter_view().unwrap();
        let r: Vec<_> = v.range(&[2, 0], &[4, 0]).map(|(k, _)| k).collect();
        assert_eq!(r, [[2, 0], [3, 0]]);
        // value prefix_iter (exercises prefix_to_end_bound on the value path).
        let p: Vec<[u8; 2]> = v.prefix_iter(&[3]).map(|(k, _)| k).collect();
        assert_eq!(p, [[3, 0]]);
    }

    #[test]
    fn degenerate_range_is_empty_not_panic() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Config::balanced()
            .shard_count(2)
            .hints(true)
            .iterable(true)
            .reversed(false)
            .build();
        let m = ConstMap::<[u8; 2], 1>::open(dir.path(), cfg).unwrap();
        for i in 1u8..=5 {
            m.insert(&[i, 0], &[i]).unwrap();
        }
        let v = m.iter_view().unwrap();
        // Excluded == Excluded (empty) and inverted bounds must not panic.
        assert_eq!(
            v.range_bounds(Bound::Excluded(&[3, 0]), Bound::Excluded(&[3, 0]))
                .count(),
            0
        );
        assert_eq!(
            v.keys_range_bounds(Bound::Included(&[4, 0]), Bound::Included(&[2, 0]))
                .count(),
            0
        );
    }

    #[test]
    fn keys_only_matches_iter_keys_and_is_ordered() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Config::balanced()
            .shard_count(3)
            .hints(true)
            .iterable(true)
            .build();
        let m = ConstMap::<[u8; 2], 1>::open(dir.path(), cfg).unwrap();
        for i in 1u8..=6 {
            m.insert(&[i, 0], &[i]).unwrap();
        }
        let v = m.iter_view().unwrap();
        let from_keys: Vec<[u8; 2]> = v.keys().collect();
        let from_iter: Vec<[u8; 2]> = v.iter().map(|(k, _)| k).collect();
        assert_eq!(from_keys, from_iter, "keys() order must match iter() keys");
        assert_eq!(
            from_keys,
            [[6, 0], [5, 0], [4, 0], [3, 0], [2, 0], [1, 0]],
            "DESC by default"
        );
    }

    #[test]
    fn keys_range_and_prefix() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Config::balanced()
            .shard_count(2)
            .hints(true)
            .iterable(true)
            .reversed(false)
            .build();
        let m = ConstMap::<[u8; 2], 1>::open(dir.path(), cfg).unwrap();
        for i in 1u8..=5 {
            m.insert(&[i, 0], &[i]).unwrap();
        }
        let v = m.iter_view().unwrap();
        let r: Vec<[u8; 2]> = v.keys_range(&[2, 0], &[4, 0]).collect();
        assert_eq!(r, [[2, 0], [3, 0]]);
        let p: Vec<[u8; 2]> = v.keys_prefix(&[3]).collect();
        assert_eq!(p, [[3, 0]]);
    }

    #[test]
    fn keys_paginate_covers_keyset_once_desc() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Config::balanced()
            .shard_count(3)
            .hints(true)
            .iterable(true)
            .build();
        let m = ConstMap::<[u8; 2], 1>::open(dir.path(), cfg).unwrap();
        for i in 1u8..=10 {
            m.insert(&[i, 0], &[i]).unwrap();
        }
        let v = m.iter_view().unwrap();
        let mut seen = Vec::new();
        let mut after: Option<[u8; 2]> = None;
        loop {
            let page = v.keys_paginate(after.as_ref(), 3);
            if page.keys.is_empty() {
                break;
            }
            seen.extend(page.keys.iter().copied());
            after = page.next;
            if after.is_none() {
                break;
            }
        }
        let mut sorted = seen.clone();
        sorted.sort();
        sorted.dedup();
        assert_eq!(sorted.len(), 10, "every key exactly once");
        assert_eq!(
            seen.first().unwrap(),
            &[10, 0],
            "DESC starts at the max key"
        );
    }

    /// R2: the Fixed replication apply helpers (`upsert_replicated` /
    /// `remove_key_if_slot_matches`) must keep the iterable companion in sync, so a
    /// follower opened `iterable(true)` reflects the replicated keyset.
    #[cfg(feature = "replication")]
    #[test]
    fn replication_apply_maintains_iterable_companion_fixed() {
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = crate::FixedConfig::test();
        cfg.iterable = true;
        let m = crate::FixedMap::<[u8; 2], 1>::open(dir.path(), cfg).unwrap();

        // Simulate a follower applying replicated upserts (slots bypass durability).
        m.upsert_replicated(&[1, 0], [10], 0);
        m.upsert_replicated(&[2, 0], [20], 1);
        m.upsert_replicated(&[3, 0], [30], 2);

        let keys: Vec<[u8; 2]> = m.iter_view().unwrap().keys().collect();
        assert_eq!(
            keys,
            [[3, 0], [2, 0], [1, 0]],
            "all replicated keys present (Fixed is DESC)"
        );

        // Apply a replicated removal — companion must drop the key.
        let slot = m.get_slot_id(&[2, 0]).unwrap();
        assert!(m.remove_key_if_slot_matches(&[2, 0], slot));
        let keys: Vec<[u8; 2]> = m.iter_view().unwrap().keys().collect();
        assert_eq!(
            keys,
            [[3, 0], [1, 0]],
            "replicated tombstone removed from companion"
        );
    }

    #[test]
    fn iter_reflects_delete() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Config::balanced()
            .shard_count(2)
            .hints(true)
            .iterable(true)
            .build();
        let m = ConstMap::<[u8; 2], 1>::open(dir.path(), cfg).unwrap();
        m.insert(&[1, 0], &[1]).unwrap();
        m.insert(&[2, 0], &[2]).unwrap();
        m.delete(&[1, 0]).unwrap();
        let got: Vec<[u8; 2]> = m.iter_view().unwrap().iter().map(|(k, _)| k).collect();
        assert_eq!(got, [[2, 0]]);
    }

    #[test]
    fn const_map_atomic_fires_hooks_in_order() {
        let dir = tempfile::tempdir().unwrap();
        let map = open_map_hooked(dir.path(), RecHook::default());
        let k = 7u64.to_be_bytes();
        map.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!(map.hook.writes.load(AtomicOrdering::Relaxed), 3);
        let seq = crate::sync::lock(&map.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));
    }

    #[test]
    fn const_map_atomic_fires_for_applied_on_err() {
        let dir = tempfile::tempdir().unwrap();
        let map = open_map_hooked(dir.path(), RecHook::default());
        let k = 1u64.to_be_bytes();
        let r: DbResult<()> = map.atomic(&k, |s| {
            s.put(&k, &[9, 9, 9, 9])?;
            Err(DbError::KeyNotFound)
        });
        assert!(r.is_err());
        assert_eq!(map.hook.writes.load(AtomicOrdering::Relaxed), 1);
    }

    #[test]
    fn const_map_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 = crate::FixedConfig {
            shard_count: 1,
            grow_step: 64,
            sync_batch_size: 1,
            ..crate::FixedConfig::test()
        };
        let map = crate::FixedMap::<[u8; 8], 4, RecHook>::open_with_hook(
            dir.path(),
            fixed_cfg,
            RecHook::default(),
        )
        .expect("open fixed hooked");

        let k = 7u64.to_be_bytes();
        map.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!(map.hook.writes.load(AtomicOrdering::Relaxed), 3);
        let seq = crate::sync::lock(&map.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));
    }

    #[allow(clippy::type_complexity)]
    struct ReplayInitReenterHook<const V: usize> {
        target: sync::Mutex<Option<std::sync::Arc<ConstMap<[u8; 8], V, Self>>>>,
        entered: AtomicBool,
    }

    impl<const V: usize> WriteHook<[u8; 8]> for ReplayInitReenterHook<V> {
        const NEEDS_OLD_VALUE: bool = false;
        const NEEDS_INIT: bool = true;
        const NEEDS_WRITE: bool = false;

        fn on_write(&self, _key: &[u8; 8], _old: Option<&[u8]>, _new: Option<&[u8]>) {}

        fn on_init(&self, key: &[u8; 8], _value: &[u8]) {
            self.entered.store(true, Ordering::Release);
            let map = sync::lock(&self.target)
                .as_ref()
                .expect("hook target must be installed")
                .clone();
            let _ = map.get(key);
        }
    }

    #[test]
    fn hook_lifecycle_replay_init_reentry_const_map() {
        let dir = tempdir().unwrap();
        let hook = ReplayInitReenterHook::<8> {
            target: sync::Mutex::new(None),
            entered: AtomicBool::new(false),
        };
        let map = std::sync::Arc::new(
            ConstMap::<[u8; 8], 8, ReplayInitReenterHook<8>>::open_hooked(
                dir.path(),
                Config::test(),
                hook,
            )
            .unwrap(),
        );
        *sync::lock(&map.hook.target) = Some(std::sync::Arc::clone(&map));

        let key = 1u64.to_be_bytes();
        map.put(&key, &[0u8; 8]).unwrap();

        let (tx, rx) = mpsc::sync_channel(1);
        let map2 = std::sync::Arc::clone(&map);
        std::thread::spawn(move || {
            map2.replay_init();
            let _ = tx.send(());
        });
        rx.recv_timeout(Duration::from_secs(2))
            .expect("replay_init deadlocked — on_init called with index lock held");
        assert!(map.hook.entered.load(Ordering::Acquire));
    }

    #[test]
    fn retain_deletes_and_persists() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Config::balanced()
            .shard_count(2)
            .hints(true)
            .iterable(true)
            .build();
        {
            let m = ConstMap::<[u8; 2], 1>::open(dir.path(), cfg.clone()).unwrap();
            for i in 1u8..=6 {
                m.insert(&[i, 0], &[i]).unwrap();
            }
            m.iter_view()
                .unwrap()
                .retain(|_k, v| v[0] % 2 == 0)
                .unwrap();
            let got: Vec<_> = m.iter_view().unwrap().iter().map(|(k, _)| k[0]).collect();
            let mut g = got.clone();
            g.sort();
            assert_eq!(g, [2, 4, 6]);
            m.close().unwrap();
        }
        let m = ConstMap::<[u8; 2], 1>::open(dir.path(), cfg).unwrap();
        let got: Vec<_> = m.iter_view().unwrap().iter().map(|(k, _)| k[0]).collect();
        let mut g = got.clone();
        g.sort();
        assert_eq!(g, [2, 4, 6]);
    }

    #[test]
    fn retain_compare_delete_skips_changed_value() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Config::balanced()
            .shard_count(2)
            .hints(true)
            .iterable(true)
            .build();
        let m = std::sync::Arc::new(ConstMap::<[u8; 2], 1>::open(dir.path(), cfg).unwrap());
        m.insert(&[1, 0], &[1]).unwrap();

        let m2 = std::sync::Arc::clone(&m);
        m.iter_view()
            .unwrap()
            .retain(|k, _v| {
                if k == &[1, 0] {
                    m2.put(k, &[9]).unwrap();
                    return false;
                }
                true
            })
            .unwrap();

        assert_eq!(m.get(&[1, 0]), Some([9]));
    }

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

        let (upd, fetched, missing) = map
            .atomic(&key, |shard| {
                let upd = shard.update(&key, |old| {
                    let mut v = *old;
                    v[0] = v[0].wrapping_add(1);
                    v
                })?;
                let fetched = shard.fetch_update(&key, |old| {
                    let mut v = *old;
                    v[0] = v[0].wrapping_add(1);
                    v
                })?;
                let missing = shard.update(&2u64.to_be_bytes(), |_| [0u8; 4])?;
                Ok((upd, fetched, missing))
            })
            .expect("atomic");

        assert_eq!(upd, Some([11, 10, 10, 10]));
        assert_eq!(fetched, Some([11, 10, 10, 10]));
        assert_eq!(missing, None);
        assert_eq!(map.get(&key), Some([12, 10, 10, 10]));
    }

    #[test]
    fn const_map_atomic_update_fires_hook_with_old_new() {
        let dir = tempdir().unwrap();
        let map = open_map_hooked(dir.path(), RecHook::default());
        let key = 1u64.to_be_bytes();
        map.put(&key, &[10u8; 4]).unwrap();
        // put fired hook once
        assert_eq!(map.hook.writes.load(AtomicOrdering::Relaxed), 1);

        map.atomic(&key, |shard| {
            shard.update(&key, |old| {
                let mut v = *old;
                v[0] = v[0].wrapping_add(5);
                v
            })
        })
        .unwrap();

        // one more write event from update
        assert_eq!(map.hook.writes.load(AtomicOrdering::Relaxed), 2);
        assert_eq!(map.get(&key), Some([15, 10, 10, 10]));
    }

    #[test]
    fn compare_delete_updates_companion() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Config::balanced()
            .shard_count(2)
            .hints(true)
            .iterable(true)
            .build();
        let m = ConstMap::<[u8; 2], 1>::open(dir.path(), cfg).unwrap();
        m.insert(&[1, 0], &[9]).unwrap();
        m.compare_delete(&[1, 0], &[9]).unwrap();
        assert_eq!(m.iter_view().unwrap().iter().count(), 0);
    }

    #[test]
    fn fixed_map_iteration_direction() {
        // default reversed=true -> DESC
        let dir = tempdir().unwrap();
        let cfg = crate::FixedConfig {
            iterable: true,
            ..crate::FixedConfig::test()
        };
        let m = crate::FixedMap::<[u8; 8], 8>::open(dir.path(), cfg).unwrap();
        for i in 1u64..=3 {
            m.put(&i.to_be_bytes(), &i.to_be_bytes()).unwrap();
        }
        let desc: Vec<u64> = m
            .iter_view()
            .unwrap()
            .keys()
            .map(u64::from_be_bytes)
            .collect();
        assert_eq!(desc, [3, 2, 1], "FixedMap default DESC");

        // reversed=false -> ASC
        let dir2 = tempdir().unwrap();
        let cfg2 = crate::FixedConfig {
            iterable: true,
            reversed: false,
            ..crate::FixedConfig::test()
        };
        let m2 = crate::FixedMap::<[u8; 8], 8>::open(dir2.path(), cfg2).unwrap();
        for i in 1u64..=3 {
            m2.put(&i.to_be_bytes(), &i.to_be_bytes()).unwrap();
        }
        let asc: Vec<u64> = m2
            .iter_view()
            .unwrap()
            .keys()
            .map(u64::from_be_bytes)
            .collect();
        assert_eq!(asc, [1, 2, 3], "FixedMap reversed=false ASC");
    }

    #[test]
    fn compare_delete_fixed_store_match_mismatch_absent() {
        let dir = tempdir().unwrap();
        let map =
            crate::FixedMap::<[u8; 8], 8>::open(dir.path(), crate::FixedConfig::test()).unwrap();

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

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

    #[test]
    fn compare_delete_bitcask_match_mismatch_absent() {
        let dir = tempdir().unwrap();
        let map = ConstMap::<[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();
        map.put(&key, &val).unwrap();

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

    #[test]
    fn fixed_map_reopen_iterable_recovers_all() {
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = crate::FixedConfig::test(); // shard_count 3
        cfg.iterable = true;

        {
            let m = crate::FixedMap::<[u8; 2], 1>::open(dir.path(), cfg.clone()).unwrap();
            m.insert(&[1, 0], &[10]).unwrap();
            m.insert(&[2, 0], &[20]).unwrap();
            m.insert(&[3, 0], &[30]).unwrap();
            m.close().unwrap(); // clean shutdown -> sidecar path
        }

        let m = crate::FixedMap::<[u8; 2], 1>::open(dir.path(), cfg).unwrap();
        // iterable companion rebuilt after recovery (lock hoisting preserves order).
        let keys: Vec<[u8; 2]> = m.iter_view().unwrap().keys().collect();
        assert_eq!(keys, [[3, 0], [2, 0], [1, 0]], "DESC, all keys recovered");
        assert_eq!(m.get(&[2, 0]).unwrap(), [20]);
    }

    #[test]
    fn fixed_map_reopen_dirty_recovers_all() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = crate::FixedConfig::test(); // shard_count 3

        {
            let m = crate::FixedMap::<[u8; 8], 1>::open(dir.path(), cfg.clone()).unwrap();
            for i in 0..50u64 {
                m.insert(&i.to_be_bytes(), &[(i & 0xff) as u8]).unwrap();
            }
            // drop without close -> dirty (full-scan) recovery on reopen.
        }

        let m = crate::FixedMap::<[u8; 8], 1>::open(dir.path(), cfg).unwrap();
        for i in 0..50u64 {
            assert_eq!(
                m.get(&i.to_be_bytes()).unwrap(),
                [(i & 0xff) as u8],
                "key {i}"
            );
        }
    }
}