armdb 0.13.0

use std::hash::Hash;
use std::mem::size_of;
use std::ops::{Bound, ControlFlow};
use std::sync::{Arc, OnceLock, Weak};

use crate::Key;

use crate::compaction::{CompactionIndex, compact_shard};
use crate::config::Config;
use crate::const_tree::ConstNew;
use crate::ctx::{Ctx, Family, Held, Owned, ShardGuards};
use crate::disk_loc::DiskLoc;
use crate::durability::{Bitcask, Durability, DurabilityInner};
use crate::engine::Engine;
use crate::error::{DbError, DbResult};
use crate::hook::{NoHook, WriteHook};
use crate::key::Location;
use crate::map_index::{
    Dir, KeyPage, MapIterView, Page, ShardIndex, ShardSource, dir_from, key_index_batch,
};
use crate::recovery::recover_const_map;
use crate::sync::{self, Mutex, MutexGuard};
use crate::ttl::{SweepOutcome, TtlSlots};
use crate::watch::{WatchEventInner, WatchWakeBatch};

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

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,
    pub(crate) hook: H,
    iterable: bool,
    reversed: bool,
    tx_self: OnceLock<Weak<Self>>,
    ttl: TtlSlots<K>,
    /// `counters_dirty` marker (T9 Task 7) for the `ZeroMap` this
    /// engine backs, bound once at managed open via
    /// `ManagedOpen::bind_counter_dirty`. Lives here (not on the hook) so a
    /// user-authored wrapper hook that forgets to forward some hook method can
    /// never silently disconnect it — see `counter_dirty.rs`.
    pub(crate) counter_dirty: OnceLock<crate::counter_dirty::CounterDirty>,
}

impl<K: Key + Send + Sync + Hash + Eq, const V: usize, H: WriteHook<K>, D: Durability>
    ConstMap<K, V, H, D>
{
    pub(crate) fn watch_channel(&self) -> Option<&std::sync::Arc<crate::watch::WatchChannel<K>>> {
        self.hook.watch_channel()
    }
}

// ==========================================================================
// 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.
    ///
    #[doc = include_str!("../docs/exclusive-access.md")]
    pub fn open(path: impl AsRef<std::path::Path>, config: Config) -> DbResult<Self> {
        Self::open_hooked(path, config, NoHook)
    }
}

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

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

    pub(crate) 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,
            tx_self: OnceLock::new(),
            ttl: TtlSlots::new(),
            counter_dirty: OnceLock::new(),
        };

        // 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.shard_ciphers(),
        )?;
        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);
        }
        for (shard_idx, file_seqs) in outcome.shard_file_seqs {
            map.durability.engine.shards()[shard_idx].install_recovery_file_seqs(file_seqs);
        }
        // The transaction replay rule compares a reserved GSN against this
        // watermark, so it must exist in every build — not only with
        // `replication` on. `ConstMap` backs `ZeroMap`, which is a transaction
        // participant; leaving the watermark at 0 makes every logged operation
        // look un-applied and replay resurrects superseded values.
        for (shard_idx, max_gsn) in &outcome.shard_max_gsns {
            map.durability.engine.shards()[*shard_idx].set_durable_recovered_gsn(*max_gsn);
        }
        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)
    }

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

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

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

    /// Durable flush: write buffers + fsync active files.
    pub(crate) fn durable_flush(&self) -> DbResult<()> {
        self.flush_buffers()?;
        self.durability.flush()
    }

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

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

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

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

    /// Trigger a 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.
    ///
    #[doc = include_str!("../docs/exclusive-access.md")]
    pub fn open(path: impl AsRef<std::path::Path>, config: FixedConfig) -> DbResult<Self> {
        Self::open_fixed_inner(path, config, NoHook)
    }
}

impl<K: Key + 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.
    ///
    #[doc = include_str!("../docs/exclusive-access.md")]
    pub fn open_with_hook(
        path: impl AsRef<std::path::Path>,
        config: FixedConfig,
        hook: H,
    ) -> DbResult<Self> {
        #[cfg(feature = "armour")]
        if H::EMITS {
            return Err(DbError::ChangeStreamRequiresManagedOpen);
        }
        Self::open_fixed_inner(path, config, hook)
    }

    pub(crate) 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| -> Option<u32> {
                let key = K::from_bytes(key_bytes);
                let mut value = [0u8; V];
                value.copy_from_slice(value_bytes);
                // Last-scanned-wins: `upsert` returns the entry this key already
                // held from an earlier slot — a post-crash duplicate (F1/F2).
                // Report that slot so the scan loop frees it, keeping exactly one
                // OCCUPIED slot per key. This policy matches ConstTree's recovery.
                guard
                    .upsert(
                        key,
                        MapEntry {
                            loc: slot_id,
                            value,
                        },
                    )
                    .map(|prev| prev.loc)
            }
        })?;

        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,
            tx_self: OnceLock::new(),
            ttl: TtlSlots::new(),
            counter_dirty: OnceLock::new(),
        })
    }

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

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

    /// Resolve one planned operation to a `(slot, target_meta)` pair, under the
    /// shard lock the caller already holds.
    pub(crate) fn tx_reserve_fixed(
        &self,
        shard: &mut ConstMapShard<'_, K, V, H, Fixed>,
        key: &[u8],
        delete: bool,
    ) -> DbResult<(u32, u32)> {
        let k = K::from_bytes(key);
        let existing = shard.held.locked.1.get(&k).map(|entry| entry.loc);
        if delete && existing.is_none() {
            return Ok((crate::fixed::slot::NO_SLOT, 0));
        }
        shard.held.locked.0.reserve_slot(existing, delete)
    }

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

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

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

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

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

    /// Current index slot for `key`, if any — see
    /// [`ConstTree::tx_fixed_index_slot`](crate::ConstTree::tx_fixed_index_slot).
    pub(crate) fn tx_fixed_index_slot(
        shard: &ConstMapShard<'_, K, V, H, Fixed>,
        key: &[u8],
    ) -> Option<u32> {
        let k = K::from_bytes(key);
        shard.held.locked.1.get(&k).map(|entry| entry.loc)
    }

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

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

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

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

#[cfg(feature = "replication")]
impl<K: Key + 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,
            },
        );
    }
}

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

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

impl<K: Key + Send + Sync + Hash + Eq, const V: usize, H: WriteHook<K>, D: Durability> Family
    for ConstMap<K, V, H, D>
{
    type Locked<'a>
        = (
        MutexGuard<'a, D::Inner>,
        MutexGuard<'a, ShardIndex<K, MapEntry<V, D::Loc>>>,
    )
    where
        Self: 'a;
    type Event<'v> = ConstMapWriteEvent<'v, K, V>;
    type Owned = ConstMapOwnedEvent<K, V>;

    #[inline]
    fn lock_shard(&self, sid: usize) -> Self::Locked<'_> {
        let inner = self.durability.lock_shard(sid);
        let index = sync::lock(&self.indexes[sid]);
        (inner, index)
    }

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

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

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

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

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

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

    const WATCHES: bool = H::WATCHES;

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

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

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

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

impl<K: Key + Send + Sync + Hash + Eq, const V: usize, H: WriteHook<K>, D: Durability>
    ConstMap<K, V, H, D>
{
    /// Build per-shard TTL indexes by walking live in-memory entries.
    #[doc(hidden)]
    pub fn build_ttl_index(&self) -> DbResult<()> {
        if !H::EXPIRES {
            return Ok(());
        }
        let n = self.durability.shard_count();
        let slots = self.ttl.install_and_clear(n);

        for (sid, slot) in slots.iter().enumerate() {
            // Lock order: durability shard, then per-shard index (same as mutations).
            let _shard = self.durability.lock_shard(sid);
            let index = sync::lock(&self.indexes[sid]);
            for (key, entry) in index.iter() {
                let deadline = self.hook.deadline(key, &entry.value[..]);
                sync::lock(slot).record(key, deadline);
            }
        }
        Ok(())
    }

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

    fn record_ttl(&self, shard_id: usize, key: &K, expires_at: Option<u64>) {
        if !H::EXPIRES {
            return;
        }
        self.ttl.record(shard_id, key, expires_at);
    }

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

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

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

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

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

                let action = match locked.1.get(&key) {
                    Some(entry) => {
                        match crate::ttl::catch_deadline(|| {
                            self.hook.deadline(&key, &entry.value[..])
                        }) {
                            Some(Some(dl)) if dl <= now_ms => {
                                ConstMapSweepAction::Delete(key, candidate_dl)
                            }
                            Some(Some(dl)) => ConstMapSweepAction::Reschedule(key, dl),
                            Some(None) => ConstMapSweepAction::Drop,
                            None => ConstMapSweepAction::Retry(key),
                        }
                    }
                    None => ConstMapSweepAction::Drop,
                };

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

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

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

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

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

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

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

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

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

    /// Insert or update a key-value pair. Returns the old value if the key existed.
    pub fn put(&self, key: &K, value: &[u8; V]) -> DbResult<Option<[u8; V]>> {
        metrics::counter!("armdb.ops", "op" => "put", "tree" => "const_map").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_map.put");
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let (inner, index) = locked;
            let old = self.put_locked(shard_id, &mut *inner, index, key, value)?;
            *ev_out = self.defer_borrowed_event(key, old, Some(value));
            Ok(old)
        })
    }

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

    /// Delete a key. Returns the old value if the key existed.
    pub fn delete(&self, key: &K) -> DbResult<Option<[u8; V]>> {
        metrics::counter!("armdb.ops", "op" => "delete", "tree" => "const_map").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("const_map.delete");
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let (inner, index) = locked;
            let old = self.delete_locked(shard_id, &mut *inner, index, key)?;
            let ev = match &old {
                Some(o) => self.defer_write_event(key, Some(*o), None),
                None => None,
            };
            *ev_out = ev;
            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");
        // Outside the shard lock: the new bytes are final regardless of whether
        // the compare succeeds, and the extractor must not run under the lock.
        let expires_at = self.deadline_for(key, new_value);
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let (inner, index) = locked;
            let entry = index.get(key).ok_or(DbError::KeyNotFound)?;
            if entry.value != *expected {
                return Err(DbError::CasMismatch);
            }
            // T10 Task 5: `cas` has its own inline body (does not go through
            // `put_locked`), so it needs its own guard call.
            self.claim_guard_bytes(key, Some(&entry.value[..]), Some(&new_value[..]))?;

            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,
                },
            );
            self.record_ttl(shard_id, key, expires_at);
            // `expected` / `new_value` outlive the closure, so the new half rides
            // borrowed. The old half cannot: an event outlives the closure while a
            // value read from the node under the lock does not, so `Option<[u8; V]>`
            // by value is structural here, not an oversight.
            *ev_out = self.defer_borrowed_event(key, Some(*expected), Some(new_value));
            Ok(())
        })
    }

    /// Compare-and-delete: if the current value == `expected`, delete the key.
    /// Returns `Ok(())` on success, `Err(CasMismatch)` if current != expected,
    /// `Err(KeyNotFound)` if the key doesn't exist.
    /// 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 ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let (inner, index) = locked;
            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;
            // T10 Task 5: `compare_delete` has its own inline body — a pure
            // release ({k} -> ∅), refused outside `db.tx()`.
            self.claim_guard_bytes(key, Some(&old_value[..]), None)?;

            inner.write_tombstone(shard_id as u8, old_loc, key.as_bytes())?;
            index.remove(key);
            self.record_ttl(shard_id, key, None);
            *ev_out = self.defer_write_event(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 ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let (inner, index) = locked;
            let (old_value, new_value) =
                match self.update_locked(shard_id, &mut *inner, index, key, f)? {
                    Some(pair) => pair,
                    None => return Ok(None),
                };
            let result = Some(if return_old { old_value } else { new_value });
            *ev_out = self.defer_write_event(key, Some(old_value), Some(new_value));
            Ok(result)
        })
    }

    /// Check if a key exists.
    pub fn contains(&self, key: &K) -> bool {
        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 ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, _ev_out| {
            let (inner, index) = locked;
            let old = self.put_locked(shard_id, &mut *inner, index, key, value)?;
            Ok(old)
        })
    }

    fn delete_no_hook(&self, key: &K) -> DbResult<Option<[u8; V]>> {
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, _ev_out| {
            let (inner, index) = locked;
            let old = self.delete_locked(shard_id, &mut *inner, index, key)?;
            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 ConstMapShard<'_, K, V, H, D>) -> DbResult<R>,
    ) -> DbResult<R> {
        let _atomic_depth = crate::AtomicDepthGuard::try_enter()?;
        let shard_id = self.shard_for(shard_key);
        let mut shard = self.tx_shard(shard_id);
        let result = f(&mut shard);
        let needs_sync = Self::should_sync(&shard.held.locked);
        let (events, wakes) = Self::tx_release_shard(shard);
        let result =
            crate::projection::finish_atomic_teardown(self, shard_id, result, needs_sync, &events);
        wakes.drain();
        result
    }

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

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

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

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

    /// Collection name for the transaction log. Standalone maps return `""`.
    pub(crate) fn name(&self) -> &'static str {
        ""
    }

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

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

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

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

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

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

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

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

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

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

    /// Release a shard built by [`tx_shard`], returning its buffered events.
    pub(crate) fn tx_release_shard(
        shard: ConstMapShard<'_, K, V, H, D>,
    ) -> (Vec<ConstMapOwnedEvent<K, V>>, crate::watch::WatchWakeBatch) {
        let ConstMapShard { held, .. } = shard;
        let Held {
            locked,
            events,
            watch_wakes,
            ..
        } = held;
        locked.release();
        (events, watch_wakes)
    }

    /// Decide-and-write: insert, update, keep, or delete based on the closure's
    /// [`Upsert`](crate::Upsert) decision. The closure must be short — the shard
    /// lock is held across it, so it must not panic and must not call back into
    /// this collection (the shard mutex is not reentrant).
    pub fn upsert<U: crate::IntoUpsert<[u8; V]>>(
        &self,
        key: &K,
        f: impl FnOnce(Option<&[u8; V]>) -> U,
    ) -> DbResult<U::Out> {
        self.atomic(key, |s| crate::upsert::upsert_via_held(s, key, f))
    }

    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]>> {
        // Before the first `?` — see `deadline_for`.
        let expires_at = self.deadline_for(key, value);
        // Fast path: key exists — write_update (dead-bytes tracking / in-place overwrite)
        if let Some(existing) = index.get(key) {
            let old_value = existing.value;
            self.claim_guard_bytes(key, Some(&old_value[..]), Some(&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,
                },
            );
            self.record_ttl(shard_id, key, expires_at);
            return Ok(Some(old_value));
        }
        self.claim_guard_bytes(key, None, Some(&value[..]))?;

        // Slow path: new key — write_new
        let loc = inner.write_new(shard_id as u8, key.as_bytes(), value)?;
        index.upsert(*key, MapEntry { loc, value: *value });
        self.record_ttl(shard_id, key, expires_at);
        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);
        self.claim_guard_bytes(key, Some(&old_value[..]), Some(&new_value[..]))?;
        // Before the first `?` — see `deadline_for`. The new value only exists
        // under the lock here, so this one cannot be hoisted out of it.
        let expires_at = self.deadline_for(key, &new_value);
        let 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,
            },
        );
        self.record_ttl(shard_id, key, expires_at);
        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);
        }
        // An insert only ever reaches here on an absent key, so `old` is
        // always `None` — still routes through the guard: a source that
        // claims must reserve its very first claim inside `db.tx()` too.
        self.claim_guard_bytes(key, None, Some(&value[..]))?;

        // Before the first `?` — see `deadline_for`.
        let expires_at = self.deadline_for(key, value);
        let loc = inner.write_new(shard_id as u8, key.as_bytes(), value)?;
        index.upsert(*key, MapEntry { loc, value: *value });
        self.record_ttl(shard_id, key, expires_at);
        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),
        };
        // T10 Task 5: a delete is a pure release ({k} -> ∅) — refused outside
        // `db.tx()` exactly like an insert's pure claim.
        self.claim_guard_bytes(key, Some(&old.value[..]), None)?;

        inner.write_tombstone(shard_id as u8, old.loc, key.as_bytes())?;
        let value = old.value;
        index.remove(key);
        self.record_ttl(shard_id, key, None);
        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())),
        })
    }

    /// Field-granular prefix / range / page builder. `None` unless opened `iterable(true)`.
    pub fn scope<const N: usize>(
        &self,
        key: &K,
    ) -> Option<crate::scope::Scope<'_, crate::scope::MapScan<&Self>, N>> {
        crate::scope::map_scope::<_, _, N>(self, key, dir_from(self.reversed()), self.iterable)
    }

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

    /// Like [`for_each`](Self::for_each), but stops when `f` returns `Break`.
    pub(crate) fn for_each_while(
        &self,
        mut f: impl FnMut(&K, &[u8; V]) -> ControlFlow<()>,
    ) -> ControlFlow<()> {
        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 {
                if let ControlFlow::Break(()) = f(&key, &value) {
                    return ControlFlow::Break(());
                }
            }
        }
        ControlFlow::Continue(())
    }

    // -- 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> {
        self.migrate_inner(f, true)
    }

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

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

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

    /// T10 Task 5: refuse a non-transactional write that would change this
    /// source's claim set, before any mutation. Byte-snapshot twin of
    /// `TypedTree::claim_guard` — see `ConstTree::claim_guard_bytes` for the
    /// full rationale (identical shape, `ConstMap` backs `ZeroMap`).
    #[allow(clippy::collapsible_if)]
    #[inline]
    fn claim_guard_bytes(&self, key: &K, old: Option<&[u8]>, new: Option<&[u8]>) -> DbResult<()> {
        #[cfg(feature = "armour")]
        if H::EMITS && !crate::txn_depth::in_stream_apply() {
            if let Some(targets) = self.hook.stream_targets() {
                let collection = self
                    .counter_dirty
                    .get()
                    .map(crate::counter_dirty::CounterDirty::name)
                    .unwrap_or("");
                return Err(DbError::ChangeStreamRequiresTx {
                    collection,
                    queue: targets.first_name(),
                });
            }
        }
        if !H::CLAIMS {
            return Ok(());
        }
        // F1 fix (review): see `ConstTree::claim_guard_bytes` — `H::CLAIMS`
        // is `true` for every `Projected` carrier, not only claiming ones;
        // check the runtime declaration before doing anything else.
        if self.hook.claim_targets().is_none() {
            return Ok(());
        }
        let collection = self
            .counter_dirty
            .get()
            .map(crate::counter_dirty::CounterDirty::name)
            .unwrap_or("");
        crate::unique::guard_nontx_write_bytes(
            collection,
            &self.hook,
            self.effective_projection_limits(),
            key,
            old,
            new,
        )
    }

    pub(crate) fn reconcile_projections(&self) -> DbResult<()> {
        if !H::PROJECTS {
            return Ok(());
        }
        let Some(targets) = self.hook.projection_targets() else {
            return Ok(());
        };
        let (max_rows, max_bytes) = self.effective_projection_limits();
        crate::projection::reconcile_carrier_projections(
            targets,
            |emit| {
                let mut err: Option<DbError> = None;
                self.for_each(|k, value| {
                    if err.is_some() {
                        return;
                    }
                    let mut sink =
                        crate::projection::ProjectionSink::new(targets, max_rows, max_bytes);
                    match self
                        .hook
                        .project(k, None, Some(&value[..]), &mut sink)
                        .and_then(|()| {
                            for row in sink.rows() {
                                emit(row.target, &row.key, &row.value)?;
                            }
                            Ok(())
                        }) {
                        Ok(()) => {}
                        Err(e) => err = Some(e),
                    }
                });
                err.map_or(Ok(()), Err)
            },
            true,
        )
    }

    /// Open-time / on-demand recount (T9 Task 8) — see `counter::recount` for
    /// the shared two-pass walk. No-op (`Ok(0)`) when this collection is not
    /// itself a counted source.
    pub(crate) fn reconcile_counters(&self) -> DbResult<u64> {
        if !H::COUNTS {
            return Ok(0);
        }
        let Some(targets) = self.hook.counter_targets() else {
            return Ok(0);
        };
        let (max_rows, max_bytes) = self.effective_projection_limits();
        crate::counter::recount(
            |k: &K, v: &[u8], sink| self.hook.counter_project(k, None, Some(v), sink),
            targets,
            (max_rows, max_bytes),
            &mut |visit| {
                let mut err: Option<DbError> = None;
                self.for_each(|k, value| {
                    if err.is_some() {
                        return;
                    }
                    if let Err(e) = visit(k, &value[..]) {
                        err = Some(e);
                    }
                });
                err.map_or(Ok(()), Err)
            },
        )
    }

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

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

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

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

/// 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>,
    pub(crate) held: Held<'a, ConstMap<K, V, H, D>>,
}

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 tree = self.tree;
        let sid = self.held.sid;
        Ctx::with_shard(&mut self.held, tree, sid, |locked, ev_out| {
            let (inner, index) = locked;
            let old = tree.put_locked(sid, &mut *inner, index, key, value)?;
            *ev_out = tree.defer_write_event(key, old, Some(*value));
            Ok(old)
        })
    }

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

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

    /// Read-modify-write within this shard: applies `f` to the current value and
    /// stores the result. Returns `Some(new_value)` if the key existed, `None`
    /// otherwise (closure not invoked). The closure must not be heavy — the shard
    /// + 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)
    }

    /// Decide-and-write within this shard. The closure must be short — shard lock is held.
    pub fn upsert<U: crate::IntoUpsert<[u8; V]>>(
        &mut self,
        key: &K,
        f: impl FnOnce(Option<&[u8; V]>) -> U,
    ) -> DbResult<U::Out> {
        crate::upsert::upsert_via_held(self, key, f)
    }

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

    pub fn try_get(&self, key: &K) -> DbResult<Option<[u8; V]>> {
        self.check_shard(key)?;
        Ok(self.held.locked.1.get(key).map(|e| e.value))
    }

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

    pub fn try_contains(&self, key: &K) -> DbResult<bool> {
        self.check_shard(key)?;
        Ok(self.held.locked.1.contains_key(key))
    }

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

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

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

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

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

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

#[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 non_iterable_map_scope_is_none() {
        let dir = tempfile::tempdir().unwrap();
        let m = ConstMap::<[u8; 8], 1>::open(dir.path(), Config::test()).unwrap();
        assert!(m.scope::<0>(&[0u8; 8]).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);
    }

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

    /// F2: `fetch_update` must fire `on_write` even when `sync()` returns `Err`.
    #[test]
    fn fetch_update_hook_fires_even_when_sync_fails() {
        use crate::durability::Fixed;
        use crate::fixed::shard::FAIL_NEXT_SYNC;

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

        map.put(&1u64.to_be_bytes(), &1u64.to_be_bytes()).unwrap();
        let baseline = map.hook.writes.load(AtomicOrdering::Relaxed);
        assert_eq!(baseline, 1);

        FAIL_NEXT_SYNC.store(true, Ordering::SeqCst);
        let res = map.fetch_update(&1u64.to_be_bytes(), |_| 9u64.to_be_bytes());
        assert!(res.is_err(), "injected sync failure must propagate");
        assert_eq!(
            map.hook.writes.load(AtomicOrdering::Relaxed),
            baseline + 1,
            "fetch_update on_write must fire despite sync error"
        );
    }

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

    /// Regression (WP-C, fixed-store F1/F2): a crash with `enable_fsync=false`
    /// can leave the same key durably present in two OCCUPIED, CRC-valid slots
    /// (delete rewrites 4 meta bytes in the old slot; the re-put lands in a new
    /// slot; the kernel persists the new page but not the old meta). Recovery
    /// must keep exactly one index entry, free the losing slot in
    /// `versions`+`bitmap`, and leave `bitmap.occupied() == len()`.
    #[test]
    fn fixed_map_recovery_dedups_duplicate_key() {
        use crate::durability::Fixed;

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

        let dir = tempfile::tempdir().unwrap();
        // Forge the post-crash duplicate directly (no crash simulation): two
        // distinct slots, same key, no clean shutdown (drop) → dirty full-scan
        // recovery on reopen.
        {
            let fixed = Fixed::open(dir.path(), cfg.clone(), 8, 8).unwrap();
            let mut inner = fixed.lock_shard(0);
            let a = inner.alloc_slot().unwrap();
            inner.write_slot(a, &key, &v1).unwrap();
            let b = inner.alloc_slot().unwrap();
            inner.write_slot(b, &key, &v2).unwrap();
            assert_ne!(a, b, "two distinct slots hold the same key");
            inner.sync().unwrap();
        }

        let m = crate::FixedMap::<[u8; 8], 8>::open(dir.path(), cfg).unwrap();
        assert_eq!(m.len(), 1, "exactly one index entry after dedup");
        assert_eq!(m.get(&key), Some(v2), "last-scanned slot (higher id) wins");

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