armdb 0.14.2

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

use crate::Key;

use crate::batch::Applied;
use crate::byte_view::ByteView;
use crate::cache::{BlockCache, BlockKey};
use crate::compaction::{CompactionIndex, compact_shard};
use crate::config::Config;
use crate::ctx::{Ctx, Family, Held, Owned, ShardGuards};
use crate::decoded_cache::DecodedCache;
use crate::disk_loc::DiskLoc;
use crate::engine::Engine;
use crate::error::{DbError, DbResult};
use crate::hook::{NoHook, WriteHook};
use crate::io::aligned_buf::AlignedBuf;
use crate::map_index::{Dir, dir_from};
use crate::recovery::recover_var_tree;
use crate::scope::DirRangeIter;
use crate::shard::ShardInner;
use crate::skiplist::iter::{SkipKeyIter, SkipKeyKeys};
use crate::skiplist::node::{SkipNode, VarNode, random_height};
use crate::skiplist::{BackFinger, InsertResult, SkipList};
use crate::sync::MutexGuard;
use crate::sync::{self};
use crate::ttl::{SweepOutcome, TTL_RETRY_BACKOFF_MS, TtlSlots};
use crate::upsert::Upsert;
use crate::value_cache::{ValueCache, ValueKey};
use crate::watch::{WatchEventInner, WatchWakeBatch};

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

/// New-value half of a Var write event: facade borrows the caller's `&[u8]`;
/// `update` / `own` may carry an owned [`ByteView`].
pub(crate) enum VarHookNew<'v> {
    Borrowed(&'v [u8]),
    Owned(ByteView),
}

impl VarHookNew<'_> {
    pub(crate) fn as_slice(&self) -> &[u8] {
        match self {
            Self::Borrowed(s) => s,
            Self::Owned(v) => v.as_ref(),
        }
    }

    pub(crate) fn into_owned(self) -> ByteView {
        match self {
            Self::Borrowed(s) => ByteView::from(s),
            Self::Owned(v) => v,
        }
    }
}

const MAX_STALE_RETRIES: usize = 3;

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

#[cfg(test)]
pub(crate) struct UpdateManyFailAfter(usize);

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

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

#[cfg(test)]
thread_local! {
    /// Counts off-disk value reads in [`read_value_cached_inner`] on the current
    /// thread. Cache hits and write-buffer reads are not counted.
    pub(crate) static TEST_DISK_VALUE_READS: std::cell::Cell<usize> =
        const { std::cell::Cell::new(0) };
    /// Counts locked value reads, including cache and write-buffer hits.
    pub(crate) static TEST_LOCKED_VALUE_READS: std::cell::Cell<usize> =
        const { std::cell::Cell::new(0) };
}

/// Snapshot of the current thread's off-disk value read count.
#[cfg(test)]
pub(crate) fn test_disk_value_reads() -> usize {
    TEST_DISK_VALUE_READS.with(|c| c.get())
}

#[cfg(test)]
pub(crate) fn test_locked_value_reads() -> usize {
    TEST_LOCKED_VALUE_READS.with(|c| c.get())
}

#[cfg(test)]
thread_local! {
    static COUNT_LIVE_KEYS_FAULT: std::cell::RefCell<Option<(usize, std::io::ErrorKind, String)>> =
        const { std::cell::RefCell::new(None) };
}

#[cfg(test)]
pub(crate) struct CountLiveKeysFaultGuard;

#[cfg(test)]
impl Drop for CountLiveKeysFaultGuard {
    fn drop(&mut self) {
        COUNT_LIVE_KEYS_FAULT.with(|fault| {
            *fault.borrow_mut() = None;
        });
    }
}

/// Fail `try_count_live_keys` after `keys_counted` live keys (0 = before the first).
#[cfg(test)]
pub(crate) fn fail_count_live_keys_after(
    keys_counted: usize,
    error: std::io::Error,
) -> CountLiveKeysFaultGuard {
    COUNT_LIVE_KEYS_FAULT.with(|fault| {
        let mut fault = fault.borrow_mut();
        assert!(
            fault.is_none(),
            "a count-live-keys fault is already registered"
        );
        *fault = Some((keys_counted, error.kind(), error.to_string()));
    });
    CountLiveKeysFaultGuard
}

#[cfg(test)]
fn injected_count_live_keys_fault(keys_counted: u64) -> Option<DbError> {
    COUNT_LIVE_KEYS_FAULT.with(|fault| {
        let fault = fault.borrow();
        let (trigger_at, kind, message) = fault.as_ref()?;
        (keys_counted == *trigger_at as u64)
            .then(|| DbError::Io(std::io::Error::new(*kind, message.clone())))
    })
}

/// Outcome of a locked read-modify-write step (no lock acquisition, no hook).
enum UpdateOutcome {
    /// Key absent — closure not invoked.
    Missing,
    /// Closure returned `None` (abort) — current value left unchanged.
    Unchanged(ByteView),
    /// Value replaced.
    Updated { old: ByteView, new: ByteView },
}

/// A tree with fixed-size keys and variable-length values.
/// Values are stored as `ByteView` (inline ≤20 bytes, heap with ref counting for larger).
/// Disk reads are cached at 4096-byte block granularity via `BlockCache`.
///
/// Each `VarTree` owns its storage engine — one tree = one database directory.
///
/// # Usage
///
/// ```ignore
/// let tree = VarTree::<16>::open("data/messages", Config::large_values().build())?;
/// tree.put(&key, b"hello world")?;
/// tree.close()?;
/// ```
///
/// # Iteration
///
/// `iter()`, `range()`, and `prefix_iter()` all return [`VarIter`] which
/// implements `Iterator + DoubleEndedIterator` with `Item = (K, ByteView)`.
/// Each `next()` / `next_back()` may perform disk I/O on a block-cache miss.
///
/// ```ignore
/// let latest = tree.prefix_iter(&group_id).take(50).collect::<Vec<_>>();
/// let oldest = tree.prefix_iter(&group_id).rev().take(10);  // DoubleEndedIterator
/// ```
///
/// # Write hooks
///
/// Uses [`WriteHook<K>`]. `on_write` fires on `put`/`insert`/`delete`/`cas`/`compare_delete`/`update`/`fetch_update`
/// and inside `atomic()` (buffered during the closure, replayed after the shard lock is released).
/// When `NEEDS_OLD_VALUE = false`, old value is omitted from `on_write` to skip disk I/O.
///
/// `on_init` fires once per live entry during `migrate()` or `replay_init()`
/// (enable via `NEEDS_INIT = true`).
///
/// # Batch operations
///
/// [`get_many`](Self::get_many) does one sorted finger-walk over the index for a
/// list of keys; [`update_many`](Self::update_many) applies a per-key closure
/// grouped by shard (per-shard isolation, no global atomicity). See
/// `docs/superpowers/specs/26-06-24-armdb-batch-tree-ops.md`.
pub struct VarTree<K: Key, H: WriteHook<K> = NoHook> {
    index: SkipList<VarNode<K>>,
    engine: Engine,
    block_cache: BlockCache,
    value_cache: ValueCache,
    compaction_threshold: f64,
    shard_prefix_bits: usize,
    reversed: bool,
    pub(crate) hook: H,
    /// Weak back-link for [`tx_deferred`](Self::tx_deferred): set once via
    /// [`bind_tx_self`](Self::bind_tx_self) when the tree lives in an `Arc`.
    tx_self: OnceLock<Weak<Self>>,
    ttl: TtlSlots<K>,
    /// `counters_dirty` marker (T9 Task 7) for the `VarTypedTree`
    /// 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> VarTree<K> {
    /// Open or create a `VarTree` 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, H: WriteHook<K>> VarTree<K, H> {
    pub(crate) fn floor_recovered_gsns(&self, gsn_high: u64, shard_gsn_high: &[u64]) {
        self.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.engine.shards().get(shard_id) {
                shard.set_durable_recovered_gsn(shard.durable_recovered_gsn().max(high));
            }
        }
    }

    /// Open or create a `VarTree` 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(true);
        if config.var_cache_warn_needed() {
            tracing::warn!(
                "Var collection opened with no cache (block_cache=0 and value_cache=0): every get hits disk"
            );
        }
        if config.app_cache_without_direct_io() {
            tracing::warn!(
                "Var collection has an app cache enabled (block_cache/value_cache) but direct_io is off: \
                 values are cached twice (app cache + OS page cache). Enable direct_io to avoid redundant \
                 RAM — exception: keeping the app cache for plaintext caching under encryption."
            );
        }
        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 reversed = config.reversed;
        let block_cache = BlockCache::new(&config.block_cache);
        let value_cache = ValueCache::new(&config.value_cache);
        let engine = Engine::open(path, config)?;

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

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

        let hints = tree.engine.hints();
        let outcome = recover_var_tree::<K>(
            &shard_dir_refs,
            &shard_ids,
            tree.index(),
            hints,
            #[cfg(feature = "encryption")]
            tree.engine.shard_ciphers(),
        )?;
        for tail in &outcome.active_tails {
            tree.engine.shards()[tail.shard_idx].apply_recovery_tail(tail)?;
        }
        for (shard_idx, dead) in outcome.shard_dead_bytes {
            tree.engine.shards()[shard_idx].install_dead_bytes(dead);
        }
        for (shard_idx, file_seqs) in outcome.shard_file_seqs {
            tree.engine.shards()[shard_idx].install_recovery_file_seqs(file_seqs);
        }
        // Unconditional for the same reason as `const_map.rs`: `VarTree` backs
        // `VarTypedTree`, a transaction participant, and the replay rule needs
        // this watermark in every build.
        for (shard_idx, max_gsn) in &outcome.shard_max_gsns {
            tree.engine.shards()[*shard_idx].set_durable_recovered_gsn(*max_gsn);
        }
        let max_gsn = outcome.max_gsn;

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

        Ok(tree)
    }

    /// Full clean shutdown: write hint files (if enabled), flush write buffers + fsync.
    pub fn clean_shutdown(&self) -> DbResult<()> {
        if self.engine.hints() {
            self.sync_hints()?;
        }
        // `close`, not `flush`: it also drains the background hint worker, which
        // is what makes "cleanly closed ⇒ every retired file has a hint" hold.
        self.engine.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.engine.flush_buffers()
    }

    /// Durable flush: write buffers + fsync active files. Migration path (D12):
    /// schema version is committed via fsync'd db.info; data must be durable first.
    /// Shards holding writes no durability barrier covers yet — the size of
    /// this collection's power-loss window right now.
    ///
    /// Counted shard by shard without a consistent cut, so a concurrent writer
    /// or barrier can move the answer mid-count; read it as a gauge.
    pub fn dirty_shards(&self) -> usize {
        self.engine.dirty_shards()
    }

    pub(crate) fn durable_flush(&self) -> DbResult<()> {
        self.engine.flush()
    }

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

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

    fn invalidate_blocks(&self, shard_id: u8, file_id: u32, total_bytes: u64) {
        self.block_cache
            .invalidate_file(shard_id, file_id, total_bytes);
    }

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

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

impl<K: Key, H: WriteHook<K>> Family for VarTree<K, H> {
    type Locked<'a>
        = MutexGuard<'a, ShardInner>
    where
        Self: 'a;
    type Event<'v> = (K, Option<ByteView>, Option<VarHookNew<'v>>);
    type Owned = (K, Option<ByteView>, Option<ByteView>);

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

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

    fn sync(&self, sid: usize) -> DbResult<()> {
        #[cfg(feature = "ctx-gates")]
        crate::ctx_gates_sync::record_sync_call();
        self.engine.shards()[sid].flush()
    }

    fn fire(&self, ev: Self::Event<'_>) {
        let (k, old, new) = ev;
        self.hook
            .on_write(&k, old.as_deref(), new.as_ref().map(|n| n.as_slice()));
    }

    fn own(ev: Self::Event<'_>) -> Self::Owned {
        let (k, old, new) = ev;
        (k, old, new.map(VarHookNew::into_owned))
    }

    fn fire_owned(&self, ev: &Self::Owned) {
        let (k, old, new) = ev;
        self.hook.on_write(k, old.as_deref(), new.as_deref());
    }

    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 = if self.hook.watch_old() {
            old.as_deref()
                .map(|bytes| bytes.to_vec().into_boxed_slice())
        } else {
            None
        };
        let new = new
            .as_ref()
            .map(|bytes| bytes.as_slice().to_vec().into_boxed_slice());
        channel.publish(
            std::sync::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| n.as_slice());
        self.apply_projection_for_bytes(k, old.as_deref(), 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_deref(), new.as_deref())
    }
}

impl<K: Key, H: WriteHook<K>> VarTree<K, H> {
    /// Build per-shard TTL indexes by walking live keys and reading values from disk.
    ///
    /// Rows whose values cannot be read are omitted from the index (with `warn`);
    /// they re-enter only after a successful write or `rebuild_ttl_index`.
    #[doc(hidden)]
    pub fn build_ttl_index(&self) -> DbResult<()> {
        if !H::EXPIRES {
            return Ok(());
        }
        let n = self.engine.shards().len();
        let slots = self.ttl.install_and_clear(n);

        let _guard = self.index.collector().enter();
        let mut current = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        let mut skipped = 0usize;
        while !current.is_null() {
            let node = unsafe { &*current };
            current = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
            if node.is_marked() {
                continue;
            }
            let sid = self.shard_for(&node.key);
            let locked = self.engine.shards()[sid].lock();
            if let Ok(bytes) = self.read_value_locked_result(&node.read_loc(), sid as u8, &locked) {
                let deadline = self.hook.deadline(&node.key, &bytes);
                sync::lock(&slots[sid]).record(&node.key, deadline);
            } else {
                tracing::warn!(
                    key = ?node.key.as_bytes(),
                    "ttl cold build: row omitted from index — value unreadable until rewrite or rebuild"
                );
                skipped += 1;
            }
        }
        if skipped > 0 {
            tracing::warn!(
                skipped,
                "ttl cold build finished with rows omitted from index — sweep will not see them until rewrite or rebuild"
            );
        }
        Ok(())
    }

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

    fn deadline_for_bytes(&self, key: &K, value: &[u8]) -> Option<u64> {
        if H::EXPIRES {
            self.hook.deadline(key, value)
        } else {
            None
        }
    }

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

    /// Test-only: write a deadline straight into the index, bypassing any write.
    ///
    /// **Not public API and not compiled out of release builds** — it is `pub`
    /// only so integration tests can reach it (a `test-utils` feature was the
    /// alternative; the plan forbids new feature flags). Calling it desynchronises
    /// the index from the data. The index is derived, so
    /// [`rebuild_ttl_index`](Self::rebuild_ttl_index) repairs it, but until then
    /// rows expire early or not at all.
    #[doc(hidden)]
    pub fn ttl_debug_record(&self, key: &K, deadline: Option<u64>) {
        self.ttl.record(self.shard_for(key), key, deadline);
    }

    /// Test-only: insert a `due` entry with no matching `at` entry, simulating the
    /// stale head a concurrent update leaves behind. Same caveats as
    /// [`ttl_debug_record`](Self::ttl_debug_record).
    #[doc(hidden)]
    pub fn ttl_debug_inject_stale(&self, key: &K, deadline: u64) {
        self.ttl
            .debug_inject_stale(self.shard_for(key), deadline, *key);
    }

    /// Sweep expired rows across shards (round-robin) until `budget` is spent or no
    /// validated due entries remain.
    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();

        let guard = self.index.collector().enter();

        while remaining > 0 && self.ttl.any_due(now_ms) {
            let sid = self.ttl.next_shard(shard_count);
            let mut shard_events: Vec<(K, Option<ByteView>, Option<ByteView>)> = 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 self.index.get(key.as_bytes(), &guard) {
                    Some(node) => {
                        let disk = node.read_loc();
                        match self.read_value_locked_result(&disk, sid as u8, &locked) {
                            Ok(bytes) => {
                                match crate::ttl::catch_deadline(|| {
                                    self.hook.deadline(&key, &bytes)
                                }) {
                                    Some(Some(dl)) if dl <= now_ms => {
                                        VarSweepAction::Delete(key, Some(bytes), candidate_dl)
                                    }
                                    Some(Some(dl)) => VarSweepAction::Reschedule(key, dl),
                                    // `ttl_decode_failed` runs the user's codec, so
                                    // it can panic just like the extractor. Unguarded,
                                    // the unwind escapes past `pop_due` — which has
                                    // already removed the key from both `due` and `at`
                                    // — and the row never expires again. Treat a panic
                                    // as "unreadable": reschedule and move on.
                                    Some(None) => {
                                        match crate::ttl::catch_deadline(|| {
                                            self.hook.ttl_decode_failed(&bytes)
                                        }) {
                                            Some(false) => VarSweepAction::Drop,
                                            _ => VarSweepAction::Retry(key),
                                        }
                                    }
                                    None => VarSweepAction::Retry(key),
                                }
                            }
                            Err(_) => VarSweepAction::Retry(key),
                        }
                    }
                    None => VarSweepAction::Drop,
                };

                match action {
                    VarSweepAction::Delete(key, stage2_bytes, candidate_dl) => {
                        drop(idx);
                        match self.delete_locked(sid, &mut locked, &guard, &key) {
                            Ok(existed) => {
                                idx = sync::lock(&slots[sid]);
                                if existed {
                                    let old = if H::NEEDS_OLD_VALUE && (H::NEEDS_WRITE || active) {
                                        stage2_bytes
                                    } else {
                                        None
                                    };
                                    let event = (key, old.clone(), None);
                                    self.publish_watch(&event, active, &mut wakes);
                                    if H::NEEDS_WRITE {
                                        shard_events.push((key, old, None));
                                    }
                                    outcome.deleted += 1;
                                }
                            }
                            Err(e) => {
                                sync::lock(&slots[sid]).record(&key, Some(candidate_dl));
                                let needs_sync = Self::should_sync(&locked);
                                locked.release();
                                // 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);
                                drop(context_active);
                                let _ = crate::projection::replay_owned_events(
                                    self,
                                    &shard_events,
                                    needs_sync,
                                    sid,
                                );
                                <Self as Family>::wake_watches(wakes);
                                return Err(e);
                            }
                        }
                    }
                    VarSweepAction::Reschedule(key, dl) => {
                        idx.record(&key, Some(dl));
                    }
                    VarSweepAction::Retry(key) => {
                        outcome.failed += 1;
                        idx.record(&key, Some(now_ms + TTL_RETRY_BACKOFF_MS));
                    }
                    VarSweepAction::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);
            <Self as Family>::wake_watches(wakes);
            replay?;
        }

        drop(guard);
        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(())
    }
}

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

    /// Get a value by key. Checks block cache first, then reads from disk.
    ///
    /// **O(log n)** — lock-free SkipList lookup; on a block-cache miss a brief shard lock
    /// fetches the file handles, then the `pread` runs outside the lock.
    pub fn get(&self, key: &K) -> Option<ByteView> {
        metrics::counter!("armdb.ops", "op" => "get", "tree" => "var_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("var_tree.get");
        let guard = self.index.collector().enter();
        let node = match self.index.get(key.as_bytes(), &guard) {
            Some(n) => n,
            None => {
                #[cfg(feature = "hot-path-tracing")]
                tracing::error!(
                    "VarTree get error: index.get returned None for key {:?}",
                    key.as_bytes()
                );
                return None;
            }
        };
        self.read_value_cached(node, &guard)
    }

    /// Get a value by key, returning `Err(KeyNotFound)` if absent or the real
    /// error (`Io`/`EncryptionError`/`StaleDiskLoc`) if present but unreadable.
    pub fn get_or_err(&self, key: &K) -> DbResult<ByteView> {
        let guard = self.index.collector().enter();
        let node = self
            .index
            .get(key.as_bytes(), &guard)
            .ok_or(DbError::KeyNotFound)?;
        self.read_value_cached_result(node, &guard)
    }

    /// Strict read: `Ok(None)` only when the key is genuinely absent; `Err` when the
    /// key exists but its value cannot be read.
    pub fn try_get(&self, key: &K) -> DbResult<Option<ByteView>> {
        let guard = self.index.collector().enter();
        match self.index.get(key.as_bytes(), &guard) {
            Some(node) => self.read_value_cached_result(node, &guard).map(Some),
            None => Ok(None),
        }
    }

    /// Strict first entry: `Ok(None)` only when there is no live entry; `Err` when
    /// the boundary live entry exists but its value cannot be read.
    pub fn try_first(&self) -> DbResult<Option<(K, ByteView)>> {
        let guard = self.index.collector().enter();
        let mut ptr = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        while !ptr.is_null() {
            let node = unsafe { &*ptr };
            if !node.is_marked() {
                return self
                    .read_value_cached_result(node, &guard)
                    .map(|v| Some((node.key, v)));
            }
            ptr = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
        }
        Ok(None)
    }

    /// Strict last entry (symmetric to `try_first`).
    pub fn try_last(&self) -> DbResult<Option<(K, ByteView)>> {
        let guard = self.index.collector().enter();
        let mut ptr = self.index.find_last(&guard);
        while !ptr.is_null() {
            let node = unsafe { &*ptr };
            if !node.is_marked() {
                return self
                    .read_value_cached_result(node, &guard)
                    .map(|v| Some((node.key, v)));
            }
            ptr = self.index.find_last_lt(node.key_bytes(), &guard);
        }
        Ok(None)
    }

    /// Insert or update a key-value pair.
    ///
    /// Returns `true` if a previous value for `key` existed (i.e. this was an
    /// overwrite), `false` for a fresh insert. The flag is read directly from
    /// the index update and adds no extra I/O or value decoding.
    pub fn put(&self, key: &K, value: &[u8]) -> DbResult<bool> {
        let expires_at = self.deadline_for_bytes(key, value);
        self.put_with_deadline(key, value, expires_at)
    }

    /// Like [`put`](Self::put) with a precomputed TTL deadline.
    pub(crate) fn put_with_deadline(
        &self,
        key: &K,
        value: &[u8],
        expires_at: Option<u64>,
    ) -> DbResult<bool> {
        metrics::counter!("armdb.ops", "op" => "put", "tree" => "var_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("var_tree.put");
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let guard = self.index.collector().enter();
            // Gated on NEEDS_WRITE as well, matching the `*Shard` path: this value
            // only ever feeds an event that is not produced when hooks are off.
            // The read can fail (`read_value_locked_result` -> DbResult), so an
            // unguarded read made the facade reject a write on an unreadable old
            // value while `atomic()` accepted it — the two contexts disagreeing on
            // whether the write happened at all.
            //
            // `|| H::CLAIMS` (T10 Task 5): the guard needs the old value too,
            // independent of `NEEDS_WRITE` — a claiming hook always forces
            // `NEEDS_OLD_VALUE`/`NEEDS_WRITE` true via `Projected` in practice,
            // but this does not lean on that coincidence.
            let old_value = if (H::NEEDS_OLD_VALUE
                && (H::NEEDS_WRITE || (H::WATCHES && crate::watch::context_active())))
                || H::CLAIMS
            {
                if let Some(node) = self.index.get(key.as_bytes(), &guard) {
                    let disk = node.read_loc();
                    Some(self.read_value_locked_result(&disk, shard_id as u8, locked)?)
                } else {
                    None
                }
            } else {
                None
            };
            self.claim_guard_bytes(key, old_value.as_deref(), Some(value))?;
            let existed = self.put_locked(shard_id, locked, &guard, key, value, expires_at)?;
            let ev = if H::NEEDS_WRITE || (H::WATCHES && crate::watch::context_active()) {
                Some((*key, old_value, Some(VarHookNew::Borrowed(value))))
            } else {
                None
            };
            *ev_out = ev;
            Ok(existed)
        })
    }

    /// 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]) -> DbResult<()> {
        metrics::counter!("armdb.ops", "op" => "insert", "tree" => "var_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("var_tree.insert");
        // Outside the shard lock: for a VarTyped collection the extractor decodes
        // the whole value, and the bytes are already in hand here — matching `put`.
        let expires_at = self.deadline_for_bytes(key, value);
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let guard = self.index.collector().enter();
            // T10 Task 5 review (F5): check presence *before* the guard, not
            // after — matches Const/Typed's `insert_locked`, which checks
            // `KeyExists` before ever calling the guard, so an insert on an
            // already-occupied key reports `KeyExists`, not
            // `UniqueIndexRequiresTx` (the guard has no reason to run at all
            // on that path — `insert_locked` below is about to reject the
            // write on presence alone). An insert only ever succeeds on an
            // absent key, so `old` is always `None` on the branch that does
            // call the guard — still routes through it (a source that
            // claims must reserve its very first claim inside `db.tx()`
            // too). Gated on `claim_targets().is_some()` (F1): a
            // non-claiming insert has no reason to pay this otherwise-
            // redundant presence lookup — `insert_locked` does its own.
            if (H::EMITS || (H::CLAIMS && self.hook.claim_targets().is_some()))
                && self.index.get(key.as_bytes(), &guard).is_none()
            {
                self.claim_guard_bytes(key, None, Some(value))?;
            }
            self.insert_locked(shard_id, locked, &guard, key, value, expires_at)?;
            let ev = if H::NEEDS_WRITE || (H::WATCHES && crate::watch::context_active()) {
                Some((*key, None, Some(VarHookNew::Borrowed(value))))
            } else {
                None
            };
            *ev_out = ev;
            Ok(())
        })
    }

    /// Delete a key. Returns `true` if the key existed.
    pub fn delete(&self, key: &K) -> DbResult<bool> {
        metrics::counter!("armdb.ops", "op" => "delete", "tree" => "var_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("var_tree.delete");
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let guard = self.index.collector().enter();
            // Gated on NEEDS_WRITE as well, matching the `*Shard` path: this value
            // only ever feeds an event that is not produced when hooks are off.
            // The read can fail (`read_value_locked_result` -> DbResult), so an
            // unguarded read made the facade reject a write on an unreadable old
            // value while `atomic()` accepted it — the two contexts disagreeing on
            // whether the write happened at all.
            //
            // `|| H::CLAIMS` (T10 Task 5): see `put_with_deadline`.
            let old_value = if (H::NEEDS_OLD_VALUE
                && (H::NEEDS_WRITE || (H::WATCHES && crate::watch::context_active())))
                || H::CLAIMS
            {
                if let Some(node) = self.index.get(key.as_bytes(), &guard) {
                    let disk = node.read_loc();
                    Some(self.read_value_locked_result(&disk, shard_id as u8, locked)?)
                } else {
                    None
                }
            } else {
                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, old_value.as_deref(), None)?;
            let existed = self.delete_locked(shard_id, locked, &guard, key)?;
            let ev =
                if existed && (H::NEEDS_WRITE || (H::WATCHES && crate::watch::context_active())) {
                    Some((*key, old_value, None))
                } else {
                    None
                };
            *ev_out = ev;
            Ok(existed)
        })
    }

    /// 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 VarShard<'_, K, H>) -> DbResult<R>,
    ) -> DbResult<R> {
        let _atomic_depth = crate::AtomicDepthGuard::try_enter()?;
        let shard_id = self.shard_for(shard_key);
        // Lock first, then enter seize — avoid pinning the epoch while waiting on the mutex.
        let locked = self.lock_shard(shard_id);
        let guard = self.index.collector().enter();
        let mut shard = VarShard {
            tree: self,
            held: Held {
                locked,
                sid: shard_id,
                events: Vec::new(),
                watch_wakes: WatchWakeBatch::default(),
                publish_watches: true,
            },
            guard,
        };
        let result = f(&mut shard);
        let VarShard { held, guard, .. } = shard;
        let Held {
            locked,
            events,
            sid,
            watch_wakes,
            ..
        } = held;
        let needs_sync = Self::should_sync(&locked);
        locked.release();
        let result =
            crate::projection::finish_atomic_teardown(self, sid, result, needs_sync, &events);
        drop(guard);
        <Self as Family>::wake_watches(watch_wakes);
        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()
    }

    /// Returns `true` if `key` already had a value (overwrite), `false` if newly inserted.
    fn put_locked(
        &self,
        shard_id: usize,
        inner: &mut ShardInner,
        guard: &seize::LocalGuard<'_>,
        key: &K,
        value: &[u8],
        expires_at: Option<u64>,
    ) -> DbResult<bool> {
        let (disk_loc, _gsn) = inner.append_entry(shard_id as u8, key.as_bytes(), value, false)?;

        // Fast path: key exists — atomic swap, no write_lock, no node allocation
        if let Some(existing) = self.index.get(key.as_bytes(), guard) {
            let old_disk = existing.read_loc();
            inner.add_dead_bytes(
                old_disk.file_id,
                crate::entry::entry_size(size_of::<K>(), old_disk.len),
            );
            existing.write_loc(disk_loc);
            self.record_ttl(shard_id, key, expires_at);
            return Ok(true);
        }

        // Slow path: new key — allocate node + take write_lock via insert
        let height = random_height();
        let node_ptr = VarNode::alloc(*key, disk_loc, height);

        match self.index.insert(node_ptr, guard) {
            InsertResult::Inserted => {
                self.record_ttl(shard_id, key, expires_at);
                Ok(false)
            }
            InsertResult::Exists(existing) => {
                // Race: another shard inserted same key between get and insert
                let old_disk = existing.read_loc();
                inner.add_dead_bytes(
                    old_disk.file_id,
                    crate::entry::entry_size(size_of::<K>(), old_disk.len),
                );
                existing.write_loc(disk_loc);
                self.record_ttl(shard_id, key, expires_at);
                unsafe {
                    VarNode::<K>::dealloc_node(node_ptr);
                }
                Ok(true)
            }
        }
    }

    fn insert_locked(
        &self,
        shard_id: usize,
        inner: &mut ShardInner,
        guard: &seize::LocalGuard<'_>,
        key: &K,
        value: &[u8],
        expires_at: Option<u64>,
    ) -> DbResult<()> {
        if self.index.get(key.as_bytes(), guard).is_some() {
            return Err(DbError::KeyExists);
        }

        let (disk_loc, _gsn) = inner.append_entry(shard_id as u8, key.as_bytes(), value, false)?;
        let height = random_height();
        let node_ptr = VarNode::alloc(*key, disk_loc, height);

        match self.index.insert(node_ptr, guard) {
            InsertResult::Inserted => {
                self.record_ttl(shard_id, key, expires_at);
                Ok(())
            }
            InsertResult::Exists(_existing) => {
                // Race: another path inserted this key after our `get` check and
                // before `index.insert`. Account for the entry we already appended
                // as dead bytes, free the unpublished node, and report KeyExists.
                inner.add_dead_bytes(
                    disk_loc.file_id,
                    crate::entry::entry_size(size_of::<K>(), disk_loc.len),
                );
                // SAFETY: node_ptr was just allocated by VarNode::alloc and has
                // not been published into the SkipList. It uniquely owns its
                // freshly-allocated DiskLoc box.
                unsafe { VarNode::<K>::dealloc_node(node_ptr) };
                Err(DbError::KeyExists)
            }
        }
    }

    fn delete_locked(
        &self,
        shard_id: usize,
        inner: &mut ShardInner,
        guard: &seize::LocalGuard<'_>,
        key: &K,
    ) -> DbResult<bool> {
        #[cfg(test)]
        if crate::durability::ttl_delete_should_fail() {
            return Err(DbError::Internal("test: TTL delete injected failure"));
        }
        if self.index.get(key.as_bytes(), guard).is_none() {
            return Ok(false);
        }

        inner.append_entry(shard_id as u8, key.as_bytes(), &[], true)?;

        let removed = self.index.remove(key.as_bytes(), guard);

        if let Some(node_ptr) = removed {
            let disk = unsafe { &*node_ptr }.read_loc();
            inner.add_dead_bytes(
                disk.file_id,
                crate::entry::entry_size(size_of::<K>(), disk.len),
            );
        }

        if removed.is_some() {
            self.record_ttl(shard_id, key, None);
        }

        Ok(removed.is_some())
    }

    /// Check if a key exists.
    pub fn contains(&self, key: &K) -> bool {
        let guard = self.index.collector().enter();
        self.index.get(key.as_bytes(), &guard).is_some()
    }

    /// Encoded value byte length for `key`, or `None` if absent.
    /// Reads only the in-memory index entry (`DiskLoc::len`); no disk I/O.
    pub fn entry_len(&self, key: &K) -> Option<u32> {
        let guard = self.index.collector().enter();
        self.index
            .get(key.as_bytes(), &guard)
            .map(|node| node.read_loc().len)
    }

    /// Return the first entry in index order, or `None` if empty.
    /// With `reversed=true` (default): the entry with the largest key.
    /// O(1) index lookup. May perform disk I/O on block-cache miss.
    pub fn first(&self) -> Option<(K, ByteView)> {
        let guard = self.index.collector().enter();
        let mut ptr = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        while !ptr.is_null() {
            let node = unsafe { &*ptr };
            if !node.is_marked()
                && let Some(v) = self.read_value_cached(node, &guard)
            {
                return Some((node.key, v));
            }
            ptr = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
        }
        None
    }

    /// Return the last entry in index order, or `None` if empty.
    /// With `reversed=true` (default): the entry with the smallest key.
    /// May perform disk I/O on block-cache miss.
    pub fn last(&self) -> Option<(K, ByteView)> {
        self.iter().next_back()
    }

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

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

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

    /// Iterate entries whose keys start with `prefix`.
    ///
    /// `reversed=true` (default): yields matching keys in DESC order.
    /// `next()` is O(1). The first reverse seek is expected O(log n); an m-row reverse walk is
    /// expected/amortized O(log n + m). Both may perform disk I/O on cache miss.
    pub fn prefix_iter(&self, prefix: &[u8]) -> VarIter<'_, K, H> {
        let guard = self.index.collector().enter();
        let (search_key, end) = self.prefix_bounds(prefix);
        let front = self.index.find_first_ge(search_key.as_bytes(), &guard);
        VarIter {
            tree: self,
            front,
            back: None,
            back_finger: BackFinger::empty(),
            last_front: None,
            last_back: None,
            end,
            start: Bound::Included(search_key),
            reversed: self.reversed,
            done: false,
            _guard: guard,
        }
    }

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

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

    /// Iterate entries in range defined by `start` and `end` bounds.
    ///
    /// Unlike [`range()`](Self::range), allows `Included`, `Excluded`, or `Unbounded`
    /// for each bound independently.
    ///
    /// `reversed=true` (default): DESC within range. `reversed=false`: ASC.
    /// `next()` is O(1). The first reverse seek is expected O(log n); an m-row reverse walk is
    /// expected/amortized O(log n + m). Both may perform disk I/O on cache miss.
    pub fn range_bounds(&self, start: Bound<&K>, end: Bound<&K>) -> VarIter<'_, K, H> {
        let guard = self.index.collector().enter();
        if self.reversed {
            let front = self.resolve_front(&end, &guard);
            VarIter {
                tree: self,
                front,
                back: None,
                back_finger: BackFinger::empty(),
                last_front: None,
                last_back: None,
                end: bound_owned(&start),
                start: bound_owned(&end),
                reversed: true,
                done: false,
                _guard: guard,
            }
        } else {
            let front = self.resolve_front(&start, &guard);
            VarIter {
                tree: self,
                front,
                back: None,
                back_finger: BackFinger::empty(),
                last_front: None,
                last_back: None,
                end: bound_owned(&end),
                start: bound_owned(&start),
                reversed: false,
                done: false,
                _guard: guard,
            }
        }
    }

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

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

    /// Strict range iterator: surfaces storage/decode errors instead of skipping.
    pub(crate) fn try_range_bounds(
        &self,
        start: Bound<&K>,
        end: Bound<&K>,
    ) -> TryVarIter<'_, K, H> {
        TryVarIter {
            inner: self.range_bounds(start, end),
            failed: false,
        }
    }

    pub(crate) fn try_range_bounds_dir(
        &self,
        start: Bound<&K>,
        end: Bound<&K>,
        dir: Dir,
    ) -> crate::scope::DirRangeIter<TryVarIter<'_, K, H>> {
        crate::scope::DirRangeIter::from_storage_match(
            self.try_range_bounds(start, end),
            dir == dir_from(self.reversed),
        )
    }

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

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

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

    /// Strict live-key count for read-only inspection: walks the skiplist under a
    /// seize guard and counts non-marked nodes without decoding values.
    pub(crate) fn try_count_live_keys(&self) -> DbResult<u64> {
        let guard = self.index.collector().enter();
        let mut count = 0u64;
        let mut current = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        while !current.is_null() {
            let node = unsafe { &*current };
            current = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
            if node.is_marked() {
                continue;
            }
            #[cfg(test)]
            if let Some(error) = injected_count_live_keys_fault(count) {
                return Err(error);
            }
            count += 1;
        }
        drop(guard);
        Ok(count)
    }

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

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

    /// Pre-populate the block cache with blocks containing live values.
    ///
    /// Walks the index to collect unique block offsets, sorts them for sequential I/O,
    /// then reads each block into the cache. Only blocks with live data are loaded —
    /// dead entries and garbage are skipped.
    pub fn warmup(&self) -> DbResult<()> {
        use std::collections::BTreeSet;

        let guard = self.index.collector().enter();

        // Collect unique (shard_id, file_id, block_offset) from live index entries
        let mut blocks: BTreeSet<(u8, u32, u64)> = BTreeSet::new();
        let mut current = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        while !current.is_null() {
            let node = unsafe { &*current };
            current = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
            if node.is_marked() {
                continue;
            }
            let disk = node.read_loc();
            // Values > 2 blocks are served by the value cache, never the block
            // cache — caching their first block here only evicts useful blocks.
            if disk.is_value_cache_routed() {
                continue;
            }
            let shard_id = self.shard_for(&node.key) as u8;
            let block_offset = disk.offset as u64 & !4095;
            blocks.insert((shard_id, disk.file_id, block_offset));
            // A value may spill into the next block (up to 2 blocks — beyond
            // that it is value-cache routed and skipped above). Without its
            // tail block the first read after warmup still pays a disk I/O in
            // `extract_from_block`'s `next_block()`.
            if (disk.offset & 4095) as usize + disk.len as usize > 4096 {
                blocks.insert((shard_id, disk.file_id, block_offset + 4096));
            }
        }
        drop(guard);

        // Read blocks in sorted order (sequential I/O per shard/file)
        for (shard_id, file_id, block_offset) in &blocks {
            let key = BlockKey {
                shard_id: *shard_id,
                file_id: *file_id,
                block_offset: *block_offset,
            };
            if self.block_cache.get(&key).is_some() {
                continue;
            }
            let shard = &self.engine.shards()[*shard_id as usize];
            let (buf, is_full_block) = shard.read_block(*file_id, *block_offset)?;
            if is_full_block {
                self.block_cache.insert(key, Arc::new(buf));
            }
        }

        Ok(())
    }

    pub(crate) fn index(&self) -> &SkipList<VarNode<K>> {
        &self.index
    }

    /// Iterate all entries and optionally mutate them. Call once at startup.
    ///
    /// The callback receives each (key, value_bytes) and returns `MigrateAction`:
    /// - `Keep` — no change (fires `on_init` if `H::NEEDS_INIT`); not counted
    /// - `Update(new_value)` — replace value (hook-free write, fires `on_init`)
    /// - `Delete` — remove entry (hook-free tombstone, no hooks)
    ///
    /// `on_write` is NEVER fired during migrate — see `docs/hooks.md` in the armdb crate.
    ///
    /// A value read failure aborts the migration with `Err` (fail-loud): the step
    /// is not committed and will re-run on next open. Consequence: one unreadable
    /// entry blocks opening until the underlying I/O issue is fixed (or data is
    /// restored from a replica/backup) — consistency over availability by design.
    ///
    /// Returns the number of mutated entries.
    pub fn migrate(
        &self,
        f: impl Fn(&K, &[u8]) -> crate::MigrateAction<ByteView>,
    ) -> DbResult<usize> {
        self.migrate_inner(f, true)
    }

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

        let guard = self.index.collector().enter();
        let mut current = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        let mut count = 0;
        while !current.is_null() {
            let node = unsafe { &*current };
            current = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
            if node.is_marked() {
                continue;
            }
            let value = match self.read_value_cached_result(node, &guard) {
                Ok(v) => v,
                Err(e) => {
                    tracing::error!(
                        key = ?node.key.as_bytes(),
                        error = %e,
                        "var_tree migrate: value read failed — aborting migration step"
                    );
                    return Err(e);
                }
            };
            match f(&node.key, &value) {
                MigrateAction::Keep => {
                    if fire_init && H::NEEDS_INIT {
                        self.hook.on_init(&node.key, &value);
                    }
                }
                MigrateAction::Update(new_value) => {
                    // T10 Task 5: `migrate_inner` calls `put_locked` directly
                    // — it does not route through `put_with_deadline`/
                    // `VarShard::put`, so the guard needs its own call here
                    // too, reusing `value` (this row's pre-migration
                    // snapshot, already read above to drive `f`) as `old`. A
                    // migration that reassigns a claimed value is refused
                    // exactly like a non-transactional write would be — it
                    // fails the open, per the same "no special case" rule as
                    // every other locked write body.
                    //
                    // Called under the shard lock (F6 fix, review) — not
                    // before it, like every other call site in this file.
                    // Migration runs single-threaded before the collection
                    // is published (no concurrent writer could race the
                    // check anyway), but there is no reason for this one
                    // site to be the exception to the stated invariant.
                    let shard_id = self.shard_for(&node.key);
                    {
                        let mut inner = self.engine.shards()[shard_id].lock();
                        self.claim_guard_bytes(&node.key, Some(&value), Some(&new_value))?;
                        let expires_at = self.deadline_for_bytes(&node.key, &new_value);
                        self.put_locked(
                            shard_id, &mut inner, &guard, &node.key, &new_value, expires_at,
                        )?;
                    }
                    if fire_init && H::NEEDS_INIT {
                        self.hook.on_init(&node.key, &new_value);
                    }
                    count += 1;
                }
                MigrateAction::Delete => {
                    // T10 Task 5: see the `Update` arm above — same "under
                    // the shard lock" fix (F6).
                    let shard_id = self.shard_for(&node.key);
                    let mut inner = self.engine.shards()[shard_id].lock();
                    self.claim_guard_bytes(&node.key, Some(&value), None)?;
                    self.delete_locked(shard_id, &mut inner, &guard, &node.key)?;
                    count += 1;
                }
            }
        }

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

    /// Replay `on_init` for every live entry. Used when no migration runs
    /// (Db calls this after `run_migration` returns `false`). Public users
    /// should invoke `migrate(|_, _| MigrateAction::Keep)` instead.
    pub(crate) fn replay_init(&self) {
        if !H::NEEDS_INIT {
            return;
        }
        let guard = self.index.collector().enter();
        let mut current = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        let mut count = 0usize;
        while !current.is_null() {
            let node = unsafe { &*current };
            current = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
            if node.is_marked() {
                continue;
            }
            let value = match self.read_value_cached(node, &guard) {
                Some(v) => v,
                None => {
                    tracing::warn!(
                        key = ?node.key.as_bytes(),
                        "var_tree replay_init: skipping entry — value read failed"
                    );
                    continue;
                }
            };
            self.hook.on_init(&node.key, &value);
            count += 1;
        }
        tracing::debug!(replayed = count, "var_tree replay_init complete");
    }

    /// 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 cfg = self.config();
        let (hook_rows, hook_bytes) = self.hook.projection_limits();
        (
            hook_rows.min(cfg.max_projected_rows),
            hook_bytes.min(cfg.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 (`VarTree` backs `VarTypedTree`, disk-resident values —
    /// callers already read the old value before the first mutation whenever
    /// a hook needs it, this just widens that existing gate to include
    /// `H::CLAIMS`).
    #[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 guard = self.index.collector().enter();
                let mut current = crate::skiplist::strip_mark(unsafe {
                    (*self.index.head_ptr())
                        .tower(0)
                        .load(std::sync::atomic::Ordering::Acquire)
                });
                let mut err: Option<DbError> = None;
                while !current.is_null() {
                    let node = unsafe { &*current };
                    current = crate::skiplist::strip_mark(
                        node.tower(0).load(std::sync::atomic::Ordering::Acquire),
                    );
                    if node.is_marked() {
                        continue;
                    }
                    if err.is_some() {
                        break;
                    }
                    // A failed read must abort: with delete_orphans=true, skipping
                    // the row shrinks the expected set and deletes live index entries.
                    let value = match crate::projection::reconcile_value_read_fault()
                        .and_then(|()| self.read_value_cached_result(node, &guard))
                    {
                        Ok(v) => v,
                        Err(e) => {
                            err = Some(e);
                            break;
                        }
                    };
                    let mut sink =
                        crate::projection::ProjectionSink::new(targets, max_rows, max_bytes);
                    match self
                        .hook
                        .project(&node.key, None, Some(value.as_ref()), &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.
    ///
    /// A failed value read aborts the whole scan rather than being skipped:
    /// skipping it would silently shrink `expected` and drive a live,
    /// merely-unreadable-right-now row to zero — the same reasoning as
    /// `reconcile_projections` above.
    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 guard = self.index.collector().enter();
                let mut current = crate::skiplist::strip_mark(unsafe {
                    (*self.index.head_ptr())
                        .tower(0)
                        .load(std::sync::atomic::Ordering::Acquire)
                });
                let mut err: Option<DbError> = None;
                while !current.is_null() {
                    let node = unsafe { &*current };
                    current = crate::skiplist::strip_mark(
                        node.tower(0).load(std::sync::atomic::Ordering::Acquire),
                    );
                    if node.is_marked() {
                        continue;
                    }
                    if err.is_some() {
                        break;
                    }
                    let value = match crate::projection::reconcile_value_read_fault()
                        .and_then(|()| self.read_value_cached_result(node, &guard))
                    {
                        Ok(v) => v,
                        Err(e) => {
                            err = Some(e);
                            break;
                        }
                    };
                    if let Err(e) = visit(&node.key, value.as_ref()) {
                        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()
    }

    /// Lock-free read body. Returns `Ok(v)` on success, `Err(StaleDiskLoc)` if
    /// compaction removed the file referenced by `disk` between the caller's
    /// snapshot and the read. Other `DbError`s indicate I/O or decryption
    /// failure.
    fn read_value_cached_inner(&self, disk: &DiskLoc, shard_id: u8) -> DbResult<ByteView> {
        let len = disk.len as usize;
        let start = (disk.offset & 4095) as usize;

        // Large values (> 2 blocks): write buffer first, then a lock-free disk
        // read. The value cache lives in `read_value_cached_result` (it has the
        // node for verify-on-hit); this node-less helper keeps its direct
        // StaleDiskLoc contract.
        if start + len > 8192 {
            {
                let shard = &self.engine.shards()[shard_id as usize];
                let inner = shard.lock();
                if inner.active.file_id == disk.file_id {
                    if let Some(bytes) = inner.write_buf.read(disk.offset as u64, len) {
                        return Ok(ByteView::new(bytes));
                    }
                    // Prefix flushed to disk, tail still buffered (page-granular
                    // non-force flush): stitch both halves under the shard lock.
                    if inner.straddles_write_buf(disk) {
                        return Ok(ByteView::from_vec(
                            inner.read_straddling_value_locked(disk)?,
                        ));
                    }
                }
            }
            let shard = &self.engine.shards()[shard_id as usize];
            #[cfg(test)]
            TEST_DISK_VALUE_READS.with(|c| c.set(c.get() + 1));
            return Ok(ByteView::from_vec(shard.read_value(
                disk.file_id,
                disk.offset,
                len,
            )?));
        }

        let block_offset = disk.offset as u64 & !4095;
        let cache_key = BlockKey {
            shard_id,
            file_id: disk.file_id,
            block_offset,
        };

        // 1. Block Cache (lock-free)
        if let Some(block) = self.block_cache.get(&cache_key) {
            metrics::counter!("armdb.block_cache.hit").increment(1);
            // Single-block values never straddle a page boundary, so they stay
            // fully lock-free. A 2-block value could straddle the active write
            // buffer's `base_offset`: its second block (`block_offset + 4096`) is
            // then unflushed, and `extract_from_block`'s closure would read it
            // zeroed from disk. Detect that under a brief lock (only for the
            // 2-block case) and stitch via the write buffer instead.
            if start + len > 4096 {
                let shard = &self.engine.shards()[shard_id as usize];
                let inner = shard.lock();
                if inner.active.file_id == disk.file_id && inner.straddles_write_buf(disk) {
                    return Ok(ByteView::from_vec(
                        inner.read_straddling_value_locked(disk)?,
                    ));
                }
            }
            return Self::extract_from_block(&block, start, len, || {
                self.get_or_read_block(shard_id, disk.file_id, block_offset + 4096)
            });
        }

        // 2. Write buffer — for unflushed data in active file (brief shard lock)
        {
            let shard = &self.engine.shards()[shard_id as usize];
            let inner = shard.lock();
            if inner.active.file_id == disk.file_id {
                if let Some(bytes) = inner.write_buf.read(disk.offset as u64, len) {
                    return Ok(ByteView::new(bytes));
                }
                // Prefix flushed to disk, tail still buffered (page-granular
                // non-force flush): stitch both halves under the shard lock,
                // avoiding the zero-padded second-block read in step 3.
                if inner.straddles_write_buf(disk) {
                    return Ok(ByteView::from_vec(
                        inner.read_straddling_value_locked(disk)?,
                    ));
                }
            }
        }

        // 3. Disk read + cache
        metrics::counter!("armdb.block_cache.miss").increment(1);
        #[cfg(test)]
        TEST_DISK_VALUE_READS.with(|c| c.set(c.get() + 1));
        let block = self.get_or_read_block(shard_id, disk.file_id, block_offset)?;
        Self::extract_from_block(&block, start, len, || {
            self.get_or_read_block(shard_id, disk.file_id, block_offset + 4096)
        })
    }

    /// Lock-free cached read, propagating `DbError`. Retries on `StaleDiskLoc`
    /// (fresh DiskLoc per attempt) up to `MAX_STALE_RETRIES`, then returns
    /// `Err(StaleDiskLoc)`. Used by `get_or_err` / `try_get` / `try_first` /
    /// `try_last`. The caller holds the `seize::LocalGuard` so `node` cannot be
    /// reclaimed between retries.
    fn read_value_cached_result(
        &self,
        node: &VarNode<K>,
        _guard: &seize::LocalGuard<'_>,
    ) -> DbResult<ByteView> {
        let shard_id = self.shard_for(&node.key) as u8;
        for _ in 0..MAX_STALE_RETRIES {
            let disk = node.read_loc();
            let is_large = disk.is_value_cache_routed();
            if is_large {
                let vkey = ValueKey {
                    shard_id,
                    file_id: disk.file_id,
                    offset: disk.offset,
                };
                if let Some(v) = self.value_cache.get(&vkey) {
                    // verify-on-hit: node still points here ⇒ file live ⇒ current.
                    let now = node.read_loc();
                    // (file_id, offset) uniquely identifies the bytes under append-only
                    // allocation, so len need not be compared — a matching loc proves the
                    // node still points here and the cached bytes are current.
                    if now.file_id == disk.file_id && now.offset == disk.offset {
                        metrics::counter!("armdb.value_cache.hit").increment(1);
                        return Ok(v);
                    }
                    continue; // node moved (edit/compaction) → retry with fresh loc
                }
                metrics::counter!("armdb.value_cache.miss").increment(1);
                match self.read_value_cached_inner(&disk, shard_id) {
                    Ok(v) => {
                        self.value_cache.insert(vkey, v.clone());
                        return Ok(v);
                    }
                    Err(DbError::StaleDiskLoc) => {
                        metrics::counter!("armdb.read.stale_retry", "tree" => "var_tree")
                            .increment(1);
                        continue;
                    }
                    Err(e) => return Err(e),
                }
            }
            match self.read_value_cached_inner(&disk, shard_id) {
                Ok(v) => return Ok(v),
                Err(DbError::StaleDiskLoc) => {
                    metrics::counter!("armdb.read.stale_retry", "tree" => "var_tree").increment(1);
                    continue;
                }
                Err(e) => return Err(e),
            }
        }
        Err(DbError::StaleDiskLoc)
    }

    /// Lock-free cached read that returns a **decoded** `Arc<T>`, consulting
    /// `cache` before touching any bytes.
    ///
    /// A deliberate sibling of [`read_value_cached_result`](Self::read_value_cached_result)
    /// rather than a generalisation of it: the byte path stays byte-identical so
    /// it cannot regress, at the price of duplicating the stale-retry protocol
    /// (spec `26-08-04-decoded-value-cache.md`, "Why a parallel function").
    /// Both loops must keep the same shape — a change to one needs the same
    /// change here.
    ///
    /// Only successful decodes are cached; a decode error is returned as is.
    pub(crate) fn read_decoded<T>(
        &self,
        node: &VarNode<K>,
        _guard: &seize::LocalGuard<'_>,
        cache: &DecodedCache<T>,
        decode: impl Fn(&ByteView) -> DbResult<T>,
    ) -> DbResult<Arc<T>> {
        let shard_id = self.shard_for(&node.key) as u8;
        for _ in 0..MAX_STALE_RETRIES {
            let disk = node.read_loc();
            let vkey = ValueKey {
                shard_id,
                file_id: disk.file_id,
                offset: disk.offset,
            };

            if let Some(v) = cache.get(&vkey) {
                // verify-on-hit: node still points here ⇒ file live ⇒ current.
                let now = node.read_loc();
                if now.file_id == disk.file_id && now.offset == disk.offset {
                    metrics::counter!("armdb.decoded_cache.hit").increment(1);
                    return Ok(v);
                }
                continue; // node moved (edit/compaction) → retry with fresh loc
            }
            if cache.is_enabled() {
                metrics::counter!("armdb.decoded_cache.miss").increment(1);
            }

            // Byte tier: same behaviour as the byte path, including populating
            // the value cache — the two budgets are independent by design.
            let routed = disk.is_value_cache_routed();
            let bytes = if routed && let Some(v) = self.value_cache.get(&vkey) {
                let now = node.read_loc();
                if now.file_id != disk.file_id || now.offset != disk.offset {
                    continue;
                }
                metrics::counter!("armdb.value_cache.hit").increment(1);
                v
            } else {
                if routed {
                    metrics::counter!("armdb.value_cache.miss").increment(1);
                }
                match self.read_value_cached_inner(&disk, shard_id) {
                    Ok(v) => {
                        if routed {
                            self.value_cache.insert(vkey, v.clone());
                        }
                        v
                    }
                    Err(DbError::StaleDiskLoc) => {
                        metrics::counter!("armdb.read.stale_retry", "tree" => "var_tree")
                            .increment(1);
                        continue;
                    }
                    Err(e) => return Err(e),
                }
            };

            let value = Arc::new(decode(&bytes)?);
            cache.insert(vkey, Arc::clone(&value), disk.len);
            return Ok(value);
        }
        Err(DbError::StaleDiskLoc)
    }

    /// Key-level decoded read: `Ok(None)` only when the key is genuinely absent;
    /// `Err` when it exists but cannot be read or decoded. The decoded twin of
    /// [`try_get`](Self::try_get).
    pub(crate) fn try_get_decoded<T>(
        &self,
        key: &K,
        cache: &DecodedCache<T>,
        decode: impl Fn(&ByteView) -> DbResult<T>,
    ) -> DbResult<Option<Arc<T>>> {
        metrics::counter!("armdb.ops", "op" => "get_shared", "tree" => "var_tree").increment(1);
        let guard = self.index.collector().enter();
        match self.index.get(key.as_bytes(), &guard) {
            Some(node) => self.read_decoded(node, &guard, cache, decode).map(Some),
            None => Ok(None),
        }
    }

    /// Lossy wrapper: `None` on absence *or* unreadable value. Keeps the existing
    /// hot-path `get` contract.
    fn read_value_cached(
        &self,
        node: &VarNode<K>,
        guard: &seize::LocalGuard<'_>,
    ) -> Option<ByteView> {
        let result = self.read_value_cached_result(node, guard);
        #[cfg(feature = "hot-path-tracing")]
        if let Err(ref _e) = result {
            tracing::error!("VarTree read_value_cached error: {:?}", _e);
        }
        result.ok()
    }

    fn extract_from_block(
        block: &AlignedBuf,
        start: usize,
        len: usize,
        next_block: impl FnOnce() -> DbResult<Arc<AlignedBuf>>,
    ) -> DbResult<ByteView> {
        debug_assert!(
            start + len <= 8192,
            "extract_from_block supports at most 2 blocks (8192 bytes)"
        );
        if start + len <= 4096 {
            Ok(ByteView::new(&block[start..start + len]))
        } else {
            let next = next_block()?;
            let first_part = &block[start..];
            let second_len = len - first_part.len();
            let mut combined = Vec::with_capacity(len);
            combined.extend_from_slice(first_part);
            combined.extend_from_slice(&next[..second_len]);
            Ok(ByteView::from_vec(combined))
        }
    }

    /// Get a block from cache, or read it from disk and cache it.
    fn get_or_read_block(
        &self,
        shard_id: u8,
        file_id: u32,
        block_offset: u64,
    ) -> DbResult<Arc<AlignedBuf>> {
        let key = BlockKey {
            shard_id,
            file_id,
            block_offset,
        };
        if let Some(cached) = self.block_cache.get(&key) {
            return Ok(cached);
        }
        let shard = &self.engine.shards()[shard_id as usize];
        let (buf, is_full_block) = shard.read_block(file_id, block_offset)?;
        let arc = Arc::new(buf);
        // Only cache settled blocks: entirely within an immutable file, or
        // entirely below the active file's write-buffer `base_offset`. The
        // block holding `base_offset` has a zero-padded tail that the next
        // flush overwrites, so caching it would serve stale zeros.
        if is_full_block {
            self.block_cache.insert(key, arc.clone());
        }
        Ok(arc)
    }

    /// 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.
    ///
    /// **Caveat:** Holds the shard lock while reading the current value. On a
    /// block-cache miss this performs disk I/O under the lock, blocking all
    /// writes to the same shard. Pre-warm the cache to avoid latency spikes.
    pub fn cas(&self, key: &K, expected: &[u8], new_value: &[u8]) -> DbResult<()> {
        metrics::counter!("armdb.ops", "op" => "cas", "tree" => "var_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("var_tree.cas");
        // Outside the shard lock — see `insert`. The new bytes are final here
        // regardless of whether the compare succeeds.
        let expires_at = self.deadline_for_bytes(key, new_value);
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let guard = self.index.collector().enter();
            let existing = self
                .index
                .get(key.as_bytes(), &guard)
                .ok_or(DbError::KeyNotFound)?;

            let disk = existing.read_loc();
            let current = self.read_value_locked_result(&disk, shard_id as u8, locked)?;
            if current.as_ref() != 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(current.as_ref()), Some(new_value))?;

            let (new_disk_loc, _gsn) =
                locked.append_entry(shard_id as u8, key.as_bytes(), new_value, false)?;

            let old_disk = existing.read_loc();
            locked.add_dead_bytes(
                old_disk.file_id,
                crate::entry::entry_size(size_of::<K>(), old_disk.len),
            );
            existing.write_loc(new_disk_loc);

            self.record_ttl(shard_id, key, expires_at);

            let ev = if H::NEEDS_WRITE || (H::WATCHES && crate::watch::context_active()) {
                Some((
                    *key,
                    if H::NEEDS_OLD_VALUE {
                        Some(current)
                    } else {
                        None
                    },
                    Some(VarHookNew::Borrowed(new_value)),
                ))
            } else {
                None
            };
            *ev_out = ev;
            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.
    ///
    /// **Caveat:** Holds the shard lock while reading the current value. On a
    /// block-cache miss this performs disk I/O under the lock, blocking all
    /// writes to the same shard. Pre-warm the cache to avoid latency spikes.
    pub fn compare_delete(&self, key: &K, expected: &[u8]) -> DbResult<()> {
        metrics::counter!("armdb.ops", "op" => "compare_delete", "tree" => "var_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("var_tree.compare_delete");
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let guard = self.index.collector().enter();
            let existing = self
                .index
                .get(key.as_bytes(), &guard)
                .ok_or(DbError::KeyNotFound)?;

            let disk = existing.read_loc();
            let current = self.read_value_locked_result(&disk, shard_id as u8, locked)?;
            if current.as_ref() != expected {
                return Err(DbError::CasMismatch);
            }
            // T10 Task 5: `compare_delete` has its own inline body — a pure
            // release ({k} -> ∅), refused outside `db.tx()`.
            self.claim_guard_bytes(key, Some(current.as_ref()), None)?;

            let removed = self.delete_locked(shard_id, locked, &guard, key)?;
            debug_assert!(
                removed,
                "compare_delete: key must exist after a successful match"
            );

            let ev = if H::NEEDS_WRITE || (H::WATCHES && crate::watch::context_active()) {
                Some((
                    *key,
                    if H::NEEDS_OLD_VALUE {
                        Some(current)
                    } else {
                        None
                    },
                    None,
                ))
            } else {
                None
            };
            *ev_out = ev;
            Ok(())
        })
    }

    /// Atomically read-modify-write. Returns `Some(new_value)` if key existed, `None` otherwise.
    /// The closure receives the current value and returns a new ByteView.
    /// The closure must not be heavy (shard lock is held).
    ///
    /// **Caveat:** Holds the shard lock while reading the current value. On a
    /// block-cache miss this performs disk I/O under the lock, blocking all
    /// writes to the same shard. Pre-warm the cache to avoid latency spikes.
    pub fn update(&self, key: &K, f: impl FnOnce(&[u8]) -> ByteView) -> DbResult<Option<ByteView>> {
        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]) -> ByteView,
    ) -> DbResult<Option<ByteView>> {
        self.update_inner(key, f, true)
    }

    /// Core read-modify-write logic: **no** lock acquisition, **no** hook, **no** drop.
    /// Reads the current value (may hit disk on cache miss), runs the closure, and on
    /// a real update appends the new entry + accounts for dead bytes.
    fn try_update_locked(
        &self,
        shard_id: usize,
        inner: &mut ShardInner,
        guard: &seize::LocalGuard<'_>,
        key: &K,
        f: impl FnOnce(&[u8]) -> DbResult<Option<ByteView>>,
    ) -> DbResult<UpdateOutcome> {
        let existing = match self.index.get(key.as_bytes(), guard) {
            Some(n) => n,
            None => return Ok(UpdateOutcome::Missing),
        };
        let disk = existing.read_loc();
        let current = self.read_value_locked_result(&disk, shard_id as u8, inner)?;

        let new_value = match f(&current)? {
            Some(v) => v,
            None => return Ok(UpdateOutcome::Unchanged(current)),
        };
        self.claim_guard_bytes(key, Some(current.as_ref()), Some(&new_value))?;

        let (new_disk_loc, _gsn) =
            inner.append_entry(shard_id as u8, key.as_bytes(), &new_value, false)?;
        let old_disk = existing.read_loc();
        inner.add_dead_bytes(
            old_disk.file_id,
            crate::entry::entry_size(size_of::<K>(), old_disk.len),
        );
        existing.write_loc(new_disk_loc);

        self.record_ttl(shard_id, key, self.deadline_for_bytes(key, &new_value));

        Ok(UpdateOutcome::Updated {
            old: current,
            new: new_value,
        })
    }

    pub(crate) fn try_update_inner(
        &self,
        key: &K,
        f: impl FnOnce(&[u8]) -> DbResult<Option<ByteView>>,
        return_old: bool,
    ) -> DbResult<Option<ByteView>> {
        metrics::counter!("armdb.ops", "op" => "update", "tree" => "var_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("var_tree.update");
        let shard_id = self.shard_for(key);
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let guard = self.index.collector().enter();
            match self.try_update_locked(shard_id, locked, &guard, key, f)? {
                UpdateOutcome::Missing => Ok(None),
                UpdateOutcome::Unchanged(current) => Ok(Some(current)),
                UpdateOutcome::Updated { old, new } => {
                    // Hoist the `H::NEEDS_WRITE` branch: with hooks off no event is
                    // built, so `new` moves into the result rather than being cloned
                    // under the lock for a value that is immediately dropped.
                    let (result, ev) =
                        if H::NEEDS_WRITE || (H::WATCHES && crate::watch::context_active()) {
                            let event_old = if H::NEEDS_OLD_VALUE {
                                Some(old.clone())
                            } else {
                                None
                            };
                            let result = Some(if return_old { old } else { new.clone() });
                            (
                                result,
                                Some((*key, event_old, Some(VarHookNew::Owned(new)))),
                            )
                        } else {
                            (Some(if return_old { old } else { new }), None)
                        };
                    {
                        *ev_out = ev;
                        Ok(result)
                    }
                }
            }
        })
    }

    fn update_inner(
        &self,
        key: &K,
        f: impl FnOnce(&[u8]) -> ByteView,
        return_old: bool,
    ) -> DbResult<Option<ByteView>> {
        self.try_update_inner(key, |bytes| Ok(Some(f(bytes))), return_old)
    }

    /// Read value when shard lock is already held, propagating `DbError`.
    /// Used by both `read_value_locked` (Option wrapper) and the large-value
    /// fallback in `read_value_cached_inner`.
    fn read_value_locked_result(
        &self,
        disk: &DiskLoc,
        shard_id: u8,
        inner: &ShardInner,
    ) -> DbResult<ByteView> {
        #[cfg(test)]
        TEST_LOCKED_VALUE_READS.with(|c| c.set(c.get() + 1));
        let len = disk.len as usize;

        // 1. Write buffer (for unflushed data), or a straddle read when the
        //    value's prefix was flushed to disk (page-granular) while its tail is
        //    still buffered. Both halves are stable under the held shard lock.
        if inner.active.file_id == disk.file_id {
            if let Some(bytes) = inner.write_buf.read(disk.offset as u64, len) {
                return Ok(ByteView::new(bytes));
            }
            if inner.straddles_write_buf(disk) {
                return Ok(ByteView::from_vec(
                    inner.read_straddling_value_locked(disk)?,
                ));
            }
        }

        // 2. Block cache (lock-free, single-block fast path)
        let block_offset = disk.offset as u64 & !4095;
        let start = (disk.offset & 4095) as usize;
        if start + len <= 4096 {
            let cache_key = BlockKey {
                shard_id,
                file_id: disk.file_id,
                block_offset,
            };
            if let Some(block) = self.block_cache.get(&cache_key) {
                return Ok(ByteView::new(&block[start..start + len]));
            }

            // 2b. Cache miss — read block under held lock, populate cache
            let (buf, is_full_block) = inner.read_block_locked(disk.file_id, block_offset)?;
            let arc = Arc::new(buf);
            if is_full_block {
                self.block_cache.insert(cache_key, arc.clone());
            }
            return Ok(ByteView::new(&arc[start..start + len]));
        }

        // 3. Multi-block: disk read via encryption-aware helper (no cache).
        let bytes = inner.read_value_from_disk_locked(disk)?;
        Ok(ByteView::new(&bytes))
    }

    pub fn shard_for(&self, key: &K) -> usize {
        crate::shard_for_key(key, self.shard_prefix_bits, self.engine.shards().len())
    }

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

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

    pub(crate) fn shard_count(&self) -> usize {
        self.engine.shards().len()
    }

    /// 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(
                "VarTree::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 `VarTree` (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) -> VarShard<'_, K, H> {
        let locked = self.lock_shard(sid);
        let guard = self.index.collector().enter();
        VarShard {
            tree: self,
            held: Held {
                locked,
                sid,
                events: Vec::new(),
                watch_wakes: WatchWakeBatch::default(),
                publish_watches: true,
            },
            guard,
        }
    }

    /// Release a shard built by [`tx_shard`], returning its buffered events.
    /// The `seize` guard is dropped by the caller — `Owned` events carry copied bytes.
    pub(crate) fn tx_release_shard(
        shard: VarShard<'_, K, H>,
    ) -> (
        Vec<<Self as Family>::Owned>,
        WatchWakeBatch,
        seize::LocalGuard<'_>,
    ) {
        let VarShard { held, guard, .. } = shard;
        let Held {
            locked,
            events,
            watch_wakes,
            ..
        } = held;
        locked.release();
        (events, watch_wakes, guard)
    }

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

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

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

    /// Batch point-read (tier-3 finger-walk). **Lossy**, mirroring single-key
    /// [`get`](Self::get): decode/disk/`StaleDiskLoc` errors become `None`. Values
    /// are read in list order under one guard through the existing cache path, so
    /// stale-retry runs against each node's current `DiskLoc`. Result aligned to `keys`.
    pub fn get_many(&self, keys: &[K]) -> Vec<Option<ByteView>> {
        let n = keys.len();
        let mut out: Vec<Option<ByteView>> = (0..n).map(|_| None).collect();
        if n == 0 {
            return out;
        }
        let mut order: Vec<usize> = (0..n).collect();
        order.sort_by(|&a, &b| {
            let ord = keys[a].as_bytes().cmp(keys[b].as_bytes());
            if self.reversed { ord.reverse() } else { ord }
        });
        let guard = self.index.collector().enter();
        let mut finger = self.index.finger();
        for &i in &order {
            let kb = keys[i].as_bytes();
            if let Some(node) = self.index.finger_seek(&mut finger, kb, &guard)
                && node.key_bytes() == kb
            {
                out[i] = self.read_value_cached(node, &guard);
            }
        }
        out
    }

    /// Batch decoded point-read: the [`get_many`](Self::get_many) finger walk
    /// with [`read_decoded`](Self::read_decoded) in the body. **Lossy**, like
    /// `get_many`: an absent key, a read error or a decode error is `None` at
    /// that index. One sorted walk under one guard; result aligned to `keys`.
    pub(crate) fn get_many_decoded<T>(
        &self,
        keys: &[K],
        cache: &DecodedCache<T>,
        decode: impl Fn(&ByteView) -> DbResult<T>,
    ) -> Vec<Option<Arc<T>>> {
        let n = keys.len();
        let mut out: Vec<Option<Arc<T>>> = (0..n).map(|_| None).collect();
        if n == 0 {
            return out;
        }
        let mut order: Vec<usize> = (0..n).collect();
        order.sort_by(|&a, &b| {
            let ord = keys[a].as_bytes().cmp(keys[b].as_bytes());
            if self.reversed { ord.reverse() } else { ord }
        });
        let guard = self.index.collector().enter();
        let mut finger = self.index.finger();
        for &i in &order {
            let kb = keys[i].as_bytes();
            if let Some(node) = self.index.finger_seek(&mut finger, kb, &guard)
                && node.key_bytes() == kb
            {
                out[i] = self.read_decoded(node, &guard, cache, &decode).ok();
            }
        }
        out
    }

    /// Batch read-modify-write (tier-2, shard-grouped). See [`crate::ConstTree::update_many`]
    /// for the contract. The old value is read from disk under the shard lock via the
    /// existing locked read path. VarTree is Bitcask-engine only — no per-op sync.
    pub fn update_many<P>(
        &self,
        items: Vec<(K, P)>,
        f: impl Fn(&K, Option<&[u8]>, &P) -> Upsert<ByteView>,
    ) -> DbResult<Vec<(K, Applied<ByteView>)>> {
        let n = items.len();
        let mut by_shard: std::collections::BTreeMap<usize, Vec<usize>> =
            std::collections::BTreeMap::new();
        for (i, (k, _)) in items.iter().enumerate() {
            by_shard.entry(self.shard_for(k)).or_default().push(i);
        }

        let mut out: Vec<Option<(K, Applied<ByteView>)>> = (0..n).map(|_| None).collect();
        let mut events: Vec<(K, Option<ByteView>, Option<ByteView>)> = Vec::new();
        let mut wakes = WatchWakeBatch::default();
        let mut err: Option<DbError> = None;

        'shards: for (shard_id, idxs) in by_shard {
            let mut inner = self.engine.shards()[shard_id].lock();
            let producer = self.watch_enter();
            let active = producer.active();
            let context_active = H::WATCHES.then(|| crate::watch::ContextActiveGuard::set(active));
            let guard = self.index.collector().enter();
            for &i in &idxs {
                #[cfg(test)]
                if UPDATE_MANY_FAIL_AFTER.with(|remaining| match remaining.get() {
                    0 => true,
                    usize::MAX => false,
                    value => {
                        remaining.set(value - 1);
                        false
                    }
                }) {
                    err = Some(DbError::Internal("test: update_many injected failure"));
                    break;
                }
                let (k, p) = &items[i];
                // Read the old value unconditionally (no NEEDS_OLD_VALUE gate, unlike single-key
                // put/delete): the closure receives it and it is part of the returned
                // Applied<ByteView> envelope, so it is always needed regardless of the hook.
                let cur: Option<ByteView> = match self.index.get(k.as_bytes(), &guard) {
                    Some(node) => {
                        let disk = node.read_loc();
                        match self.read_value_locked_result(&disk, shard_id as u8, &inner) {
                            Ok(v) => Some(v),
                            Err(e) => {
                                err = Some(e);
                                break;
                            }
                        }
                    }
                    None => None,
                };
                let decision = f(k, cur.as_deref(), p);
                // T10 Task 5: before any mutation. `cur` is already read
                // unconditionally above, so this costs nothing extra beyond
                // `claim_guard_bytes`'s own `H::CLAIMS` check.
                let guard_result = match &decision {
                    Upsert::Put(v) => self.claim_guard_bytes(k, cur.as_deref(), Some(v)),
                    Upsert::Delete => self.claim_guard_bytes(k, cur.as_deref(), None),
                    Upsert::Keep => Ok(()),
                };
                if let Err(e) = guard_result {
                    err = Some(e);
                    break;
                }
                let applied = match decision {
                    Upsert::Put(v) => {
                        let expires_at = self.deadline_for_bytes(k, &v);
                        match self.put_locked(shard_id, &mut inner, &guard, k, &v, expires_at) {
                            Ok(_existed) => {
                                if H::NEEDS_WRITE || active {
                                    let event =
                                        (*k, cur.clone(), Some(VarHookNew::Owned(v.clone())));
                                    self.publish_watch(&event, active, &mut wakes);
                                    if H::NEEDS_WRITE {
                                        events.push((*k, cur.clone(), Some(v.clone())));
                                    }
                                }
                                Applied::Written { old: cur, new: v }
                            }
                            Err(e) => {
                                err = Some(e);
                                break;
                            }
                        }
                    }
                    Upsert::Keep => Applied::Kept,
                    Upsert::Delete => {
                        match self.delete_locked(shard_id, &mut inner, &guard, k) {
                            Ok(true) => {
                                // delete_locked returned Ok(true) under the shard lock, so the key existed and
                                // `cur` (read above under the same lock) is provably Some — not a "shouldn't happen".
                                let old = cur.expect("existing key has an old value");
                                if H::NEEDS_WRITE || active {
                                    let event = (*k, Some(old.clone()), None);
                                    self.publish_watch(&event, active, &mut wakes);
                                    if H::NEEDS_WRITE {
                                        events.push((*k, Some(old.clone()), None));
                                    }
                                }
                                Applied::Deleted(old)
                            }
                            Ok(false) => Applied::Kept, // delete-miss
                            Err(e) => {
                                err = Some(e);
                                break;
                            }
                        }
                    }
                };
                out[i] = Some((*k, applied));
            }
            drop(inner);
            drop(guard);
            drop(context_active);
            if err.is_some() {
                break 'shards;
            }
        }

        if H::NEEDS_WRITE {
            for (k, old, new) in &events {
                self.hook.on_write(k, old.as_deref(), new.as_deref());
            }
        }

        <Self as Family>::wake_watches(wakes);

        if let Some(e) = err {
            return Err(e);
        }
        Ok(out
            .into_iter()
            .map(|slot| slot.expect("every index assigned on the success path"))
            .collect())
    }
}

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

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

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

        if header.is_tombstone() {
            let guard = self.index.collector().enter();
            let removed = self.index.remove(key.as_bytes(), &guard);
            match removed {
                Some(node_ptr) => {
                    let old_disk = unsafe { &*node_ptr }.read_loc();
                    Ok(ApplyOutcome::TombstoneRemoved(old_disk))
                }
                None => Ok(ApplyOutcome::Inserted), // no-op tombstone — no dead bytes
            }
        } else {
            let guard = self.index.collector().enter();
            let height = random_height();
            let node_ptr = VarNode::alloc(key, disk, height);
            match self.index.insert(node_ptr, &guard) {
                InsertResult::Inserted => Ok(ApplyOutcome::Inserted),
                InsertResult::Exists(existing) => {
                    let old_disk = existing.read_loc();
                    existing.write_loc(disk);
                    unsafe {
                        VarNode::<K>::dealloc_node(node_ptr);
                    }
                    Ok(ApplyOutcome::Replaced(old_disk))
                }
            }
        }
    }

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

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

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

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

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

    pub fn start_replication_server_with_options(
        &self,
        bind_addr: std::net::SocketAddr,
        signal: crate::shutdown::ShutdownSignal,
        options: crate::replication::ReplicationServerOptions,
    ) -> crate::error::DbResult<crate::replication::ReplicationServer> {
        let consumers = self.install_replication_producers()?;
        crate::replication::ReplicationServer::start_with_options(
            bind_addr,
            self.engine.shards().clone(),
            consumers,
            self.engine.config().max_file_size,
            signal,
            options,
        )
    }

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

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

    pub fn start_replication_client_with_options(
        &self,
        leader_addr: std::net::SocketAddr,
        registry: std::sync::Arc<crate::replication::ReplicationRegistry>,
        signal: crate::shutdown::ShutdownSignal,
        options: crate::replication::ReplicationClientOptions,
    ) -> crate::error::DbResult<crate::replication::ReplicationClient> {
        crate::replication::ReplicationClient::start_with_options(
            leader_addr,
            self.engine.shards().clone(),
            registry,
            size_of::<K>() as u16,
            signal,
            options,
        )
    }
}

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

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

impl<K: Key, H: WriteHook<K>> VarShard<'_, K, H> {
    /// Insert or overwrite `key`. Returns `true` if the key already existed (overwrite),
    /// `false` for a fresh insert.
    pub fn put(&mut self, key: &K, value: &[u8]) -> DbResult<bool> {
        self.check_shard(key)?;
        let tree = self.tree;
        let sid = self.held.sid;
        let guard = &self.guard;
        Ctx::with_shard(&mut self.held, tree, sid, |locked, ev_out| {
            // `|| H::CLAIMS` (T10 Task 5): see `VarTree::put_with_deadline`.
            let old = if (H::NEEDS_OLD_VALUE
                && (H::NEEDS_WRITE || (H::WATCHES && crate::watch::context_active())))
                || H::CLAIMS
            {
                if let Some(node) = tree.index.get(key.as_bytes(), guard) {
                    let disk = node.read_loc();
                    Some(tree.read_value_locked_result(&disk, sid as u8, locked)?)
                } else {
                    None
                }
            } else {
                None
            };
            tree.claim_guard_bytes(key, old.as_deref(), Some(value))?;
            let expires_at = tree.deadline_for_bytes(key, value);
            let existed = tree.put_locked(sid, locked, guard, key, value, expires_at)?;
            let ev = if H::NEEDS_WRITE || (H::WATCHES && crate::watch::context_active()) {
                Some((*key, old, Some(VarHookNew::Borrowed(value))))
            } else {
                None
            };
            *ev_out = ev;
            Ok(existed)
        })
    }

    pub fn insert(&mut self, key: &K, value: &[u8]) -> DbResult<()> {
        self.check_shard(key)?;
        let tree = self.tree;
        let sid = self.held.sid;
        let guard = &self.guard;
        Ctx::with_shard(&mut self.held, tree, sid, |locked, ev_out| {
            // T10 Task 5 review (F5): check presence before the guard, like
            // the facade `VarTree::insert` — see there for the full
            // rationale (`KeyExists` must win over `UniqueIndexRequiresTx`
            // on an occupied key, and a non-claiming insert should not pay
            // this lookup at all).
            if (H::EMITS || (H::CLAIMS && tree.hook.claim_targets().is_some()))
                && tree.index.get(key.as_bytes(), guard).is_none()
            {
                tree.claim_guard_bytes(key, None, Some(value))?;
            }
            let expires_at = tree.deadline_for_bytes(key, value);
            tree.insert_locked(sid, locked, guard, key, value, expires_at)?;
            let ev = if H::NEEDS_WRITE || (H::WATCHES && crate::watch::context_active()) {
                Some((*key, None, Some(VarHookNew::Borrowed(value))))
            } else {
                None
            };
            *ev_out = ev;
            Ok(())
        })
    }

    pub fn delete(&mut self, key: &K) -> DbResult<bool> {
        self.check_shard(key)?;
        let tree = self.tree;
        let sid = self.held.sid;
        let guard = &self.guard;
        Ctx::with_shard(&mut self.held, tree, sid, |locked, ev_out| {
            // `|| H::CLAIMS` (T10 Task 5): see `VarTree::put_with_deadline`.
            let old = if (H::NEEDS_OLD_VALUE
                && (H::NEEDS_WRITE || (H::WATCHES && crate::watch::context_active())))
                || H::CLAIMS
            {
                if let Some(node) = tree.index.get(key.as_bytes(), guard) {
                    let disk = node.read_loc();
                    Some(tree.read_value_locked_result(&disk, sid as u8, locked)?)
                } else {
                    None
                }
            } else {
                None
            };
            // A delete is a pure release, refused outside `db.tx()` exactly
            // like an insert's pure claim.
            tree.claim_guard_bytes(key, old.as_deref(), None)?;
            let existed = tree.delete_locked(sid, locked, guard, key)?;
            let ev =
                if existed && (H::NEEDS_WRITE || (H::WATCHES && crate::watch::context_active())) {
                    Some((*key, old, None))
                } else {
                    None
                };
            *ev_out = ev;
            Ok(existed)
        })
    }

    pub fn try_get(&self, key: &K) -> DbResult<Option<ByteView>> {
        self.check_shard(key)?;
        match self.tree.index.get(key.as_bytes(), &self.guard) {
            Some(node) => {
                let disk = node.read_loc();
                let view = self.tree.read_value_locked_result(
                    &disk,
                    self.held.sid as u8,
                    &self.held.locked,
                )?;
                Ok(Some(view))
            }
            None => Ok(None),
        }
    }

    pub fn get_or_err(&self, key: &K) -> DbResult<ByteView> {
        self.check_shard(key)?;
        let node = self
            .tree
            .index
            .get(key.as_bytes(), &self.guard)
            .ok_or(DbError::KeyNotFound)?;
        let disk = node.read_loc();
        self.tree
            .read_value_locked_result(&disk, self.held.sid as u8, &self.held.locked)
    }

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

    /// Atomically read-modify-write within the shard. Returns `Some(new_value)` if key
    /// existed, `None` if absent. No hook is fired directly — the event is buffered and
    /// replayed by `VarTree::atomic` after the shard lock is released.
    pub fn update(
        &mut self,
        key: &K,
        f: impl FnOnce(&[u8]) -> ByteView,
    ) -> DbResult<Option<ByteView>> {
        self.update_inner(key, |b| Ok(Some(f(b))), 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]) -> ByteView,
    ) -> DbResult<Option<ByteView>> {
        self.update_inner(key, |b| Ok(Some(f(b))), true)
    }

    /// Abort-capable update used by `VarTypedShard`. The closure returns
    /// `Ok(None)` to abort (leave value unchanged), `Ok(Some(v))` to replace.
    pub(crate) fn update_inner(
        &mut self,
        key: &K,
        f: impl FnOnce(&[u8]) -> DbResult<Option<ByteView>>,
        return_old: bool,
    ) -> DbResult<Option<ByteView>> {
        self.check_shard(key)?;
        let tree = self.tree;
        let sid = self.held.sid;
        let guard = &self.guard;
        Ctx::with_shard(&mut self.held, tree, sid, |locked, ev_out| {
            match tree.try_update_locked(sid, locked, guard, key, f)? {
                UpdateOutcome::Missing => Ok(None),
                UpdateOutcome::Unchanged(current) => Ok(Some(current)),
                UpdateOutcome::Updated { old, new } => {
                    // Hoist the `H::NEEDS_WRITE` branch: with hooks off no event is
                    // built, so `new` moves into the result rather than being cloned
                    // under the lock for a value that is immediately dropped.
                    let (result, ev) =
                        if H::NEEDS_WRITE || (H::WATCHES && crate::watch::context_active()) {
                            let event_old = if H::NEEDS_OLD_VALUE {
                                Some(old.clone())
                            } else {
                                None
                            };
                            let result = Some(if return_old { old } else { new.clone() });
                            (
                                result,
                                Some((*key, event_old, Some(VarHookNew::Owned(new)))),
                            )
                        } else {
                            (Some(if return_old { old } else { new }), None)
                        };
                    {
                        *ev_out = ev;
                        Ok(result)
                    }
                }
            }
        })
    }

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

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

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

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

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

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

/// Iterator over entries in a `VarTree`. Returned by `iter()`, `range()`, and `prefix_iter()`.
///
/// Weakly-consistent: concurrent inserts/updates may be visible during iteration.
/// Deleted entries are skipped. The `seize` guard prevents use-after-free.
/// Each `next()` may perform disk I/O on a block-cache miss.
pub struct VarIter<'a, K: Key, H: WriteHook<K> = NoHook> {
    tree: &'a VarTree<K, H>,
    front: *mut VarNode<K>,
    /// `None` = not yet resolved (lazy). Computed on first `next_back()` call.
    back: Option<*mut VarNode<K>>,
    back_finger: BackFinger<VarNode<K>>,
    /// Last key actually yielded by `next()` / `next_back()`. The pointer-based
    /// convergence check cannot fire when the node a cursor is parked on is
    /// concurrently deleted and physically unlinked from the level-0 chain, so
    /// the opposing cursor's progress is tracked by key as well.
    last_front: Option<K>,
    last_back: Option<K>,
    end: Bound<K>,
    start: Bound<K>,
    reversed: bool,
    done: bool,
    _guard: seize::LocalGuard<'a>,
}

impl<K: Key, H: WriteHook<K>> Iterator for VarIter<'_, K, H> {
    type Item = (K, ByteView);

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

impl<K: Key, H: WriteHook<K>> DoubleEndedIterator for VarIter<'_, K, H> {
    fn next_back(&mut self) -> Option<Self::Item> {
        if self.back.is_none() {
            self.back = Some(self.resolve_back());
            if self.front.is_null() {
                self.done = true;
            }
        }
        loop {
            let back = self.back.unwrap_or(std::ptr::null_mut());
            if self.done || back.is_null() {
                return None;
            }
            let node = unsafe { &*back };
            let key = node.key;
            let converged = std::ptr::eq(self.front, back);
            self.back = Some(self.tree.index().back_step(
                &mut self.back_finger,
                key.as_bytes(),
                &self._guard,
            ));
            if converged {
                self.done = true;
            }
            if node.is_marked() {
                if converged {
                    return None;
                }
                continue;
            }
            if self.crossed_front(&key) {
                self.done = true;
                return None;
            }
            if !self.check_start(&key) {
                self.done = true;
                return None;
            }
            match self.tree.read_value_cached(node, &self._guard) {
                Some(value) => {
                    self.last_back = Some(key);
                    return Some((key, value));
                }
                None => {
                    if converged {
                        return None;
                    }
                    continue;
                }
            }
        }
    }
}

impl<K: Key, H: WriteHook<K>> VarIter<'_, K, H> {
    /// Lazily resolve the back pointer for DoubleEndedIterator.
    fn resolve_back(&mut self) -> *mut VarNode<K> {
        let index = self.tree.index();
        match &self.end {
            Bound::Unbounded => index.back_seek_last(&mut self.back_finger, &self._guard),
            Bound::Excluded(k) => {
                index.back_step(&mut self.back_finger, k.as_bytes(), &self._guard)
            }
            Bound::Included(k) => {
                let ge = index.find_first_ge(k.as_bytes(), &self._guard);
                if !ge.is_null()
                    && !unsafe { &*ge }.is_marked()
                    && unsafe { &*ge }.key_bytes() == k.as_bytes()
                {
                    ge
                } else {
                    index.back_step(&mut self.back_finger, k.as_bytes(), &self._guard)
                }
            }
        }
    }

    /// True once the front cursor reaches a key the back cursor has already
    /// delivered. `std::ptr::eq(front, back)` alone cannot detect this: a
    /// concurrent delete physically unlinks the convergence node from the
    /// level-0 chain (`skiplist::remove_with_frame`), so the front walk steps
    /// straight past it and the pointers never match.
    #[inline(always)]
    fn crossed_back(&self, key: &K) -> bool {
        self.last_back.as_ref().is_some_and(|last| {
            if self.reversed {
                key.as_bytes() <= last.as_bytes()
            } else {
                key.as_bytes() >= last.as_bytes()
            }
        })
    }

    /// Mirror of [`crossed_back`](Self::crossed_back) for the back cursor.
    #[inline(always)]
    fn crossed_front(&self, key: &K) -> bool {
        self.last_front.as_ref().is_some_and(|last| {
            if self.reversed {
                key.as_bytes() >= last.as_bytes()
            } else {
                key.as_bytes() <= last.as_bytes()
            }
        })
    }

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

    #[inline(always)]
    fn check_start(&self, key: &K) -> bool {
        match &self.start {
            Bound::Unbounded => true,
            Bound::Excluded(s) => {
                if self.reversed {
                    key.as_bytes() < s.as_bytes()
                } else {
                    key.as_bytes() > s.as_bytes()
                }
            }
            Bound::Included(s) => {
                if self.reversed {
                    key.as_bytes() <= s.as_bytes()
                } else {
                    key.as_bytes() >= s.as_bytes()
                }
            }
        }
    }
    /// Collect only the keys. Convenience for backward compatibility.
    pub fn collect_keys(&mut self) -> Vec<K> {
        self.map(|(k, _)| k).collect()
    }

    /// Collect all remaining entries. Convenience for backward compatibility.
    pub fn collect_entries(&mut self) -> Vec<(K, ByteView)> {
        self.collect()
    }

    /// Strict forward step: advances over marked/deleted nodes; returns the first
    /// real read error instead of skipping it.
    pub(crate) fn try_next_raw(&mut self) -> DbResult<Option<(K, ByteView)>> {
        loop {
            if self.done || self.front.is_null() {
                return Ok(None);
            }
            let node = unsafe { &*self.front };
            let converged = self.back.is_some_and(|back| std::ptr::eq(self.front, back));
            self.front = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
            if converged {
                self.done = true;
            }
            if node.is_marked() {
                if converged {
                    return Ok(None);
                }
                continue;
            }
            if !self.check_end(&node.key) {
                self.done = true;
                return Ok(None);
            }
            return self
                .tree
                .read_value_cached_result(node, &self._guard)
                .map(|value| Some((node.key, value)));
        }
    }

    pub(crate) fn try_next_raw_back(&mut self) -> DbResult<Option<(K, ByteView)>> {
        if self.back.is_none() {
            self.back = Some(self.resolve_back());
            if self.front.is_null() {
                self.done = true;
            }
        }
        loop {
            let back = self.back.unwrap_or(std::ptr::null_mut());
            if self.done || back.is_null() {
                return Ok(None);
            }
            let node = unsafe { &*back };
            let key = node.key;
            let converged = std::ptr::eq(self.front, back);
            self.back = Some(self.tree.index().back_step(
                &mut self.back_finger,
                key.as_bytes(),
                &self._guard,
            ));
            if converged {
                self.done = true;
            }
            if node.is_marked() {
                if converged {
                    return Ok(None);
                }
                continue;
            }
            if !self.check_start(&key) {
                self.done = true;
                return Ok(None);
            }
            return self
                .tree
                .read_value_cached_result(node, &self._guard)
                .map(|value| Some((key, value)));
        }
    }
}

/// Strict iterator over entries in a `VarTree`. Surfaces storage/decode errors
/// instead of skipping corrupt rows.
#[doc(hidden)]
pub struct TryVarIter<'a, K: Key, H: WriteHook<K> = NoHook> {
    inner: VarIter<'a, K, H>,
    failed: bool,
}

impl<K: Key, H: WriteHook<K>> Iterator for TryVarIter<'_, K, H> {
    type Item = DbResult<(K, ByteView)>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.failed {
            return None;
        }
        match self.inner.try_next_raw() {
            Ok(Some(row)) => Some(Ok(row)),
            Ok(None) => None,
            Err(error) => {
                self.failed = true;
                Some(Err(error))
            }
        }
    }
}

impl<K: Key, H: WriteHook<K>> DoubleEndedIterator for TryVarIter<'_, K, H> {
    fn next_back(&mut self) -> Option<Self::Item> {
        if self.failed {
            return None;
        }
        match self.inner.try_next_raw_back() {
            Ok(Some(row)) => Some(Ok(row)),
            Ok(None) => None,
            Err(error) => {
                self.failed = true;
                Some(Err(error))
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::Config;
    use crate::compaction::compact_shard;
    use tempfile::tempdir;

    use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};

    /// Test hook with counters for on_write / on_init calls and the last observed new value for
    /// on_write. NEEDS_INIT and NEEDS_OLD_VALUE are parameterized by const generics.
    #[derive(Default)]
    struct CountingHook<const NEEDS_INIT: bool, const NEEDS_OLD: bool> {
        writes: AtomicUsize,
        writes_with_old: AtomicUsize,
        inits: AtomicUsize,
        last_write_new: crate::sync::Mutex<Option<Vec<u8>>>,
        last_init_value: crate::sync::Mutex<Option<Vec<u8>>>,
    }

    impl<const NEEDS_INIT: bool, const NEEDS_OLD: bool> WriteHook<[u8; 8]>
        for CountingHook<NEEDS_INIT, NEEDS_OLD>
    {
        const NEEDS_OLD_VALUE: bool = NEEDS_OLD;
        const NEEDS_INIT: bool = NEEDS_INIT;

        fn on_write(&self, _key: &[u8; 8], old: Option<&[u8]>, new: Option<&[u8]>) {
            self.writes.fetch_add(1, AtomicOrdering::Relaxed);
            if old.is_some() {
                self.writes_with_old.fetch_add(1, AtomicOrdering::Relaxed);
            }
            *crate::sync::lock(&self.last_write_new) = new.map(<[u8]>::to_vec);
        }

        fn on_init(&self, _key: &[u8; 8], value: &[u8]) {
            self.inits.fetch_add(1, AtomicOrdering::Relaxed);
            *crate::sync::lock(&self.last_init_value) = Some(value.to_vec());
        }
    }

    fn open_test_tree(dir: &std::path::Path) -> VarTree<[u8; 8]> {
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 8192;
        cfg.write_buffer_size = 8192;
        cfg.compaction_threshold = 0.0;
        VarTree::open(dir, cfg).expect("open test tree")
    }

    fn open_test_tree_hooked<const NEEDS_INIT: bool, const NEEDS_OLD: bool>(
        dir: &std::path::Path,
        hook: CountingHook<NEEDS_INIT, NEEDS_OLD>,
    ) -> VarTree<[u8; 8], CountingHook<NEEDS_INIT, NEEDS_OLD>> {
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 8192;
        cfg.write_buffer_size = 8192;
        cfg.compaction_threshold = 0.0;
        VarTree::open_hooked(dir, cfg, hook).expect("open hooked test tree")
    }

    #[test]
    fn strict_range_bounds_dir_surfaces_stale_and_terminates() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());
        let key = 1u64.to_be_bytes();
        tree.put(&key, b"payload").unwrap();

        let shard = &tree.engine.shards()[tree.shard_for(&key)];
        let mut inner = shard.lock();
        inner.immutable = Vec::new();
        inner.active.file_id = u32::MAX;
        drop(inner);

        let mut rows = tree.try_range_bounds_dir(Bound::Unbounded, Bound::Unbounded, Dir::Asc);
        assert!(matches!(rows.next(), Some(Err(DbError::StaleDiskLoc))));
        assert!(
            rows.next().is_none(),
            "strict iterator must terminate after an error"
        );
    }

    #[test]
    fn strict_range_bounds_back_matches_ordinary_valid_data_sequence() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.reversed = false;
        let tree = VarTree::<[u8; 8]>::open(dir.path(), cfg).expect("open test tree");
        for value in [10u64, 20, 30, 40] {
            tree.put(&value.to_be_bytes(), &value.to_be_bytes())
                .expect("seed row");
        }

        let start = 10u64.to_be_bytes();
        let end = 40u64.to_be_bytes();
        let mut ordinary = tree.range_bounds(Bound::Included(&start), Bound::Excluded(&end));
        let ordinary_keys: Vec<_> =
            std::iter::from_fn(|| ordinary.next_back().map(|(key, _)| key)).collect();

        let mut strict = tree.try_range_bounds(Bound::Included(&start), Bound::Excluded(&end));
        let mut strict_keys = Vec::new();
        while let Some((key, _)) = strict.inner.try_next_raw_back().expect("strict row") {
            strict_keys.push(key);
        }

        assert_eq!(
            ordinary_keys,
            vec![
                30u64.to_be_bytes(),
                20u64.to_be_bytes(),
                10u64.to_be_bytes()
            ]
        );
        assert_eq!(strict_keys, ordinary_keys);
    }

    /// Regression (bug 26-06-18, codex coverage note): the reclaim fix must also
    /// hold for `VarTree`, where hints are on by default and values are read back
    /// from disk (not inlined). Under pure overwrite churn, on-disk size must
    /// collapse toward the live set and reads must still return the latest value.
    #[test]
    fn var_compaction_reclaims_dead_bytes_under_overwrite_churn() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 64 * 1024;
        cfg.write_buffer_size = 16 * 1024;
        cfg.compaction_threshold = 0.30;
        let max_file_size = cfg.max_file_size;
        let tree = VarTree::<[u8; 8]>::open(dir.path(), cfg).unwrap();

        const N: u64 = 500;
        const M: u64 = 40;
        let entry_sz: u64 = 16 + 8 + 64;
        let live_set = N * entry_sz;
        let total_written = N * (M + 1) * entry_sz;

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

        let shard = &tree.engine.shards()[0];
        let disk = || {
            let inner = shard.lock();
            inner.active.write_offset + inner.immutable.iter().map(|f| f.total_bytes).sum::<u64>()
        };
        assert!(disk() > total_written / 2, "expected dead-byte buildup");

        for _ in 0..200 {
            let before = disk();
            tree.compact().unwrap();
            if disk() == before {
                break;
            }
        }
        assert!(
            disk() <= live_set + max_file_size,
            "VarTree compaction failed to reclaim: disk={} live_set={live_set}",
            disk()
        );

        // Disk-backed reads must still return the latest value after rewrites.
        let mut v = [0u8; 64];
        v[0] = M as u8;
        assert_eq!(tree.get(&0u64.to_be_bytes()).as_deref(), Some(&v[..]));
        assert_eq!(tree.get(&(N - 1).to_be_bytes()).as_deref(), Some(&v[..]));
        assert_eq!(tree.len(), N as usize);
    }

    fn put_until_compactable(tree: &VarTree<[u8; 8]>, key: [u8; 8]) -> DiskLoc {
        // Capture the DiskLoc after the first put — this points to an early
        // file that will be erased by compaction (all later puts overwrite
        // the index, leaving this entry dead in its file).
        tree.put(&key, &[0u8; 256]).expect("first put");
        let snap = {
            let guard = tree.index.collector().enter();
            let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
            node.read_loc()
        };
        // Each put is 280 bytes (16 header + 8 key + 256 value, 8-byte aligned).
        // With max_file_size=8192, ~30 puts cross the first rotation boundary.
        // Use 65 puts so the snap file is sealed as immutable AND fully dead.
        for i in 1..65u8 {
            tree.put(&key, &[i; 256]).expect("put");
        }
        tree.put(&key, b"final-value-payload-XX")
            .expect("final put");
        snap
    }

    use crate::config::CacheConfig;
    use crate::decoded_cache::DecodedCache;

    fn decoded_cache_on() -> DecodedCache<Vec<u8>> {
        DecodedCache::new(&CacheConfig {
            max_size: 1 << 20,
            estimated_items: 64,
        })
    }

    /// Decode closure that counts how often it actually ran.
    fn counting_decode(counter: &AtomicUsize) -> impl Fn(&ByteView) -> DbResult<Vec<u8>> + '_ {
        move |b| {
            counter.fetch_add(1, AtomicOrdering::Relaxed);
            Ok(b.as_bytes().to_vec())
        }
    }

    #[test]
    fn read_decoded_hit_does_not_decode() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());
        let key = 3u64.to_be_bytes();
        tree.put(&key, b"payload").unwrap();

        let cache = decoded_cache_on();
        let decodes = AtomicUsize::new(0);
        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");

        let first = tree
            .read_decoded(node, &guard, &cache, counting_decode(&decodes))
            .expect("first read");
        let second = tree
            .read_decoded(node, &guard, &cache, counting_decode(&decodes))
            .expect("second read");

        assert_eq!(first.as_slice(), b"payload");
        assert_eq!(second.as_slice(), b"payload");
        assert_eq!(
            decodes.load(AtomicOrdering::Relaxed),
            1,
            "the second read must be served from the decoded cache"
        );
        // Pointer equality here is a *consequence* of controlled residency
        // (1 MiB budget, single thread, no writes), not a general contract —
        // see the spec's "Arc identity is opportunistic".
        assert!(Arc::ptr_eq(&first, &second));
    }

    #[test]
    fn read_decoded_disabled_cache_decodes_every_time() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());
        let key = 4u64.to_be_bytes();
        tree.put(&key, b"payload").unwrap();

        let cache: DecodedCache<Vec<u8>> = DecodedCache::new(&CacheConfig {
            max_size: 0,
            estimated_items: 0,
        });
        let decodes = AtomicUsize::new(0);
        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");

        for _ in 0..3 {
            let v = tree
                .read_decoded(node, &guard, &cache, counting_decode(&decodes))
                .expect("read");
            assert_eq!(v.as_slice(), b"payload");
        }
        assert_eq!(decodes.load(AtomicOrdering::Relaxed), 3);
    }

    #[test]
    fn read_decoded_sees_a_write_and_decodes_again() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());
        let key = 5u64.to_be_bytes();
        tree.put(&key, b"v1").unwrap();

        let cache = decoded_cache_on();
        let decodes = AtomicUsize::new(0);

        let held = {
            let guard = tree.index.collector().enter();
            let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
            tree.read_decoded(node, &guard, &cache, counting_decode(&decodes))
                .expect("first read")
        };
        assert_eq!(held.as_slice(), b"v1");

        tree.put(&key, b"v2").unwrap();

        let after = {
            let guard = tree.index.collector().enter();
            let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
            tree.read_decoded(node, &guard, &cache, counting_decode(&decodes))
                .expect("second read")
        };
        assert_eq!(after.as_slice(), b"v2", "a write must not be served stale");
        assert_eq!(decodes.load(AtomicOrdering::Relaxed), 2);
        // The handle taken before the write still observes the old version.
        assert_eq!(held.as_slice(), b"v1");
    }

    #[test]
    fn read_decoded_correct_after_compaction() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());
        let key = 11u64.to_be_bytes();
        let _snap = put_until_compactable(&tree, key);
        // Seal the active file so the live entry can be relocated by compaction.
        tree.engine.shards()[0]
            .rotate_active_for_test(8)
            .expect("seal active file");

        let cache = decoded_cache_on();
        let decodes = AtomicUsize::new(0);

        // Populate the decoded cache at the pre-compaction location.
        {
            let guard = tree.index.collector().enter();
            let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
            tree.read_decoded(node, &guard, &cache, counting_decode(&decodes))
                .expect("pre-compaction read");
        }

        let shard = &tree.engine.shards()[0];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        let v = tree
            .read_decoded(node, &guard, &cache, counting_decode(&decodes))
            .expect("post-compaction read");
        assert_eq!(v.as_slice(), b"final-value-payload-XX");
        assert_eq!(
            decodes.load(AtomicOrdering::Relaxed),
            2,
            "compaction moves the value, so the post-compaction read is a cold miss"
        );
    }

    #[test]
    fn read_decoded_covers_both_byte_routes() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 128 * 1024;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        let tree = VarTree::open(dir.path(), cfg).expect("open test tree");
        let cache = decoded_cache_on();
        let decodes = AtomicUsize::new(0);

        // Small value: served through the block cache.
        let small_key = 21u64.to_be_bytes();
        tree.put(&small_key, &[7u8; 64]).unwrap();
        // Large value: `(offset & 4095) + len > 8192`, i.e. more than two blocks.
        let big_key = 22u64.to_be_bytes();
        tree.put(&big_key, &vec![9u8; 12 * 1024]).unwrap();

        let guard = tree.index.collector().enter();

        let small_node = tree
            .index
            .get(small_key.as_bytes(), &guard)
            .expect("indexed");
        assert!(
            !small_node.read_loc().is_value_cache_routed(),
            "test setup: the small value must take the block-cache route"
        );
        let small = tree
            .read_decoded(small_node, &guard, &cache, counting_decode(&decodes))
            .expect("small read");
        assert_eq!(small.as_slice(), &vec![7u8; 64][..]);

        let big_node = tree.index.get(big_key.as_bytes(), &guard).expect("indexed");
        assert!(
            big_node.read_loc().is_value_cache_routed(),
            "test setup: the large value must take the value-cache route"
        );
        let big = tree
            .read_decoded(big_node, &guard, &cache, counting_decode(&decodes))
            .expect("big read");
        assert_eq!(big.as_slice(), &vec![9u8; 12 * 1024][..]);

        // Both are now cached: a second read of each decodes nothing more.
        let before = decodes.load(AtomicOrdering::Relaxed);
        tree.read_decoded(small_node, &guard, &cache, counting_decode(&decodes))
            .expect("small re-read");
        tree.read_decoded(big_node, &guard, &cache, counting_decode(&decodes))
            .expect("big re-read");
        assert_eq!(decodes.load(AtomicOrdering::Relaxed), before);
    }

    #[test]
    fn try_get_decoded_absent_key_is_none() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());
        let cache = decoded_cache_on();
        let decodes = AtomicUsize::new(0);

        let missing = tree
            .try_get_decoded(&99u64.to_be_bytes(), &cache, counting_decode(&decodes))
            .expect("absent key is not an error");
        assert!(missing.is_none());
        assert_eq!(decodes.load(AtomicOrdering::Relaxed), 0);
    }

    #[test]
    fn try_get_decoded_absent_after_delete_redecodes_on_reinsert() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());
        let key = 41u64.to_be_bytes();
        tree.put(&key, b"v1").unwrap();

        let cache = decoded_cache_on();
        let decodes = AtomicUsize::new(0);

        let v1 = tree
            .try_get_decoded(&key, &cache, counting_decode(&decodes))
            .unwrap()
            .unwrap();
        assert_eq!(v1.as_slice(), b"v1");

        assert!(tree.delete(&key).unwrap());
        assert!(
            tree.try_get_decoded(&key, &cache, counting_decode(&decodes))
                .unwrap()
                .is_none()
        );

        tree.put(&key, b"v2").unwrap();
        let before = decodes.load(AtomicOrdering::Relaxed);
        let v2 = tree
            .try_get_decoded(&key, &cache, counting_decode(&decodes))
            .unwrap()
            .unwrap();
        assert_eq!(v2.as_slice(), b"v2");
        assert_eq!(
            decodes.load(AtomicOrdering::Relaxed) - before,
            1,
            "re-insert must decode again"
        );
    }

    #[test]
    fn read_decoded_returns_stale_after_retry_limit() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 51u64.to_be_bytes();
        tree.put(&key, b"payload").expect("put");

        let guard = tree.index.collector().enter();
        assert!(tree.index.get(key.as_bytes(), &guard).is_some());
        drop(guard);

        {
            let shard_id = tree.shard_for(&key) as u8;
            let shard = &tree.engine.shards()[shard_id as usize];
            let mut inner = shard.lock();
            inner.immutable = Vec::new();
            inner.active.file_id = u32::MAX;
        }

        let cache = decoded_cache_on();
        let decodes = AtomicUsize::new(0);
        let guard = tree.index.collector().enter();
        let node = tree
            .index
            .get(key.as_bytes(), &guard)
            .expect("still indexed");
        match tree.read_decoded(node, &guard, &cache, counting_decode(&decodes)) {
            Err(DbError::StaleDiskLoc) => {}
            Ok(v) => panic!("expected StaleDiskLoc, got Ok({:?})", v.as_slice()),
            Err(e) => panic!("expected StaleDiskLoc, got Err({e})"),
        }
    }

    #[test]
    fn try_get_decoded_propagates_decode_errors_and_caches_nothing() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());
        let key = 31u64.to_be_bytes();
        tree.put(&key, b"whatever").unwrap();

        let cache = decoded_cache_on();
        let attempts = AtomicUsize::new(0);
        let failing = |_b: &ByteView| -> DbResult<Vec<u8>> {
            attempts.fetch_add(1, AtomicOrdering::Relaxed);
            Err(DbError::CorruptedEntry { offset: 0 })
        };

        for _ in 0..2 {
            assert!(matches!(
                tree.try_get_decoded(&key, &cache, failing),
                Err(DbError::CorruptedEntry { .. })
            ));
        }
        assert_eq!(
            attempts.load(AtomicOrdering::Relaxed),
            2,
            "a failed decode must not be cached"
        );
    }

    // ---------------------------------------------------------------------------
    // Tests — get_many_decoded (Task 5)
    // ---------------------------------------------------------------------------

    #[test]
    fn get_many_decoded_is_aligned_and_lossy() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());
        for i in [1u64, 2, 3] {
            tree.put(&i.to_be_bytes(), format!("v{i}").as_bytes())
                .unwrap();
        }

        let cache = decoded_cache_on();
        let decodes = AtomicUsize::new(0);
        // Deliberately unsorted, with an absent key in the middle.
        let keys = [
            3u64.to_be_bytes(),
            99u64.to_be_bytes(),
            1u64.to_be_bytes(),
            2u64.to_be_bytes(),
        ];

        let out = tree.get_many_decoded(&keys, &cache, counting_decode(&decodes));
        assert_eq!(out.len(), 4);
        assert_eq!(out[0].as_deref().map(Vec::as_slice), Some(&b"v3"[..]));
        assert!(out[1].is_none(), "absent key must be None at its own index");
        assert_eq!(out[2].as_deref().map(Vec::as_slice), Some(&b"v1"[..]));
        assert_eq!(out[3].as_deref().map(Vec::as_slice), Some(&b"v2"[..]));
        assert_eq!(decodes.load(AtomicOrdering::Relaxed), 3);
    }

    #[test]
    fn get_many_decoded_second_pass_hits_the_cache() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());
        for i in 0..8u64 {
            tree.put(&i.to_be_bytes(), format!("v{i}").as_bytes())
                .unwrap();
        }
        let cache = decoded_cache_on();
        let decodes = AtomicUsize::new(0);
        let keys: Vec<[u8; 8]> = (0..8u64).map(u64::to_be_bytes).collect();

        let first = tree.get_many_decoded(&keys, &cache, counting_decode(&decodes));
        assert_eq!(first.iter().filter(|v| v.is_some()).count(), 8);
        assert_eq!(decodes.load(AtomicOrdering::Relaxed), 8);

        let second = tree.get_many_decoded(&keys, &cache, counting_decode(&decodes));
        assert_eq!(second.iter().filter(|v| v.is_some()).count(), 8);
        assert_eq!(
            decodes.load(AtomicOrdering::Relaxed),
            8,
            "a fully cached batch must not decode again"
        );
    }

    #[test]
    fn get_many_decoded_empty_input() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());
        let cache = decoded_cache_on();
        let decodes = AtomicUsize::new(0);
        let out = tree.get_many_decoded(&[], &cache, counting_decode(&decodes));
        assert!(out.is_empty());
    }

    #[test]
    fn get_many_decoded_decode_error_is_lossy_and_does_not_cache_failure() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());
        let good_key = 1u64.to_be_bytes();
        let bad_key = 2u64.to_be_bytes();
        tree.put(&good_key, b"ok").unwrap();
        tree.put(&bad_key, b"bad-bytes").unwrap();

        let cache = decoded_cache_on();
        let attempts = AtomicUsize::new(0);
        let decode = |b: &ByteView| -> DbResult<Vec<u8>> {
            if b.as_bytes() == b"bad-bytes" {
                attempts.fetch_add(1, AtomicOrdering::Relaxed);
                return Err(DbError::CorruptedEntry { offset: 0 });
            }
            Ok(b.as_bytes().to_vec())
        };

        let keys = [bad_key, good_key, 99u64.to_be_bytes()];
        let out = tree.get_many_decoded(&keys, &cache, decode);
        assert_eq!(out.len(), 3);
        assert!(out[0].is_none(), "decode error must map to None");
        assert_eq!(out[1].as_deref().map(Vec::as_slice), Some(&b"ok"[..]));
        assert!(out[2].is_none(), "absent key must be None");
        assert_eq!(attempts.load(AtomicOrdering::Relaxed), 1);

        let before = attempts.load(AtomicOrdering::Relaxed);
        let again = tree.get_many_decoded(&keys, &cache, decode);
        assert_eq!(again[1].as_deref().map(Vec::as_slice), Some(&b"ok"[..]));
        assert!(again[0].is_none());
        assert_eq!(
            attempts.load(AtomicOrdering::Relaxed) - before,
            1,
            "a failed decode must not be cached; the bad key retries on the next batch"
        );
    }

    #[test]
    fn put_reports_whether_key_existed() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());
        let key = 42u64.to_be_bytes();

        // Fresh insert -> false.
        assert!(!tree.put(&key, b"v1").expect("put"));
        // Overwrite of the same key -> true.
        assert!(tree.put(&key, b"v2").expect("put"));
        assert!(tree.put(&key, b"v3").expect("put"));
        // Survives a flush (overwrite still detected after entries hit disk).
        tree.flush_buffers().expect("flush");
        assert!(tree.put(&key, b"v4").expect("put"));
        // A different key is a fresh insert again.
        assert!(!tree.put(&7u64.to_be_bytes(), b"x").expect("put"));
    }

    #[test]
    fn warmup_threads_correct_shard_across_shards() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());
        // 256 keys spread across both shards
        for i in 0..256u64 {
            tree.put(&i.to_be_bytes(), &[i as u8; 64]).expect("put");
        }
        tree.flush_buffers().expect("flush");
        // warmup threads shard from node.key across both shards
        tree.warmup().expect("warmup");
        // values read correctly after warmup
        for i in 0..256u64 {
            assert_eq!(
                tree.get(&i.to_be_bytes()).as_deref(),
                Some(&[i as u8; 64][..])
            );
        }
    }

    #[test]
    fn read_value_cached_inner_returns_stale_after_compaction() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 7u64.to_be_bytes();
        let snap = put_until_compactable(&tree, key);
        let shard_id = tree.shard_for(&key) as u8;

        let shard = &tree.engine.shards()[shard_id as usize];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        match tree.read_value_cached_inner(&snap, shard_id) {
            Err(DbError::StaleDiskLoc) => {}
            Ok(v) => panic!("expected StaleDiskLoc, got Ok({:?})", v.as_bytes()),
            Err(e) => panic!("expected StaleDiskLoc, got Err({e})"),
        }
    }

    #[test]
    fn read_value_cached_returns_some_after_compaction() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 11u64.to_be_bytes();
        let _snap = put_until_compactable(&tree, key);
        let shard = &tree.engine.shards()[0];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        let v = tree
            .read_value_cached(node, &guard)
            .expect("post-compaction value must be readable");
        assert_eq!(v.as_bytes(), b"final-value-payload-XX");
    }

    #[test]
    fn get_during_compaction_returns_some() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 13u64.to_be_bytes();
        let _snap = put_until_compactable(&tree, key);
        let shard = &tree.engine.shards()[0];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        let v = tree.get(&key).expect("post-compaction get");
        assert_eq!(v.as_bytes(), b"final-value-payload-XX");
    }

    #[test]
    fn iter_during_compaction_yields_all_live_keys() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        for k in 1u64..=3 {
            put_until_compactable(&tree, k.to_be_bytes());
        }
        let shard = &tree.engine.shards()[0];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        let collected: std::collections::BTreeMap<[u8; 8], Vec<u8>> = tree
            .iter()
            .map(|(k, v)| (k, v.as_bytes().to_vec()))
            .collect();
        assert_eq!(collected.len(), 3);
        for k in 1u64..=3 {
            let bytes = collected
                .get(&k.to_be_bytes())
                .expect("every original key must remain");
            assert_eq!(bytes.as_slice(), b"final-value-payload-XX");
        }
    }

    #[test]
    fn get_or_read_block_returns_stale_for_unknown_file_id() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        match tree.get_or_read_block(0, 9999, 0) {
            Err(DbError::StaleDiskLoc) => {}
            Ok(_) => panic!("expected StaleDiskLoc, got Ok"),
            Err(e) => panic!("expected StaleDiskLoc, got Err({e})"),
        }
    }

    #[test]
    fn extract_from_block_propagates_next_block_error() {
        let block = AlignedBuf::zeroed(4096);
        let start = 4090;
        let len = 32;
        let result: DbResult<ByteView> =
            VarTree::<[u8; 8]>::extract_from_block(&block, start, len, || {
                Err(DbError::StaleDiskLoc)
            });
        match result {
            Err(DbError::StaleDiskLoc) => {}
            Ok(_) => panic!("expected StaleDiskLoc, got Ok"),
            Err(e) => panic!("expected StaleDiskLoc, got Err({e})"),
        }
    }

    #[test]
    fn extract_from_block_multi_block_first_cached_second_stale() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 21u64.to_be_bytes();
        let value = vec![0xCDu8; 4073];
        tree.put(&key, &value).expect("initial put");

        // Capture the multi-block DiskLoc BEFORE overwriting.
        let snap = {
            let guard = tree.index.collector().enter();
            let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
            node.read_loc()
        };

        // Make the initial entry dead by overwriting many times so compaction
        // erases the file containing `snap` (need enough writes to force file
        // rotation past max_file_size=8192 and seal the file as immutable).
        // 65 overwrites ensures the snap file is fully dead after two rotations.
        for i in 0..65u8 {
            tree.put(&key, &[i; 256]).expect("overwrite");
        }
        tree.put(&key, b"final").expect("final");

        let shard = &tree.engine.shards()[0];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        let block_offset = snap.offset as u64 & !4095;
        let shard_id = tree.shard_for(&key) as u8;
        let first_block = AlignedBuf::zeroed(4096);
        let cache_key = BlockKey {
            shard_id,
            file_id: snap.file_id,
            block_offset,
        };
        tree.block_cache.insert(cache_key, Arc::new(first_block));

        match tree.read_value_cached_inner(&snap, shard_id) {
            Err(DbError::StaleDiskLoc) => {}
            Ok(_) => panic!("expected StaleDiskLoc from second block, got Ok"),
            Err(e) => panic!("expected StaleDiskLoc, got Err({e})"),
        }
    }

    #[test]
    fn retry_limit_returns_none_on_persistent_stale() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 99u64.to_be_bytes();
        tree.put(&key, b"payload").expect("put");

        // Force every read_value_cached_inner call to see Stale by manually
        // clearing inner.immutable AND moving the active.file_id out of the
        // way. node.read_loc() still points at the original (now-unreachable)
        // file_id, so each retry hits Shard::read_block, which returns
        // StaleDiskLoc.
        let guard = tree.index.collector().enter();
        assert!(tree.index.get(key.as_bytes(), &guard).is_some());
        drop(guard);

        {
            let shard_id = tree.shard_for(&key) as u8;
            let shard = &tree.engine.shards()[shard_id as usize];
            let mut inner = shard.lock();
            inner.immutable = Vec::new();
            // Reassign active.file_id to a sentinel value that does not match
            // snap.file_id, so the active-file branch in Shard::read_block
            // never matches either.
            inner.active.file_id = u32::MAX;
        }

        let guard = tree.index.collector().enter();
        let node = tree
            .index
            .get(key.as_bytes(), &guard)
            .expect("still indexed");
        assert!(
            tree.read_value_cached(node, &guard).is_none(),
            "MAX_STALE_RETRIES must terminate the loop and return None"
        );
    }

    #[test]
    fn var_tree_replay_init_fires_on_init_per_live_key_raw() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<true, false>(dir.path(), CountingHook::default());

        for i in 0u64..5 {
            tree.put(&i.to_be_bytes(), &[i as u8; 16]).expect("put");
        }
        // Delete one — replay_init must not see it.
        tree.delete(&3u64.to_be_bytes()).expect("delete");

        // Reset counters so we measure only the effect of replay_init.
        tree.hook.writes.store(0, AtomicOrdering::Relaxed);
        tree.hook.inits.store(0, AtomicOrdering::Relaxed);

        tree.replay_init();

        assert_eq!(
            tree.hook.inits.load(AtomicOrdering::Relaxed),
            4,
            "4 live keys"
        );
        assert_eq!(
            tree.hook.writes.load(AtomicOrdering::Relaxed),
            0,
            "no on_write"
        );
    }

    #[test]
    fn var_tree_replay_init_no_hook_is_noop() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        for i in 0u64..3 {
            tree.put(&i.to_be_bytes(), &[i as u8; 8]).expect("put");
        }
        // Must finish instantly with no effects.
        tree.replay_init();
        // sanity: tree is intact.
        assert!(tree.get(&0u64.to_be_bytes()).is_some());
    }

    #[test]
    fn var_tree_migrate_keep_fires_on_init_not_on_write_raw() {
        use crate::MigrateAction;
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<true, false>(dir.path(), CountingHook::default());

        for i in 0u64..4 {
            tree.put(&i.to_be_bytes(), &[i as u8; 16]).expect("put");
        }
        tree.hook.writes.store(0, AtomicOrdering::Relaxed);
        tree.hook.inits.store(0, AtomicOrdering::Relaxed);

        let mutated = tree.migrate(|_, _| MigrateAction::Keep).expect("migrate");

        assert_eq!(mutated, 0);
        assert_eq!(
            tree.hook.inits.load(AtomicOrdering::Relaxed),
            4,
            "4 keeps -> 4 on_init"
        );
        assert_eq!(
            tree.hook.writes.load(AtomicOrdering::Relaxed),
            0,
            "Keep must not fire on_write"
        );
    }

    #[test]
    fn var_tree_migrate_update_fires_on_init_with_new_value_raw() {
        use crate::MigrateAction;
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<true, false>(dir.path(), CountingHook::default());

        let key = 42u64.to_be_bytes();
        tree.put(&key, b"old-value").expect("put");
        tree.hook.writes.store(0, AtomicOrdering::Relaxed);
        tree.hook.inits.store(0, AtomicOrdering::Relaxed);

        let new = ByteView::new(b"new-value");
        let mutated = tree
            .migrate(move |_, _| MigrateAction::Update(new.clone()))
            .expect("migrate");

        assert_eq!(mutated, 1);
        assert_eq!(tree.hook.inits.load(AtomicOrdering::Relaxed), 1);
        assert_eq!(
            crate::sync::lock(&tree.hook.last_init_value).as_deref(),
            Some(b"new-value".as_ref()),
            "on_init must receive the NEW value"
        );
        assert_eq!(
            tree.hook.writes.load(AtomicOrdering::Relaxed),
            0,
            "Update must NOT fire on_write (was double-firing through self.put)"
        );
        assert_eq!(tree.get(&key).unwrap().as_bytes(), b"new-value");
    }

    #[test]
    fn var_tree_migrate_delete_fires_no_hooks_raw() {
        use crate::MigrateAction;
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<true, false>(dir.path(), CountingHook::default());

        let key = 7u64.to_be_bytes();
        tree.put(&key, b"x").expect("put");
        tree.hook.writes.store(0, AtomicOrdering::Relaxed);
        tree.hook.inits.store(0, AtomicOrdering::Relaxed);

        let mutated = tree.migrate(|_, _| MigrateAction::Delete).expect("migrate");

        assert_eq!(mutated, 1);
        assert_eq!(tree.hook.inits.load(AtomicOrdering::Relaxed), 0);
        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 0);
        assert!(tree.get(&key).is_none());
    }

    #[test]
    fn var_tree_migrate_no_init_hook_is_silent_for_keep_and_update() {
        use crate::MigrateAction;
        let dir = tempdir().unwrap();
        // NEEDS_INIT = false: on_init must never be called regardless of action.
        let tree = open_test_tree_hooked::<false, false>(dir.path(), CountingHook::default());

        for i in 0u64..3 {
            tree.put(&i.to_be_bytes(), &[i as u8; 16]).expect("put");
        }
        tree.hook.writes.store(0, AtomicOrdering::Relaxed);
        tree.hook.inits.store(0, AtomicOrdering::Relaxed);

        // Keep
        tree.migrate(|_, _| MigrateAction::Keep)
            .expect("migrate keep");
        assert_eq!(
            tree.hook.inits.load(AtomicOrdering::Relaxed),
            0,
            "Keep with NEEDS_INIT=false"
        );
        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 0);

        // Update
        let new = ByteView::new(b"new");
        tree.migrate(move |_, _| MigrateAction::Update(new.clone()))
            .expect("migrate update");
        assert_eq!(
            tree.hook.inits.load(AtomicOrdering::Relaxed),
            0,
            "Update with NEEDS_INIT=false"
        );
        assert_eq!(
            tree.hook.writes.load(AtomicOrdering::Relaxed),
            0,
            "Update must not fire on_write either"
        );
    }

    #[test]
    fn var_tree_public_put_still_fires_on_write_once() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<true, false>(dir.path(), CountingHook::default());

        tree.put(&1u64.to_be_bytes(), b"v").expect("put");
        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 1);
    }

    #[test]
    fn var_tree_atomic_fires_hooks() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<true, true>(dir.path(), CountingHook::default());
        let key = 1u64.to_be_bytes();
        tree.atomic(&key, |shard| {
            shard.put(&key, b"a")?; // new key
            shard.delete(&key)?; // delete, old=b"a"
            Ok(())
        })
        .expect("atomic");
        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 2);
        assert_eq!(tree.hook.writes_with_old.load(AtomicOrdering::Relaxed), 1); // delete saw old
    }

    #[test]
    fn var_tree_atomic_fires_for_applied_on_err() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<true, true>(dir.path(), CountingHook::default());
        let key = 1u64.to_be_bytes();
        let r: DbResult<()> = tree.atomic(&key, |shard| {
            shard.put(&key, b"x")?;
            Err(DbError::KeyNotFound)
        });
        assert!(r.is_err());
        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 1);
    }

    #[test]
    fn var_tree_atomic_same_key_old_from_disk() {
        // NEEDS_OLD_VALUE=true: the old value of the 2nd put must be the 1st put's value,
        // read back through the write buffer under the lock.
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<false, true>(dir.path(), CountingHook::default());
        let key = 1u64.to_be_bytes();
        tree.atomic(&key, |shard| {
            shard.put(&key, b"first")?;
            shard.put(&key, b"second")?;
            Ok(())
        })
        .expect("atomic");
        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 2);
        assert_eq!(tree.hook.writes_with_old.load(AtomicOrdering::Relaxed), 1); // 2nd put saw old
        assert_eq!(
            crate::sync::lock(&tree.hook.last_write_new).clone(),
            Some(b"second".to_vec())
        );
    }

    #[test]
    fn var_tree_shard_put_returns_existed() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<true, true>(dir.path(), CountingHook::default());
        let key = 1u64.to_be_bytes();
        let (fresh, overwrite) = tree
            .atomic(&key, |shard| {
                let fresh = shard.put(&key, b"a")?; // fresh insert -> false
                let overwrite = shard.put(&key, b"b")?; // overwrite -> true
                Ok((fresh, overwrite))
            })
            .expect("atomic");
        assert!(!fresh, "fresh insert must return false");
        assert!(overwrite, "overwrite must return true");
    }

    #[test]
    fn compare_delete_match_mismatch_absent() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let k = [1u8; 8];
        tree.put(&k, b"hello").unwrap();

        assert!(matches!(
            tree.compare_delete(&k, b"WRONG"),
            Err(DbError::CasMismatch)
        ));
        assert_eq!(tree.get(&k).as_deref(), Some(&b"hello"[..]));

        assert!(tree.compare_delete(&k, b"hello").is_ok());
        assert!(tree.get(&k).is_none());

        assert!(matches!(
            tree.compare_delete(&k, b"hello"),
            Err(DbError::KeyNotFound)
        ));
    }

    #[test]
    fn compare_delete_hook_needs_old_true_gets_old_value() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<false, true>(dir.path(), CountingHook::default());

        let k = [2u8; 8];
        tree.put(&k, b"payload").unwrap();
        let with_old_before = tree.hook.writes_with_old.load(AtomicOrdering::Relaxed);

        tree.compare_delete(&k, b"payload").unwrap();

        // NEEDS_OLD_VALUE = true → on_write received Some(old)
        assert_eq!(
            tree.hook.writes_with_old.load(AtomicOrdering::Relaxed) - with_old_before,
            1
        );
        assert_eq!(*crate::sync::lock(&tree.hook.last_write_new), None);
    }

    #[test]
    fn compare_delete_hook_needs_old_false_gets_none() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<false, false>(dir.path(), CountingHook::default());

        let k = [3u8; 8];
        tree.put(&k, b"payload").unwrap();
        let writes_before = tree.hook.writes.load(AtomicOrdering::Relaxed);
        let with_old_before = tree.hook.writes_with_old.load(AtomicOrdering::Relaxed);

        tree.compare_delete(&k, b"payload").unwrap();

        // exactly one on_write, but old = None (NEEDS_OLD_VALUE = false)
        assert_eq!(
            tree.hook.writes.load(AtomicOrdering::Relaxed) - writes_before,
            1
        );
        assert_eq!(
            tree.hook.writes_with_old.load(AtomicOrdering::Relaxed) - with_old_before,
            0
        );
    }

    /// `_result` returns Ok when the value sits in the active write buffer.
    #[test]
    fn read_value_locked_result_ok_from_write_buf() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 1u64.to_be_bytes();
        let payload = b"in-write-buffer-value";
        tree.put(&key, payload).expect("put");

        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        let disk = node.read_loc();
        drop(guard);

        let shard_id = tree.shard_for(&key) as u8;
        let shard = &tree.engine.shards()[shard_id as usize];
        let inner = shard.lock();
        let v = tree
            .read_value_locked_result(&disk, shard_id, &inner)
            .expect("write-buf read must succeed");
        assert_eq!(v.as_bytes(), payload);
    }

    /// `_result` returns Ok when the value is on an immutable file and the
    /// entry sits within a single 4 KiB block.
    #[test]
    fn read_value_locked_result_ok_from_disk_immutable() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 2u64.to_be_bytes();
        let payload = b"single-block-immutable";
        tree.put(&key, payload).expect("put");
        // Force rotation so the entry moves out of the active write buffer.
        // Each put is 280 bytes; with max_file_size=8192, need >30 puts to
        // cross the rotation boundary. Use keys 100..135 to avoid overwriting key 2.
        for i in 100u64..135 {
            tree.put(&i.to_be_bytes(), &[i as u8; 256])
                .expect("rotator");
        }

        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        let disk = node.read_loc();
        drop(guard);

        // `open_test_tree` uses the default CacheConfig (max_size = 0), so the
        // cache is disabled and step 2 (single-block cache lookup) returns
        // None unconditionally — the read naturally exercises step 3 (disk).

        let shard_id = tree.shard_for(&key) as u8;
        let shard = &tree.engine.shards()[shard_id as usize];
        let inner = shard.lock();
        // Sanity: the entry must NOT be in the write buffer anymore — the rotator
        // loop above is sized to force rotation past max_file_size=8192.
        assert_ne!(
            disk.file_id, inner.active.file_id,
            "test setup failed: key=2 entry is still in the active file's write buffer",
        );
        let v = tree
            .read_value_locked_result(&disk, shard_id, &inner)
            .expect("disk read must succeed");
        assert_eq!(v.as_bytes(), payload);
    }

    /// `_result` propagates `StaleDiskLoc` from the disk read.
    #[test]
    fn read_value_locked_result_propagates_stale_disk_loc() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 3u64.to_be_bytes();
        let snap = put_until_compactable(&tree, key);

        let shard_id = tree.shard_for(&key) as u8;
        let shard = &tree.engine.shards()[shard_id as usize];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        let inner = shard.lock();
        match tree.read_value_locked_result(&snap, shard_id, &inner) {
            Err(DbError::StaleDiskLoc) => {}
            Ok(v) => panic!("expected StaleDiskLoc, got Ok({:?})", v.as_bytes()),
            Err(e) => panic!("expected StaleDiskLoc, got Err({e})"),
        }
    }

    /// Pins the "size check sits before cache lookup" invariant.
    ///
    /// Manually inject the first 4 KiB block of a large value into the
    /// cache. Without the early `start + len > 8192` check, the cached path
    /// would call `extract_from_block`, which only supports two blocks and
    /// would panic on `next[..second_len]` for `second_len > 4096`.
    ///
    /// With the early check, the call routes to the locked path and returns
    /// the correct bytes.
    #[test]
    fn large_value_with_first_block_cached_uses_fallback() {
        let dir = tempdir().unwrap();
        // Custom geometry: write_buffer_size big enough for 20 KiB entry,
        // max_file_size == write_buffer_size (required by Config::validate);
        // rotation is forced explicitly via rotate_active_for_test.
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 128 * 1024;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        // Enable the block cache so `cache.insert` actually takes effect.
        // Default `CacheConfig::default` has `max_size = 0` which makes the
        // cache a no-op; we need real inserts to set up the warm-cache scenario.
        cfg.block_cache.max_size = 1 << 20;
        let tree: VarTree<[u8; 8]> = VarTree::open(dir.path(), cfg).expect("open");

        let key = 42u64.to_be_bytes();
        // Build a value whose body spans ~5 blocks: > 8192 bytes.
        let payload: Vec<u8> = (0..20_000u32).map(|i| i as u8).collect();
        tree.put(&key, &payload).expect("put large");
        // Force rotation so the large entry moves to an immutable file.
        tree.engine.shards()[0]
            .rotate_active_for_test(8)
            .expect("rotate");

        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        let disk = node.read_loc();
        drop(guard);

        // Sanity: this is a multi-block large value.
        let start = (disk.offset & 4095) as usize;
        assert!(
            start + disk.len as usize > 8192,
            "test precondition: large value must span >2 blocks",
        );

        // Inject the first block (zeroes) into the cache to ensure the cache
        // lookup would succeed for the first block — exactly the warm-cache
        // scenario that would have triggered the old panic.
        let block_offset = disk.offset as u64 & !4095;
        let shard_id = tree.shard_for(&key) as u8;
        let cache_key = BlockKey {
            shard_id,
            file_id: disk.file_id,
            block_offset,
        };
        tree.block_cache
            .insert(cache_key, Arc::new(AlignedBuf::zeroed(4096)));
        // Sanity: insert took effect.
        assert!(
            tree.block_cache.get(&cache_key).is_some(),
            "cache must be enabled for warm-cache scenario",
        );

        let v = tree
            .read_value_cached_inner(&disk, shard_id)
            .expect("large value must read via locked fallback");
        assert_eq!(v.as_bytes(), payload.as_slice());
    }

    /// Value entirely in one block: `start + len <= 4096`.
    #[test]
    fn extract_from_block_single_block() {
        let mut block = AlignedBuf::zeroed(4096);
        for (i, byte) in block.iter_mut().enumerate() {
            *byte = i as u8;
        }
        let v = VarTree::<[u8; 8]>::extract_from_block(&block, 100, 50, || {
            panic!("next_block must not be called for single-block reads")
        })
        .expect("ok");
        let expected: Vec<u8> = (100u8..150u8).collect();
        assert_eq!(v.as_bytes(), expected.as_slice());
    }

    /// Value ends exactly at the second block's end: `start + len == 8192`.
    #[test]
    fn extract_from_block_two_blocks_exact() {
        let mut first = AlignedBuf::zeroed(4096);
        for byte in first.iter_mut() {
            *byte = 0xAA;
        }
        let mut second = AlignedBuf::zeroed(4096);
        for byte in second.iter_mut() {
            *byte = 0xBB;
        }
        // start = 4096 - 1 (last byte of first block); len = 4097.
        // start + len = 8192 — the boundary that must remain supported.
        let v = VarTree::<[u8; 8]>::extract_from_block(&first, 4095, 4097, || Ok(Arc::new(second)))
            .expect("ok");
        let bytes = v.as_bytes();
        assert_eq!(bytes.len(), 4097);
        assert_eq!(bytes[0], 0xAA);
        assert_eq!(bytes[1], 0xBB);
        assert_eq!(bytes[4096], 0xBB);
    }

    /// Value spans two blocks partially: `4096 < start + len < 8192`.
    #[test]
    fn extract_from_block_two_blocks_partial() {
        let mut first = AlignedBuf::zeroed(4096);
        for byte in first.iter_mut() {
            *byte = 0x11;
        }
        let mut second = AlignedBuf::zeroed(4096);
        for byte in second.iter_mut() {
            *byte = 0x22;
        }
        // start = 4000, len = 200 -> first 96 bytes from first, next 104 from second.
        let v = VarTree::<[u8; 8]>::extract_from_block(&first, 4000, 200, || Ok(Arc::new(second)))
            .expect("ok");
        let bytes = v.as_bytes();
        assert_eq!(bytes.len(), 200);
        assert!(bytes[..96].iter().all(|b| *b == 0x11));
        assert!(bytes[96..].iter().all(|b| *b == 0x22));
    }

    /// Fixed geometry that forces large entries onto an immutable file:
    /// `write_buffer_size` fits the 50 KiB entry; rotation is done explicitly
    /// via `rotate_active_for_test` so max_file_size == write_buffer_size
    /// satisfies Config::validate.
    fn open_large_value_tree(dir: &std::path::Path) -> VarTree<[u8; 8]> {
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 128 * 1024;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        VarTree::open(dir, cfg).expect("open large-value test tree")
    }

    fn build_large_payload(seed: u8) -> Vec<u8> {
        (0..50_000u32)
            .map(|i| (i as u8).wrapping_add(seed))
            .collect()
    }

    fn open_value_cache_tree(dir: &std::path::Path) -> VarTree<[u8; 8]> {
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 128 * 1024;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        cfg.value_cache.max_size = 4 << 20; // enable the value cache
        VarTree::open(dir, cfg).expect("open value-cache test tree")
    }

    #[test]
    fn large_value_miss_then_cached() {
        let dir = tempdir().unwrap();
        let tree = open_value_cache_tree(dir.path());
        let key = 1u64.to_be_bytes();
        let payload = build_large_payload(0x33);
        tree.put(&key, &payload).unwrap();

        // First read is a miss that populates the cache.
        assert_eq!(tree.get(&key).unwrap().as_bytes(), &payload[..]);

        // The value is now cached at its current loc.
        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).unwrap();
        let disk = node.read_loc();
        let vkey = crate::value_cache::ValueKey {
            shard_id: 0,
            file_id: disk.file_id,
            offset: disk.offset,
        };
        assert!(
            tree.value_cache.get(&vkey).is_some(),
            "value cached after read"
        );
        drop(guard);

        // Second read still correct.
        assert_eq!(tree.get(&key).unwrap().as_bytes(), &payload[..]);
    }

    #[test]
    fn large_value_read_then_edit_returns_new() {
        let dir = tempdir().unwrap();
        let tree = open_value_cache_tree(dir.path());
        let key = 7u64.to_be_bytes();
        let v1 = build_large_payload(0x11);
        let v2 = build_large_payload(0x22);
        tree.put(&key, &v1).unwrap();
        assert_eq!(tree.get(&key).unwrap().as_bytes(), &v1[..]); // caches v1 at loc1
        tree.put(&key, &v2).unwrap(); // node -> loc2
        // verify-on-hit: fresh loc misses the v1 entry, reads v2.
        assert_eq!(tree.get(&key).unwrap().as_bytes(), &v2[..]);
    }

    #[test]
    fn large_value_read_then_compact_returns_live() {
        let dir = tempdir().unwrap();
        let tree = open_value_cache_tree(dir.path());
        let key = 102u64.to_be_bytes();
        let payload = build_large_payload(0x42);
        tree.put(&key, &payload).unwrap();
        assert_eq!(tree.get(&key).unwrap().as_bytes(), &payload[..]); // cache the large value

        tree.engine.shards()[0]
            .rotate_active_for_test(8)
            .expect("rotate");
        for i in 1..20u8 {
            tree.put(&key, &[i; 256]).unwrap();
        }
        tree.put(&key, b"live-after-compaction").unwrap();
        let shard = &tree.engine.shards()[0];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        // Must return the live value, never the stale cached large payload.
        assert_eq!(tree.get(&key).unwrap().as_bytes(), b"live-after-compaction");
    }

    #[test]
    fn large_value_disabled_cache_reads_correctly() {
        let dir = tempdir().unwrap();
        let tree = open_large_value_tree(dir.path()); // value_cache disabled (Config::test)
        let key = 3u64.to_be_bytes();
        let payload = build_large_payload(0x55);
        tree.put(&key, &payload).unwrap();
        assert_eq!(tree.get(&key).unwrap().as_bytes(), &payload[..]);
        assert_eq!(tree.get(&key).unwrap().as_bytes(), &payload[..]);
    }

    /// End-to-end: 50 KiB value, force rotation to immutable, then `get`
    /// must return correct bytes via read_value_from_disk_locked.
    ///
    /// `open_large_value_tree` uses the default CacheConfig (max_size = 0),
    /// so the block cache is disabled and the read goes through the
    /// large-value locked fallback → disk read on every call.
    #[test]
    fn large_value_read_via_locked_fallback() {
        let dir = tempdir().unwrap();
        let tree = open_large_value_tree(dir.path());

        let key = 100u64.to_be_bytes();
        let payload = build_large_payload(0);
        tree.put(&key, &payload).expect("put large");
        tree.engine.shards()[0]
            .rotate_active_for_test(8)
            .expect("rotate");

        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        let v = tree
            .read_value_cached(node, &guard)
            .expect("read must succeed via locked fallback");
        assert_eq!(v.as_bytes(), payload.as_slice());
    }

    #[cfg(feature = "encryption")]
    #[test]
    fn get_or_err_and_try_get_surface_read_fault() {
        use crate::test_faults::{big_value, corrupt_tags};
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 128 * 1024;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        cfg.encryption_key = Some([7u8; 32]);
        let tree: VarTree<[u8; 8]> = VarTree::open(dir.path(), cfg).expect("open enc");

        let key = 1u64.to_be_bytes();
        tree.put(&key, &big_value(0xAB)).expect("put");
        tree.engine.shards()[0]
            .rotate_active_for_test(8)
            .expect("rotate");
        corrupt_tags(&tree.engine.shard_dirs()[0]);

        // lossy get still swallows the error
        assert!(tree.get(&key).is_none());
        // strict reads surface it — NOT KeyNotFound
        assert!(matches!(
            tree.get_or_err(&key),
            Err(DbError::EncryptionError(_))
        ));
        assert!(matches!(
            tree.try_get(&key),
            Err(DbError::EncryptionError(_))
        ));
        // genuinely-absent key is still KeyNotFound / Ok(None)
        let absent = 999u64.to_be_bytes();
        assert!(matches!(
            tree.get_or_err(&absent),
            Err(DbError::KeyNotFound)
        ));
        assert!(matches!(tree.try_get(&absent), Ok(None)));
    }

    #[cfg(feature = "encryption")]
    #[test]
    fn mutators_surface_read_fault_not_keynotfound() {
        use crate::test_faults::{big_value, corrupt_tags};
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 128 * 1024;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        cfg.encryption_key = Some([7u8; 32]);
        let tree: VarTree<[u8; 8]> = VarTree::open(dir.path(), cfg).expect("open enc");

        let key = 1u64.to_be_bytes();
        tree.put(&key, &big_value(0xAB)).unwrap();
        tree.engine.shards()[0]
            .rotate_active_for_test(8)
            .expect("rotate");
        corrupt_tags(&tree.engine.shard_dirs()[0]);

        assert!(matches!(
            tree.cas(&key, b"x", b"y"),
            Err(DbError::EncryptionError(_))
        ));
        assert!(matches!(
            tree.compare_delete(&key, b"x"),
            Err(DbError::EncryptionError(_))
        ));
        assert!(matches!(
            tree.update(&key, |_| ByteView::new(b"z")),
            Err(DbError::EncryptionError(_))
        ));
        assert!(matches!(
            tree.fetch_update(&key, |_| ByteView::new(b"z")),
            Err(DbError::EncryptionError(_))
        ));
    }

    #[cfg(feature = "encryption")]
    #[test]
    fn first_skips_unreadable_boundary_try_first_surfaces_it() {
        use crate::test_faults::{big_value, corrupt_tags};
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 128 * 1024;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        cfg.encryption_key = Some([7u8; 32]);
        // Default reversed=true: first() == largest key. Make key 2 the boundary.
        let tree: VarTree<[u8; 8]> = VarTree::open(dir.path(), cfg).expect("open enc");

        let k1 = 1u64.to_be_bytes();
        let k2 = 2u64.to_be_bytes();
        tree.put(&k1, &big_value(0x11)).unwrap();
        tree.put(&k2, &big_value(0x22)).unwrap();
        tree.engine.shards()[0]
            .rotate_active_for_test(8)
            .expect("rotate");
        corrupt_tags(&tree.engine.shard_dirs()[0]); // both entries now unreadable

        // Both unreadable -> lossy first()/last() skip everything -> None, and match iter.
        assert!(tree.first().is_none());
        assert_eq!(
            tree.first().map(|(k, _)| k),
            tree.iter().next().map(|(k, _)| k)
        );
        // Strict try_first surfaces the boundary read error, not None.
        assert!(matches!(tree.try_first(), Err(DbError::EncryptionError(_))));
        assert!(matches!(tree.try_last(), Err(DbError::EncryptionError(_))));
    }

    #[cfg(feature = "encryption")]
    #[test]
    fn large_value_read_encrypted() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 128 * 1024;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        cfg.encryption_key = Some([7u8; 32]);
        // Cache stays at default (max_size = 0, disabled). The locked
        // fallback always goes to step 3 -> read_value_from_disk_locked,
        // which routes to pread_value_encrypted under this feature gate.
        let tree: VarTree<[u8; 8]> = VarTree::open(dir.path(), cfg).expect("open enc");

        let key = 101u64.to_be_bytes();
        let payload = build_large_payload(0xAB);
        tree.put(&key, &payload).expect("put encrypted large");
        tree.engine.shards()[0]
            .rotate_active_for_test(8)
            .expect("rotate");

        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        let v = tree
            .read_value_cached(node, &guard)
            .expect("encrypted large read must succeed via pread_value_encrypted");
        assert_eq!(v.as_bytes(), payload.as_slice());
    }

    /// Deterministic stale-DiskLoc test for large values: build a large
    /// entry, capture its DiskLoc, overwrite enough to make the file fully
    /// dead, compact (removes the file), then the captured DiskLoc must
    /// surface as StaleDiskLoc via the locked fallback. The public
    /// retry path then returns the live value.
    #[test]
    fn large_value_stale_disk_loc_deterministic() {
        let dir = tempdir().unwrap();
        let tree = open_large_value_tree(dir.path());

        let key = 102u64.to_be_bytes();
        let payload = build_large_payload(0x42);
        tree.put(&key, &payload).expect("first large put");

        // Snapshot the DiskLoc of the first put — points at the file we
        // will erase via compaction.
        let snap = {
            let guard = tree.index.collector().enter();
            let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
            node.read_loc()
        };

        // Rotate the active file so the large entry moves to an immutable file.
        tree.engine.shards()[0]
            .rotate_active_for_test(8)
            .expect("rotate after large put");

        // Overwrite many times with small payloads to move the live pointer off
        // the original file (making it 100% dead so compaction erases it).
        for i in 1..20u8 {
            tree.put(&key, &[i; 256]).expect("overwrite");
        }
        tree.put(&key, b"live-after-compaction").expect("final put");

        let shard_id = tree.shard_for(&key) as u8;
        let shard = &tree.engine.shards()[shard_id as usize];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        // Direct call on the stale snapshot — must propagate StaleDiskLoc
        // through the large-value locked fallback.
        match tree.read_value_cached_inner(&snap, shard_id) {
            Err(DbError::StaleDiskLoc) => {}
            Ok(v) => panic!("expected StaleDiskLoc, got Ok({:?})", v.as_bytes()),
            Err(e) => panic!("expected StaleDiskLoc, got Err({e})"),
        }

        // Public retry path picks up the live value.
        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        let v = tree
            .read_value_cached(node, &guard)
            .expect("public path must retry and return live value");
        assert_eq!(v.as_bytes(), b"live-after-compaction");
    }

    /// `_result` step 2: single-block cache fast path.
    /// Writes a value, snapshots its DiskLoc, manually inserts the matching
    /// 4 KiB block into the cache, then calls `_result` and verifies the
    /// cache-hit path returns the correct bytes.
    #[test]
    fn read_value_locked_result_ok_from_cache_single_block() {
        let dir = tempdir().unwrap();
        // Enable the cache so `cache.insert` actually takes effect.
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 8192;
        cfg.write_buffer_size = 8192;
        cfg.compaction_threshold = 0.0;
        cfg.block_cache.max_size = 1 << 20;
        let tree: VarTree<[u8; 8]> = VarTree::open(dir.path(), cfg).expect("open");

        let key = 9u64.to_be_bytes();
        let payload = b"small-single-block-value";
        tree.put(&key, payload).expect("put");
        // Force rotation so the entry leaves the active write buffer (otherwise
        // step 1 short-circuits and step 2 never runs). Need enough writes to
        // push write_offset past max_file_size=8192.
        for i in 100u64..135 {
            tree.put(&i.to_be_bytes(), &[i as u8; 256])
                .expect("rotator");
        }

        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        let disk = node.read_loc();
        drop(guard);

        // Sanity: this value fits in a single block (step 2 only runs when
        // start + len <= 4096).
        let start = (disk.offset & 4095) as usize;
        let len = disk.len as usize;
        assert!(
            start + len <= 4096,
            "test precondition: value must fit in a single block",
        );

        // Pre-warm the cache BEFORE acquiring the shard lock.
        // get_or_read_block → shard.read_block → sync::lock(&inner), so it
        // must not be called while the shard lock is already held.
        let block_offset = disk.offset as u64 & !4095;
        let shard_id = tree.shard_for(&key) as u8;
        let cache_key = BlockKey {
            shard_id,
            file_id: disk.file_id,
            block_offset,
        };
        let block = tree
            .get_or_read_block(shard_id, disk.file_id, block_offset)
            .expect("read block");
        tree.block_cache.insert(cache_key, block);
        assert!(
            tree.block_cache.get(&cache_key).is_some(),
            "cache must contain the block for step 2 to fire",
        );

        let shard = &tree.engine.shards()[shard_id as usize];
        let inner = shard.lock();
        assert_ne!(
            disk.file_id, inner.active.file_id,
            "test setup failed: key entry is still in the active write buffer",
        );

        let v = tree
            .read_value_locked_result(&disk, shard_id, &inner)
            .expect("cache-hit read must succeed");
        assert_eq!(v.as_bytes(), payload);
    }

    /// End-to-end roundtrip with file_id above u16::MAX.
    ///
    /// Uses write_buffer_size=128 KiB so a single 512-byte entry fits in one
    /// flush cycle; max_file_size=4096 forces per-entry rotation; cache
    /// enabled so the fast path is also exercised.  After bumping
    /// next_file_id to 70_000 and rotating, the active file gets an id that
    /// exceeds u16::MAX.  The entry is flushed + rotated to make it
    /// immutable, then `get` must return the correct bytes via the locked
    /// disk-read path.  Before the file_id u32 widening this would either
    /// panic or return wrong bytes because the high bits were truncated.
    #[test]
    fn var_tree_get_with_file_id_above_u16() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 128 * 1024;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        cfg.block_cache.max_size = 1 << 20;
        let tree: VarTree<[u8; 8]> = VarTree::open(dir.path(), cfg).expect("open");

        let shard = &tree.engine.shards()[0];

        // Bump file id well past u16::MAX and rotate so the next active file
        // carries the new id.
        shard.set_next_file_id(70_000);
        shard.rotate_active_for_test(8).expect("first rotate");
        assert!(
            shard.active_file_id() >= 70_000,
            "active_file_id should be >= 70_000 after rotation"
        );

        let key = 42u64.to_be_bytes();
        let value = vec![0xC3u8; 512];
        tree.put(&key, &value).expect("put");

        // Confirm the DiskLoc has file_id > u16::MAX.
        {
            let guard = tree.index.collector().enter();
            let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
            let disk = node.read_loc();
            assert!(
                disk.file_id > u16::MAX as u32,
                "DiskLoc.file_id must be above u16::MAX, got {}",
                disk.file_id,
            );
        }

        // Flush the write buffer and rotate so the entry lands on an
        // immutable file — reads will go through Shard::read_block.
        shard.flush().expect("flush");
        shard.rotate_active_for_test(8).expect("second rotate");

        let got = tree.get(&key).expect("get must return Some");
        assert_eq!(got.as_bytes(), value.as_slice());
    }

    /// Recovery roundtrip with file_id above u16::MAX.
    ///
    /// Phase A: open, bump next_file_id past u16::MAX, rotate once (so the
    /// active file gets an id > 65535), write entries, close.  `close()` calls
    /// `sync_hints()` which flushes the write buffer and writes the hint file
    /// with the correct key length (8), then `engine.flush()` fsyncs.
    /// Phase B: reopen from the same tempdir, read every entry back, confirm
    /// at least one on-disk file_id still exceeds u16::MAX.
    ///
    /// Before the file_id u32 widening the high bits were truncated during
    /// hint-file serialization and recovery; this test pins that code path so
    /// a future regression cannot go undetected.
    #[test]
    fn recovery_handles_file_id_above_u16() {
        let dir = tempdir().unwrap();

        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 128 * 1024;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        cfg.block_cache.max_size = 1 << 20;

        // --- Phase A: open, bump file_id past u16, write, close ---
        {
            let tree: VarTree<[u8; 8]> = VarTree::open(dir.path(), cfg.clone()).expect("open A");
            let shard = &tree.engine.shards()[0];

            // Bump the id counter and rotate so the active file gets id 70_000.
            shard.set_next_file_id(70_000);
            shard
                .rotate_active_for_test(8)
                .expect("rotate to file_id 70_000");
            assert!(
                shard.active_file_id() >= 70_000,
                "active_file_id should be >= 70_000 after rotation"
            );

            for i in 0u64..4 {
                let key = i.to_be_bytes();
                let value = vec![i as u8; 200];
                tree.put(&key, &value).expect("put phase A");
            }

            tree.close().expect("close phase A");
        }

        // --- Phase B: reopen, read entries, assert wide file_id persists ---
        {
            let tree: VarTree<[u8; 8]> = VarTree::open(dir.path(), cfg).expect("open B");

            assert_eq!(tree.len(), 4, "all 4 entries must survive recovery");

            for i in 0u64..4 {
                let key = i.to_be_bytes();
                let expected = vec![i as u8; 200];
                let got = tree
                    .get(&key)
                    .unwrap_or_else(|| panic!("key {i} not found after recovery"));
                assert_eq!(
                    got.as_bytes(),
                    expected.as_slice(),
                    "value mismatch for key {i} after recovery"
                );
            }

            // Verify at least one on-disk file_id exceeded u16::MAX so that
            // the test actually exercises the wide-id path.
            let shard = &tree.engine.shards()[0];
            let max_fid = shard.file_ids().into_iter().max().expect("non-empty");
            assert!(
                max_fid > u16::MAX as u32,
                "max file_id should exceed u16::MAX after recovery (got {})",
                max_fid
            );
        }
    }

    #[test]
    fn warmup_skips_large_values_block_cache() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 128 * 1024;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        cfg.block_cache.max_size = 1 << 20; // enable block cache so inserts are observable
        let tree = VarTree::<[u8; 8]>::open(dir.path(), cfg).unwrap();

        let key = 5u64.to_be_bytes();
        let payload = build_large_payload(0x42); // 50_000 bytes -> large
        tree.put(&key, &payload).unwrap();
        // Rotate to immutable so warmup can actually cache the block (the
        // unflushed active block is not cacheable — is_full_block=false).
        tree.engine.shards()[0]
            .rotate_active_for_test(8)
            .expect("rotate");

        // Precondition: the value is large (spans > 2 blocks).
        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).unwrap();
        let disk = node.read_loc();
        assert!(
            disk.is_value_cache_routed(),
            "test precondition: value must be large"
        );
        drop(guard);

        tree.warmup().unwrap();

        // The large value's first block must NOT be in the block cache.
        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).unwrap();
        let disk = node.read_loc();
        let bkey = BlockKey {
            shard_id: 0,
            file_id: disk.file_id,
            block_offset: disk.offset as u64 & !4095,
        };
        assert!(
            tree.block_cache.get(&bkey).is_none(),
            "warmup must not cache a large value's first block"
        );
    }

    /// `warmup` only ever inserted `disk.offset & !4095` — the *first* block of
    /// each live value. A value that straddles a 4 KiB boundary (but stays
    /// under the 2-block value-cache threshold) left its tail block cold, so the
    /// first read after warmup still went to disk via `next_block()`.
    #[test]
    fn warmup_loads_both_blocks_of_a_straddling_value() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 1 << 20;
        cfg.write_buffer_size = 1 << 20;
        cfg.compaction_threshold = 0.0;
        cfg.block_cache.max_size = 1 << 20;
        let tree = VarTree::<[u8; 8]>::open(dir.path(), cfg).unwrap();

        // Entries are laid down back to back behind a 24-byte header, so the
        // offsets below are deterministic:
        //   filler  -> 24,   spans block 0 only
        //   straddler -> 3048, spans blocks 0 and 4096   <- the value under test
        //   tail    -> 4572, > 2 blocks, so warmup skips it entirely
        // The tail matters twice: it fills block 4096 (a partial block is never
        // cached at all), and being value-cache routed it is not itself a reason
        // for block 4096 to be warmed — only the straddler can put it there.
        tree.put(&0u64.to_be_bytes(), &vec![0x11u8; 3000]).unwrap();
        tree.put(&1u64.to_be_bytes(), &vec![0x5Au8; 1500]).unwrap();
        tree.put(&2u64.to_be_bytes(), &vec![0x77u8; 30_000])
            .unwrap();
        tree.engine.shards()[0]
            .rotate_active_for_test(8)
            .expect("rotate");

        let guard = tree.index.collector().enter();
        let straddler = tree
            .index
            .get(1u64.to_be_bytes().as_bytes(), &guard)
            .unwrap()
            .read_loc();
        let start = (straddler.offset & 4095) as usize;
        assert!(
            !straddler.is_value_cache_routed() && start + straddler.len as usize > 4096,
            "test precondition: value must straddle a block boundary and stay block-cache routed \
             (start={start}, len={})",
            straddler.len
        );
        let tail = tree
            .index
            .get(2u64.to_be_bytes().as_bytes(), &guard)
            .unwrap()
            .read_loc();
        assert!(
            tail.is_value_cache_routed(),
            "test precondition: the tail value must be value-cache routed so warmup skips it"
        );
        drop(guard);

        tree.warmup().unwrap();

        let block_offset = straddler.offset as u64 & !4095;
        let bkey = |off| BlockKey {
            shard_id: 0,
            file_id: straddler.file_id,
            block_offset: off,
        };
        assert!(
            tree.block_cache.get(&bkey(block_offset)).is_some(),
            "warmup must cache the value's first block"
        );
        assert!(
            tree.block_cache.get(&bkey(block_offset + 4096)).is_some(),
            "warmup must also cache the block the value spills into"
        );
    }

    // ---------------------------------------------------------------------------
    // Tests — get_many / update_many (Task 5)
    // ---------------------------------------------------------------------------

    #[test]
    fn var_get_many_lossy_aligned() {
        let dir = tempdir().unwrap();
        let cfg = Config::test();
        let tree = VarTree::<[u8; 8]>::open(dir.path(), cfg).unwrap();
        tree.put(&1u64.to_be_bytes(), b"alpha").unwrap();
        tree.put(&2u64.to_be_bytes(), b"beta").unwrap();
        let keys: Vec<[u8; 8]> = vec![2u64.to_be_bytes(), 9u64.to_be_bytes(), 1u64.to_be_bytes()];
        let got = tree.get_many(&keys);
        assert_eq!(got.len(), 3);
        assert_eq!(got[0].as_deref(), Some(&b"beta"[..]));
        assert_eq!(got[1], None);
        assert_eq!(got[2].as_deref(), Some(&b"alpha"[..]));
    }

    #[test]
    fn var_get_many_large_value_cache_path() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        // Enable the whole-value cache so this regression actually exercises the
        // cache-backed read path named by the test (Config::test() disables it).
        cfg.value_cache.max_size = 4 << 20;
        let tree = VarTree::<[u8; 8]>::open(dir.path(), cfg).unwrap();
        let big = vec![7u8; 16 * 1024]; // > 8 KiB -> value-cache route
        tree.put(&1u64.to_be_bytes(), &big).unwrap();
        // First read populates the value cache (miss), second read is a cache hit;
        // both must return the full value.
        let got1 = tree.get_many(&[1u64.to_be_bytes()]);
        assert_eq!(got1[0].as_deref(), Some(&big[..]));
        let got2 = tree.get_many(&[1u64.to_be_bytes()]);
        assert_eq!(got2[0].as_deref(), Some(&big[..]));
    }

    #[test]
    fn var_update_many_set_keep_delete() {
        use crate::{Applied, Upsert};
        let dir = tempdir().unwrap();
        let cfg = Config::test();
        let tree = VarTree::<[u8; 8]>::open(dir.path(), cfg).unwrap();
        tree.put(&1u64.to_be_bytes(), b"old").unwrap();
        tree.put(&2u64.to_be_bytes(), b"gone").unwrap();

        let items: Vec<([u8; 8], &[u8])> = vec![
            (1u64.to_be_bytes(), b"new"),
            (2u64.to_be_bytes(), b""),    // delete
            (3u64.to_be_bytes(), b"ins"), // insert
            (4u64.to_be_bytes(), b""),    // delete-miss
        ];
        let out = tree
            .update_many(items, |k, _cur, p| {
                let k2 = 2u64.to_be_bytes();
                let k4 = 4u64.to_be_bytes();
                if *k == k2 || *k == k4 {
                    Upsert::Delete
                } else {
                    Upsert::Put(ByteView::new(p))
                }
            })
            .unwrap();
        assert!(matches!(out[0].1, Applied::Written { .. }));
        assert!(matches!(out[1].1, Applied::Deleted(_)));
        assert!(matches!(out[2].1, Applied::Written { old: None, .. }));
        assert_eq!(out[3].1, Applied::Kept);

        assert_eq!(tree.get(&1u64.to_be_bytes()).as_deref(), Some(&b"new"[..]));
        assert!(tree.get(&2u64.to_be_bytes()).is_none());
        assert_eq!(tree.get(&3u64.to_be_bytes()).as_deref(), Some(&b"ins"[..]));
    }

    #[test]
    fn var_tree_shard_update_and_fetch_update() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<true, true>(dir.path(), CountingHook::default());
        let key = 1u64.to_be_bytes();
        tree.put(&key, b"a").unwrap();

        let (upd, fetched, missing) = tree
            .atomic(&key, |shard| {
                let upd = shard.update(&key, |old| {
                    let mut v = old.to_vec();
                    v.push(b'b');
                    ByteView::from(v.as_slice())
                })?;
                let fetched = shard.fetch_update(&key, |old| {
                    let mut v = old.to_vec();
                    v.push(b'c');
                    ByteView::from(v.as_slice())
                })?;
                let missing = shard.update(&2u64.to_be_bytes(), |old| ByteView::from(old))?;
                Ok((upd, fetched, missing))
            })
            .expect("atomic");

        assert_eq!(upd.as_deref(), Some(&b"ab"[..])); // new value
        assert_eq!(fetched.as_deref(), Some(&b"ab"[..])); // old value before +c
        assert!(missing.is_none());
        assert_eq!(tree.get(&key).as_deref(), Some(&b"abc"[..]));
    }

    #[test]
    fn var_tree_atomic_update_fires_hook() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<true, true>(dir.path(), CountingHook::default());
        let key = 1u64.to_be_bytes();

        tree.atomic(&key, |shard| {
            shard.put(&key, b"init")?; // write 1: new insert, no old value
            shard.update(&key, |old| {
                let mut v = old.to_vec();
                v.push(b'!');
                ByteView::from(v.as_slice())
            })?; // write 2: update, old = b"init"
            Ok(())
        })
        .expect("atomic");

        // Hook fires for both put and update after the shard lock is released.
        assert_eq!(
            tree.hook.writes.load(AtomicOrdering::Relaxed),
            2,
            "hook must fire once for put and once for update"
        );
        // The update saw the old value (NEEDS_OLD_VALUE=true) → writes_with_old incremented.
        assert_eq!(
            tree.hook.writes_with_old.load(AtomicOrdering::Relaxed),
            1,
            "only the update event carries an old value"
        );
        // Final value is b"init!"
        assert_eq!(tree.get(&key).as_deref(), Some(&b"init!"[..]));
    }

    /// Regression (WP-B, page-straddle read): under `direct_io` a non-force flush
    /// writes only *complete* 4096-byte pages; a value whose prefix landed on
    /// disk but whose tail is still buffered straddles `base_offset`.
    /// `WriteBuffer::read` returns `None` for such a loc, so before the fix the
    /// ≤2-block cached path silently stitched the disk prefix to a zero-padded
    /// second block and returned the value with a zeroed tail.
    #[test]
    fn straddle_read_2block_direct_io_returns_full_value() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 1 << 20;
        cfg.write_buffer_size = 8192;
        cfg.direct_io = true;
        cfg.compaction_threshold = 0.0;
        let tree = VarTree::<[u8; 8]>::open(dir.path(), cfg).expect("open");

        // Two ~3000-byte values. entry = pad8(16 + 8 + 3000) = 3024 bytes.
        // v1 value bytes [24, 3024); v2 value bytes [3048, 6048) — straddles 4096.
        let v1: Vec<u8> = (0..3000).map(|i| ((i % 251) + 1) as u8).collect();
        let v2: Vec<u8> = (0..3000).map(|i| ((i % 241) + 2) as u8).collect();
        let k1 = 1u64.to_be_bytes();
        let k2 = 2u64.to_be_bytes();
        tree.put(&k1, &v1).expect("put v1");
        tree.put(&k2, &v2).expect("put v2");

        // Non-force flush: writes page 0 (4096 bytes), keeps the 1952-byte tail in
        // the buffer, advancing base_offset to 4096.
        tree.flush_buffers().expect("flush");

        assert_eq!(
            tree.get(&k2).as_deref(),
            Some(&v2[..]),
            "straddling value must read back byte-for-byte (tail was zeroed before the fix)"
        );
        // The fully-flushed prefix value must still read correctly too.
        assert_eq!(tree.get(&k1).as_deref(), Some(&v1[..]));
    }

    /// Regression (WP-B, page-straddle read): the large-value (>2-block) path
    /// `pread`s the whole value lock-free. When the value straddles `base_offset`
    /// the tail is not yet on disk, so `aligned_read_exact` hit `UnexpectedEof`
    /// and `try_get` returned `Err(Io)` for a live key. The fix stitches the
    /// flushed prefix to the buffered tail under the shard lock.
    #[test]
    fn straddle_read_large_value_direct_io_try_get_ok() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 1 << 20;
        cfg.write_buffer_size = 16 * 1024;
        cfg.direct_io = true;
        cfg.compaction_threshold = 0.0;
        let tree = VarTree::<[u8; 8]>::open(dir.path(), cfg).expect("open");

        // v1 ~3000 bytes -> entry 3024; v2 9000 bytes -> value bytes [3048, 12048).
        // (3048 & 4095) + 9000 = 12048 > 8192 -> large-value path.
        let v1: Vec<u8> = (0..3000).map(|i| ((i % 251) + 1) as u8).collect();
        let v2: Vec<u8> = (0..9000).map(|i| ((i % 239) + 3) as u8).collect();
        let k1 = 1u64.to_be_bytes();
        let k2 = 2u64.to_be_bytes();
        tree.put(&k1, &v1).expect("put v1");
        tree.put(&k2, &v2).expect("put v2");

        // Non-force flush: writes pages 0..2 (8192 bytes), keeps the 3856-byte tail
        // in the buffer, advancing base_offset to 8192.
        tree.flush_buffers().expect("flush");

        let got = tree
            .try_get(&k2)
            .expect("try_get must not error for a live key");
        assert_eq!(
            got.as_deref(),
            Some(&v2[..]),
            "straddling large value must read back byte-for-byte"
        );
    }

    /// Regression (WP-B, page-straddle read, encrypted): the same straddle under
    /// encryption made the ≤2-block path fault on a missing page tag for the
    /// unflushed second page (`EncryptionError`). The straddle helper reads the
    /// flushed prefix through the tagged pages and copies the buffered tail.
    #[cfg(feature = "encryption")]
    #[test]
    fn straddle_read_2block_encrypted_returns_full_value() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 1 << 20;
        cfg.write_buffer_size = 8192;
        cfg.encryption_key = Some([7u8; 32]);
        cfg.compaction_threshold = 0.0;
        let tree = VarTree::<[u8; 8]>::open(dir.path(), cfg).expect("open enc");

        let v1: Vec<u8> = (0..3000).map(|i| ((i % 251) + 1) as u8).collect();
        let v2: Vec<u8> = (0..3000).map(|i| ((i % 241) + 2) as u8).collect();
        let k1 = 1u64.to_be_bytes();
        let k2 = 2u64.to_be_bytes();
        tree.put(&k1, &v1).expect("put v1");
        tree.put(&k2, &v2).expect("put v2");
        tree.flush_buffers().expect("flush");

        let got = tree
            .try_get(&k2)
            .expect("encrypted straddle read must not error");
        assert_eq!(got.as_deref(), Some(&v2[..]));
        assert_eq!(tree.get(&k1).as_deref(), Some(&v1[..]));
    }

    /// Regression (WP-B follow-up, page-straddle read on the block-cache HIT
    /// path): when the value's first (prefix) block is already in `block_cache`,
    /// `read_value_cached_inner` returns early via `extract_from_block`, whose
    /// second-block closure reads `block_offset + 4096` straight from disk and
    /// ignores the write buffer. For a value straddling `base_offset` that second
    /// block is unflushed, so the tail came back zeroed. Settled active blocks
    /// are cached automatically now, so the path is reachable through an
    /// ordinary read of `k1`; the cache is still seeded directly here so the
    /// test does not depend on read order.
    #[test]
    fn straddle_read_2block_cache_hit_returns_full_value() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 1 << 20;
        cfg.write_buffer_size = 8192;
        cfg.direct_io = true;
        cfg.block_cache.max_size = 1 << 20;
        cfg.compaction_threshold = 0.0;
        let tree = VarTree::<[u8; 8]>::open(dir.path(), cfg).expect("open");

        let v1: Vec<u8> = (0..3000).map(|i| ((i % 251) + 1) as u8).collect();
        let v2: Vec<u8> = (0..3000).map(|i| ((i % 241) + 2) as u8).collect();
        let k1 = 1u64.to_be_bytes();
        let k2 = 2u64.to_be_bytes();
        tree.put(&k1, &v1).expect("put v1");
        tree.put(&k2, &v2).expect("put v2");
        // Non-force flush: page 0 to disk, tail of v2 ([4096, 6048)) stays buffered.
        tree.flush_buffers().expect("flush");

        // Seed the cache with the fully-flushed prefix block 0 to force the
        // step-1 cache-hit branch regardless of read order.
        let shard_id = tree.shard_for(&k2) as u8;
        let key0 = BlockKey {
            shard_id,
            file_id: 1,
            block_offset: 0,
        };
        let (blk0, _is_full) = tree.engine.shards()[shard_id as usize]
            .read_block(1, 0)
            .expect("read block 0");
        tree.block_cache.insert(key0, Arc::new(blk0));

        assert_eq!(
            tree.get(&k2).as_deref(),
            Some(&v2[..]),
            "straddling value via cache-hit path must read back byte-for-byte (tail was zeroed before the fix)"
        );
    }

    /// Config for the settled-active-block tests: one shard, a file far larger
    /// than the corpus (so nothing ever rotates), an 8 KiB write buffer and a
    /// block cache.
    fn settled_block_cfg(direct_io: bool) -> Config {
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 1 << 20;
        cfg.write_buffer_size = 8192;
        cfg.direct_io = direct_io;
        cfg.block_cache.max_size = 1 << 20;
        cfg.compaction_threshold = 0.0;
        cfg
    }

    fn active_file_id(tree: &VarTree<[u8; 8]>) -> u32 {
        tree.engine.shards()[0].lock().active.file_id
    }

    /// Two ~3 KiB values, then a non-force flush: page 0 lands on disk while
    /// the tail of the second value stays buffered. Returns `(k1, v1, k2, v2)`.
    #[allow(clippy::type_complexity)]
    fn put_two_and_flush(tree: &VarTree<[u8; 8]>) -> ([u8; 8], Vec<u8>, [u8; 8], Vec<u8>) {
        let v1: Vec<u8> = (0..3000).map(|i| ((i % 251) + 1) as u8).collect();
        let v2: Vec<u8> = (0..3000).map(|i| ((i % 241) + 2) as u8).collect();
        let k1 = 1u64.to_be_bytes();
        let k2 = 2u64.to_be_bytes();
        tree.put(&k1, &v1).expect("put v1");
        tree.put(&k2, &v2).expect("put v2");
        tree.flush_buffers().expect("flush");
        (k1, v1, k2, v2)
    }

    /// A block of the **active** file that lies entirely below the write
    /// buffer's `base_offset` is settled — append-only allocation never rewrites
    /// it — so it is cached like an immutable block, and the second read is
    /// served without disk I/O.
    #[test]
    fn settled_active_block_is_cached_and_served_without_disk_reads() {
        let dir = tempdir().unwrap();
        let tree = VarTree::<[u8; 8]>::open(dir.path(), settled_block_cfg(true)).expect("open");
        let (k1, v1, _, _) = put_two_and_flush(&tree);
        let key0 = BlockKey {
            shard_id: 0,
            file_id: active_file_id(&tree),
            block_offset: 0,
        };
        assert!(
            tree.block_cache.get(&key0).is_none(),
            "nothing cached before the first read"
        );

        let before = test_disk_value_reads();
        assert_eq!(tree.get(&k1).as_deref(), Some(&v1[..]));
        assert_eq!(test_disk_value_reads(), before + 1, "first read misses");
        assert!(
            tree.block_cache.get(&key0).is_some(),
            "block 0 is fully below base_offset and must be cached even though the file is active"
        );
        assert_eq!(tree.get(&k1).as_deref(), Some(&v1[..]));
        assert_eq!(
            test_disk_value_reads(),
            before + 1,
            "second read must hit the block cache"
        );
    }

    /// The block containing `base_offset` still has unflushed bytes in the
    /// write buffer: reading it yields a zero-padded tail, so it must not be
    /// cached.
    #[test]
    fn active_block_containing_base_offset_is_not_cached() {
        let dir = tempdir().unwrap();
        let tree = VarTree::<[u8; 8]>::open(dir.path(), settled_block_cfg(true)).expect("open");
        let (_, _, k2, v2) = put_two_and_flush(&tree);
        let file_id = active_file_id(&tree);
        assert_eq!(
            tree.engine.shards()[0].lock().write_buf.base_offset(),
            4096,
            "non-force page-granular flush leaves base_offset at the page boundary"
        );

        let _ = tree
            .get_or_read_block(0, file_id, 4096)
            .expect("read block 1");
        let key1 = BlockKey {
            shard_id: 0,
            file_id,
            block_offset: 4096,
        };
        assert!(
            tree.block_cache.get(&key1).is_none(),
            "block 1 straddles base_offset and must not be cached"
        );
        // The straddling value itself still reads back intact.
        assert_eq!(tree.get(&k2).as_deref(), Some(&v2[..]));
    }

    /// Byte-granular (buffered I/O) mode: a non-force flush writes the whole
    /// buffer, so `base_offset` stops mid-block at 6000-ish bytes. Block 0 is
    /// settled and cached; block 1 contains `base_offset` and is not.
    #[test]
    fn settled_active_block_is_cached_in_byte_granular_mode() {
        let dir = tempdir().unwrap();
        let tree = VarTree::<[u8; 8]>::open(dir.path(), settled_block_cfg(false)).expect("open");
        let (k1, v1, _, _) = put_two_and_flush(&tree);
        let file_id = active_file_id(&tree);
        let base = tree.engine.shards()[0].lock().write_buf.base_offset();
        assert!(
            (4096..8192).contains(&base),
            "byte-granular flush must leave base_offset mid-block, got {base}"
        );

        let before = test_disk_value_reads();
        assert_eq!(tree.get(&k1).as_deref(), Some(&v1[..]));
        assert_eq!(tree.get(&k1).as_deref(), Some(&v1[..]));
        assert_eq!(
            test_disk_value_reads(),
            before + 1,
            "second read must hit the block cache"
        );
        let key1 = BlockKey {
            shard_id: 0,
            file_id,
            block_offset: 4096,
        };
        let _ = tree
            .get_or_read_block(0, file_id, 4096)
            .expect("read block 1");
        assert!(
            tree.block_cache.get(&key1).is_none(),
            "block 1 contains base_offset"
        );
    }

    /// The scenario from the msg-back report: a corpus of small values that
    /// never rotates out of the active file. After close + reopen every block
    /// below the resumed `base_offset` is settled, so a second pass over the
    /// whole prefix pays zero disk reads.
    #[test]
    fn reopened_active_file_serves_second_pass_from_block_cache() {
        let dir = tempdir().unwrap();
        let value = vec![0xAB; 200];
        {
            let tree = VarTree::<[u8; 8]>::open(dir.path(), settled_block_cfg(true)).expect("open");
            for i in 0..200u64 {
                tree.put(&i.to_be_bytes(), &value).expect("put");
            }
            tree.close().expect("close");
        }
        let tree = VarTree::<[u8; 8]>::open(dir.path(), settled_block_cfg(true)).expect("reopen");
        assert_eq!(
            tree.engine.shards()[0].file_ids().len(),
            1,
            "no rotation happened"
        );

        let first_pass_start = test_disk_value_reads();
        for i in 0..200u64 {
            assert_eq!(tree.get(&i.to_be_bytes()).as_deref(), Some(&value[..]));
        }
        assert!(
            test_disk_value_reads() > first_pass_start,
            "cold pass reads disk"
        );
        let second_pass_start = test_disk_value_reads();
        for i in 0..200u64 {
            assert_eq!(tree.get(&i.to_be_bytes()).as_deref(), Some(&value[..]));
        }
        assert_eq!(
            test_disk_value_reads(),
            second_pass_start,
            "warm pass over the active file must be served entirely from the block cache"
        );
    }

    /// Encrypted active file: a settled page is sealed exactly once, so its
    /// decrypted block is cacheable exactly like a plaintext one.
    #[cfg(feature = "encryption")]
    #[test]
    fn settled_active_block_is_cached_encrypted() {
        let dir = tempdir().unwrap();
        let mut cfg = settled_block_cfg(false);
        cfg.encryption_key = Some([7u8; 32]);
        let tree = VarTree::<[u8; 8]>::open(dir.path(), cfg).expect("open enc");
        let (k1, v1, k2, v2) = put_two_and_flush(&tree);
        let file_id = active_file_id(&tree);

        let before = test_disk_value_reads();
        assert_eq!(tree.try_get(&k1).expect("get").as_deref(), Some(&v1[..]));
        assert_eq!(tree.try_get(&k1).expect("get").as_deref(), Some(&v1[..]));
        assert_eq!(
            test_disk_value_reads(),
            before + 1,
            "second read must hit the block cache"
        );
        let key1 = BlockKey {
            shard_id: 0,
            file_id,
            block_offset: 4096,
        };
        assert!(
            tree.block_cache.get(&key1).is_none(),
            "unsealed page is not cached"
        );
        assert_eq!(tree.try_get(&k2).expect("get").as_deref(), Some(&v2[..]));
    }

    #[test]
    fn keys_range_bounds_dir_matches_iter_keys_without_value_reads() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());
        for i in [1u8, 5, 10] {
            let mut k = [0u8; 8];
            k[7] = i;
            tree.put(&k, &[i; 64]).unwrap();
        }
        let from_iter: Vec<_> = tree.iter().map(|(k, _)| k).collect();
        let from_keys: Vec<_> = tree
            .keys_range_bounds_dir(Bound::Unbounded, Bound::Unbounded, dir_from(tree.reversed))
            .collect();
        assert_eq!(from_iter, from_keys);
    }
}