armdb/

var_typed_tree.rs

use std::marker::PhantomData;
use std::ops::Bound;
use std::sync::Arc;

use crate::Key;
use crate::batch::Applied;
use crate::byte_view::ByteView;
use crate::codec::Codec;
use crate::compaction::CompactionIndex;
use crate::config::Config;
use crate::decoded_cache::DecodedCache;
use crate::disk_loc::DiskLoc;
use crate::error::{DbError, DbResult};
use crate::hook::{NoHook, TypedWriteHook, VarTypedHookAdapter, WriteHook};
use crate::map_index::{Dir, dir_from};
use crate::scope::DirRangeIter;
use crate::upsert::Upsert;
use crate::var_tree::{TryVarIter, VarIter, VarShard, VarTree};
use crate::watch::{self, CodecWatchDecoder, WatchOptions, WatchReceiver, WatchSpec};

/// A tree with fixed-size keys and typed values `T`. Values are encoded via
/// a [`Codec`] and stored on disk (variable length), with a block cache
/// (≤ 8 KB values) and a value cache (> 8 KB values) for reads — unlike
/// [`TypedTree`](crate::TypedTree) which keeps values in memory.
///
/// Thin wrapper around [`VarTree<K, VarTypedHookAdapter<K, T, C, H>>`].
/// Each `VarTypedTree` owns its storage engine — one tree = one database directory.
///
/// Facade mutations and `atomic()` forward to the inner Var tree, so they ride
/// Var's `Family` / `Ctx` seam (borrowed facade events; `ByteView::from` only in
/// `own` for Held) without a separate VarTyped `Family` impl.
///
/// # When to use
///
/// - Typed values too large to keep in RAM → `VarTypedTree` (disk-resident).
/// - Typed values that fit in RAM → [`TypedTree`](crate::TypedTree) (in-memory).
/// - Fixed-size zerocopy values → [`ZeroTree`](crate::ZeroTree).
///
/// # Error handling
///
/// Follows `VarTree` / `TypedTree` precedent (variant A):
/// - `get` / `first` / `last` → `None` if key missing **or** decode fails.
/// - Iterators skip entries with decode errors (`continue` in `next()`).
/// - `migrate` is strict: a decode/encode error aborts it with `Err`.
///
/// # Decoded value cache
///
/// [`get_shared`](Self::get_shared) / [`try_get_shared`](Self::try_get_shared) /
/// [`get_many_shared`](Self::get_many_shared) serve `Arc<T>` from an optional
/// per-collection cache of **decoded** values (`Config::decoded_cache`, off by
/// default), keyed by the value's physical location — so an update, a delete or
/// a compaction invalidates nothing explicitly. The byte-level `get` / `try_get`
/// / iterators are unaffected and never consult it.
///
/// # Write hooks
///
/// Uses [`TypedWriteHook<K, T>`] via [`VarTypedHookAdapter`]. The hook receives
/// `&T` directly; the adapter decodes raw bytes via the codec. `on_write` fires
/// on `put`/`insert`/`delete`/`cas`/`compare_delete`/`update`/`fetch_update` and inside `atomic()`.
///
/// # Usage
///
/// ```ignore
/// let tree = VarTypedTree::<[u8; 16], Message, RapiraCodec>::open(
///     "data/messages",
///     Config::large_values().build(),
///     RapiraCodec,
/// )?;
/// tree.put(&key, &msg)?;
/// if let Some(m) = tree.get(&key) {
///     println!("{:?}", m);
/// }
/// tree.close()?;
/// ```
///
/// # Iteration
///
/// `iter()`, `range()`, and `prefix_iter()` return [`VarTypedIter`] with
/// `Item = (K, T)`. Each `next()` / `next_back()` may perform disk I/O on a
/// block-cache miss and performs a decode.
///
/// # 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 VarTypedTree<
    K: Key,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T> = NoHook,
> {
    inner: Arc<VarTree<K, VarTypedHookAdapter<K, T, C, H>>>,
    codec: C,
    decoded_cache: DecodedCache<T>,
    _marker: PhantomData<fn() -> T>,
}

impl<K: Key, T: Send + Sync, C: Codec<T> + Clone, H: TypedWriteHook<K, T>> watch::sealed::Spec
    for VarTypedTree<K, T, C, H>
{
}
impl<K: Key, T: Send + Sync, C: Codec<T> + Clone, H: TypedWriteHook<K, T>> WatchSpec
    for VarTypedTree<K, T, C, H>
{
    type Key = K;
    type Value = T;
    type Decoder = CodecWatchDecoder<C, T>;
}
impl<
    K: Key,
    T: Send + Sync,
    C: Codec<T> + Clone + Sync,
    H: TypedWriteHook<K, T> + watch::sealed::Enabled,
> VarTypedTree<K, T, C, H>
{
    /// Subscribe to this collection's mutations.
    ///
    /// Infallible and repeatable; every receiver is independent and owns a
    /// bounded ring. Delivery is **at-most-once** — a full ring drops the
    /// oldest event and reports [`WatchError::Lagged`] at the gap.
    ///
    /// The consumer recipe is `watch()` **first**, then scan, and rescan on
    /// `Lagged`. It is complete because of what this call guarantees on return:
    /// every write that decided not to publish to this receiver has already
    /// made its row visible.
    ///
    /// An event is a pointer, not a snapshot — a rebuild re-reads rows by key
    /// instead of applying payloads. There is no ordering across keys or
    /// collections, and transaction boundaries are invisible. A `new_only` hook
    /// reports every non-delete as [`WatchKind::Upsert`] — and, for this Var
    /// family, is what keeps a watched write from reading the old value.
    ///
    /// Subscriptions end with the database: `Db::shutdown` (and `Drop`) closes
    /// every channel; `Db::close` leaves receivers live. See `docs/watch.md`.
    pub fn watch(&self, options: WatchOptions) -> WatchReceiver<Self> {
        WatchReceiver::subscribe(
            self.inner.hook.watch_channel().expect("enabled watch hook"),
            CodecWatchDecoder::new(self.codec.clone()),
            options,
        )
    }
}

impl<K: Key, T: Send + Sync, C: Codec<T> + Clone> VarTypedTree<K, T, C> {
    /// Open or create a `VarTypedTree` 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, codec: C) -> DbResult<Self> {
        Self::open_hooked_inner(path, config, codec, NoHook)
    }
}

impl<K: Key, T: Send + Sync, C: Codec<T> + Clone, H: TypedWriteHook<K, T>>
    VarTypedTree<K, T, C, H>
{
    /// Open or create a `VarTypedTree` with a typed write hook.
    ///
    #[doc = include_str!("../docs/exclusive-access.md")]
    pub fn open_hooked(
        path: impl AsRef<std::path::Path>,
        config: Config,
        codec: C,
        hook: H,
    ) -> DbResult<Self> {
        crate::counter_dirty::check_no_counters(&hook)?;
        crate::unique::check_no_claims(&hook)?;
        #[cfg(feature = "armour")]
        if H::EMITS {
            return Err(DbError::ChangeStreamRequiresManagedOpen);
        }
        Self::open_hooked_inner(path, config, codec, hook)
    }

    /// Skips [`check_no_counters`](crate::counter_dirty::check_no_counters) —
    /// only `Db::open_managed` may call this, immediately followed by
    /// `ManagedOpen::bind_counter_dirty` before the collection is published.
    pub(crate) fn open_hooked_inner(
        path: impl AsRef<std::path::Path>,
        config: Config,
        codec: C,
        hook: H,
    ) -> DbResult<Self> {
        let adapter = VarTypedHookAdapter {
            inner: hook,
            codec: codec.clone(),
            _marker: PhantomData,
        };
        let decoded_cache = DecodedCache::new(&config.decoded_cache);
        if config.value_and_decoded_cache_both_on() {
            tracing::warn!(
                "VarTyped collection has both value_cache and decoded_cache enabled: every \
                 value-cache-routed value is held twice (raw bytes + decoded T). Consider \
                 value_cache: 0 when reads go through get_shared."
            );
        }
        let inner = Arc::new(VarTree::open_inner(path, config, adapter)?);
        Ok(Self {
            inner,
            codec,
            decoded_cache,
            _marker: PhantomData,
        })
    }

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

    pub fn close(self) -> DbResult<()> {
        self.clean_shutdown()
    }

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

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

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

    /// Trigger a compaction pass across all shards.
    pub fn compact(&self) -> DbResult<usize> {
        self.inner.compact()
    }

    /// Write hint files for all active shard files. Call during graceful shutdown.
    pub fn sync_hints(&self) -> DbResult<()> {
        self.inner.sync_hints()
    }

    /// Pre-populate the block cache with blocks containing live values.
    pub fn warmup(&self) -> DbResult<()> {
        self.inner.warmup()
    }

    /// Access the underlying `VarTree` for raw byte operations.
    pub fn as_inner(&self) -> &VarTree<K, VarTypedHookAdapter<K, T, C, H>> {
        &self.inner
    }

    #[cfg(test)]
    pub(crate) fn test_inner_arc(&self) -> Arc<VarTree<K, VarTypedHookAdapter<K, T, C, H>>> {
        Arc::clone(&self.inner)
    }

    /// Register the `Arc` handle the inner [`VarTree`] needs for transaction deferred replay.
    pub(crate) fn bind_tx_self(arc: &Arc<Self>) {
        VarTree::bind_tx_self(&arc.inner);
    }

    pub(crate) fn tx_install_projection_metadata(&self, db: &crate::armour::Db) -> DbResult<()> {
        self.as_inner().tx_install_projection_metadata(db)
    }

    pub(crate) fn tx_preplan_projection_targets(
        &self,
    ) -> DbResult<Vec<Arc<dyn crate::txn::participant::TxParticipant>>> {
        self.as_inner().tx_preplan_projection_targets()
    }

    pub(crate) fn tx_project_writes(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
        self.as_inner().tx_project_writes(key, old, new)
    }

    pub(crate) fn tx_counter_deltas_for(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
        self.as_inner().tx_counter_deltas_for(key, old, new)
    }

    /// Claim ops for one decided write inside a transaction (T10 Task 6) —
    /// `name` is passed straight through to [`VarTree::tx_claims_for`], which
    /// cannot name `T` itself.
    pub(crate) fn tx_claims_for(
        &self,
        name: &'static str,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::ClaimOp>> {
        self.as_inner().tx_claims_for(name, key, old, new)
    }

    /// Access the codec used for encoding / decoding values.
    pub fn codec(&self) -> &C {
        &self.codec
    }

    #[doc(hidden)]
    pub fn build_ttl_index(&self) -> DbResult<()> {
        self.inner.build_ttl_index()
    }

    pub fn sweep_expired(&self, now_ms: u64, budget: usize) -> DbResult<crate::ttl::SweepOutcome> {
        self.inner.sweep_expired(now_ms, budget)
    }

    pub fn set_ttl_sweep(&self, enabled: bool) -> DbResult<()> {
        self.inner.set_ttl_sweep(enabled)
    }

    pub fn rebuild_ttl_index(&self) -> DbResult<()> {
        self.inner.rebuild_ttl_index()
    }

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

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

    #[doc(hidden)]
    pub fn ttl_debug_record(&self, key: &K, deadline: Option<u64>) {
        self.inner.ttl_debug_record(key, deadline)
    }

    #[doc(hidden)]
    pub fn ttl_debug_inject_stale(&self, key: &K, deadline: u64) {
        self.inner.ttl_debug_inject_stale(key, deadline)
    }

    // -- Reads ----------------------------------------------------------------

    /// Get and decode a value by key.
    /// Returns `None` if the key is missing or decoding fails.
    ///
    /// **O(log n)** — lock-free SkipList lookup; brief shard lock + `pread` on block-cache
    /// miss, then decode.
    pub fn get(&self, key: &K) -> Option<T> {
        let bytes = self.inner.get(key)?;
        self.codec.decode_from(&bytes).ok()
    }

    /// Get a value by key, returning `Err(KeyNotFound)` if absent or
    /// `Err(CorruptedEntry)` if present but undecodable.
    pub fn get_or_err(&self, key: &K) -> DbResult<T> {
        self.try_get(key)?.ok_or(DbError::KeyNotFound)
    }

    /// Strict read: `Ok(None)` only when absent; `Err` when present but the value
    /// cannot be read or decoded.
    pub fn try_get(&self, key: &K) -> DbResult<Option<T>> {
        match self.inner.try_get(key)? {
            Some(bytes) => self.codec.decode_from(&bytes).map(Some),
            None => Ok(None),
        }
    }

    /// Get a value as a shared `Arc<T>`, decoding **only on a cache miss**.
    ///
    /// Lossy, like [`get`](Self::get): `None` when the key is missing, the value
    /// is unreadable, or it fails to decode. Works with the decoded cache
    /// disabled (`Config::decoded_cache.max_size == 0`) — it then decodes on
    /// every call, exactly like `get`, and wraps the result in an `Arc`.
    ///
    /// **Arc identity is opportunistic.** The returned value is the version the
    /// index pointed at when the read took its `DiskLoc`; that is the only
    /// invariant. Repeated calls are pointer-equal while the entry stays
    /// resident, but eviction, a concurrent cold miss (misses are deliberately
    /// not coalesced) and compaction each produce a new `Arc` with no write
    /// involved. A handle taken before a write keeps observing the old version.
    ///
    /// **O(log n)** — SkipList lookup, then a hash lookup on a hit; on a miss,
    /// the ordinary byte read plus one decode.
    pub fn get_shared(&self, key: &K) -> Option<Arc<T>> {
        self.try_get_shared(key).ok().flatten()
    }

    /// Strict shared read: `Ok(None)` only when the key is absent; `Err` when it
    /// is present but cannot be read or decoded. Shared twin of
    /// [`try_get`](Self::try_get); see [`get_shared`](Self::get_shared) for the
    /// `Arc` identity contract.
    pub fn try_get_shared(&self, key: &K) -> DbResult<Option<Arc<T>>> {
        self.inner
            .try_get_decoded(key, &self.decoded_cache, |b| self.codec.decode_from(b))
    }

    /// Batch shared point-read: one sorted finger-walk over the index, result
    /// aligned to `keys`. Lossy per element, like
    /// [`get_many`](Self::get_many); cached keys are not decoded again.
    pub fn get_many_shared(&self, keys: &[K]) -> Vec<Option<Arc<T>>> {
        self.inner
            .get_many_decoded(keys, &self.decoded_cache, |b| self.codec.decode_from(b))
    }

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

    /// Return the first entry in index order, or `None` if empty or every boundary
    /// entry is unproducible. Skips unreadable/undecodable entries, matching
    /// `iter().next()`.
    pub fn first(&self) -> Option<(K, T)> {
        self.iter().next()
    }

    /// Return the last entry in index order (symmetric to `first`).
    pub fn last(&self) -> Option<(K, T)> {
        self.iter().next_back()
    }

    /// Strict first entry: `Ok(None)` only when there is no live entry; `Err` when
    /// the boundary entry cannot be read or decoded.
    pub fn try_first(&self) -> DbResult<Option<(K, T)>> {
        match self.inner.try_first()? {
            Some((k, bytes)) => Ok(Some((k, self.codec.decode_from(&bytes)?))),
            None => Ok(None),
        }
    }

    /// Strict last entry (symmetric to `try_first`).
    pub fn try_last(&self) -> DbResult<Option<(K, T)>> {
        match self.inner.try_last()? {
            Some((k, bytes)) => Ok(Some((k, self.codec.decode_from(&bytes)?))),
            None => Ok(None),
        }
    }

    // -- Writes ---------------------------------------------------------------

    /// Insert or update a key-value pair.
    ///
    /// Returns `true` if a previous value for `key` existed (overwrite),
    /// `false` for a fresh insert.
    pub fn put(&self, key: &K, value: &T) -> DbResult<bool> {
        let expires_at = self.inner.hook.typed_deadline(key, value);
        let mut buf = Vec::new();
        self.codec.encode_to(value, &mut buf)?;
        self.inner.put_with_deadline(key, &buf, expires_at)
    }

    /// 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: &T) -> DbResult<()> {
        let mut buf = Vec::new();
        self.codec.encode_to(value, &mut buf)?;
        self.inner.insert(key, &buf)
    }

    /// Delete a key. Returns `true` if the key existed.
    pub fn delete(&self, key: &K) -> DbResult<bool> {
        self.inner.delete(key)
    }

    /// Compare-and-swap based on encoded bytes. Relies on deterministic codec output.
    /// Returns `Ok(())` on success, `Err(CasMismatch)` if current != expected,
    /// `Err(KeyNotFound)` if key doesn't exist.
    ///
    /// **Caveat:** inherits `VarTree::cas` behavior — holds the shard lock while
    /// reading the current value; block-cache miss causes disk I/O under the lock.
    pub fn cas(&self, key: &K, expected: &T, new_value: &T) -> DbResult<()> {
        let mut exp_buf = Vec::new();
        self.codec.encode_to(expected, &mut exp_buf)?;
        let mut new_buf = Vec::new();
        self.codec.encode_to(new_value, &mut new_buf)?;
        self.inner.cas(key, &exp_buf, &new_buf)
    }

    /// Compare-and-delete based on encoded bytes. Relies on deterministic codec output.
    /// Returns `Ok(())` on success, `Err(CasMismatch)` if current != expected,
    /// `Err(KeyNotFound)` if the key doesn't exist.
    pub fn compare_delete(&self, key: &K, expected: &T) -> DbResult<()> {
        let mut exp_buf = Vec::new();
        self.codec.encode_to(expected, &mut exp_buf)?;
        self.inner.compare_delete(key, &exp_buf)
    }

    /// Atomically read-modify-write. Returns `Some(new_value)` if key existed,
    /// `None` if the key is absent; `Err(CorruptedEntry)` if the current value
    /// cannot be decoded.
    ///
    /// **Caveat:** inherits `VarTree::update` behavior — shard lock held during
    /// possible disk I/O.
    pub fn update(&self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
        use std::cell::Cell;
        let out: Cell<Option<T>> = Cell::new(None);
        let result = self.inner.try_update_inner(
            key,
            |bytes| {
                let current = self.codec.decode_from(bytes)?;
                let new_val = f(&current);
                let mut buf = Vec::new();
                self.codec.encode_to(&new_val, &mut buf)?;
                out.set(Some(new_val));
                Ok(Some(ByteView::from_vec(buf)))
            },
            false,
        )?;
        if result.is_none() {
            return Ok(None);
        }
        Ok(out.into_inner())
    }

    /// Like [`update()`](Self::update), but returns `Some(old_value)` instead
    /// of the new one.
    pub fn fetch_update(&self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
        use std::cell::Cell;
        let out: Cell<Option<T>> = Cell::new(None);
        let result = self.inner.try_update_inner(
            key,
            |bytes| {
                let current = self.codec.decode_from(bytes)?;
                let new_val = f(&current);
                let mut buf = Vec::new();
                self.codec.encode_to(&new_val, &mut buf)?;
                out.set(Some(current));
                Ok(Some(ByteView::from_vec(buf)))
            },
            true,
        )?;
        if result.is_none() {
            return Ok(None);
        }
        Ok(out.into_inner())
    }

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

    // -- Atomic ---------------------------------------------------------------

    /// 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 VarTypedShard<'_, K, T, C, H>) -> DbResult<R>,
    ) -> DbResult<R> {
        self.inner.atomic(shard_key, |var_shard| {
            // SAFETY: erase VarShard lifetime via `*mut ()` — see VarTypedShard doc.
            // The pointer is valid for the duration of this closure.
            let inner_ptr: *mut () = (var_shard as *mut VarShard<'_, _, _>).cast();
            let mut shard = VarTypedShard {
                tree: self,
                inner: inner_ptr,
                _marker: PhantomData,
            };
            f(&mut shard)
        })
    }

    // -- Iteration ------------------------------------------------------------

    /// Iterate entries whose keys start with `prefix`.
    ///
    /// `reversed=true` (default): yields matching keys in DESC order.
    /// The delegated iterator has `next()` O(1); its first reverse seek is expected O(log n), and
    /// an m-row reverse walk is expected/amortized O(log n + m). Disk I/O on cache miss.
    /// Entries with decode errors are skipped.
    pub fn prefix_iter(&self, prefix: &[u8]) -> VarTypedIter<'_, K, T, C, H> {
        VarTypedIter {
            inner: self.inner.prefix_iter(prefix),
            codec: &self.codec,
            _marker: PhantomData,
        }
    }

    /// Iterate all entries in index order. Disk I/O on cache miss.
    /// The delegated iterator has `next()` O(1); its first reverse seek is expected O(log n), and
    /// an m-row reverse walk is expected/amortized O(log n + m).
    /// Entries with decode errors are skipped.
    pub fn iter(&self) -> VarTypedIter<'_, K, T, C, H> {
        VarTypedIter {
            inner: self.inner.iter(),
            codec: &self.codec,
            _marker: PhantomData,
        }
    }

    /// Iterate entries in `[start, end)` — start inclusive, end exclusive.
    /// The delegated iterator has `next()` O(1); its first reverse seek is expected O(log n), and
    /// an m-row reverse walk is expected/amortized O(log n + m). Disk I/O on cache miss.
    pub fn range(&self, start: &K, end: &K) -> VarTypedIter<'_, K, T, C, H> {
        VarTypedIter {
            inner: self.inner.range(start, end),
            codec: &self.codec,
            _marker: PhantomData,
        }
    }

    /// Iterate entries in range defined by `start` and `end` bounds.
    /// The delegated iterator has `next()` O(1); its first reverse seek is expected O(log n), and
    /// an m-row reverse walk is expected/amortized O(log n + m). Disk I/O on cache miss.
    pub fn range_bounds(&self, start: Bound<&K>, end: Bound<&K>) -> VarTypedIter<'_, K, T, C, H> {
        VarTypedIter {
            inner: self.inner.range_bounds(start, end),
            codec: &self.codec,
            _marker: PhantomData,
        }
    }

    /// Range iterator with explicit direction override. See [`VarTree::range_bounds_dir`].
    pub(crate) fn range_bounds_dir(
        &self,
        start: Bound<&K>,
        end: Bound<&K>,
        dir: Dir,
    ) -> VarTypedDirRangeIter<'_, K, T, C, H> {
        VarTypedDirRangeIter {
            inner: self.inner.range_bounds_dir(start, end, dir),
            codec: &self.codec,
            _marker: PhantomData,
        }
    }

    /// Key-only range iterator with explicit direction override.
    pub(crate) fn keys_range_bounds_dir(
        &self,
        start: Bound<&K>,
        end: Bound<&K>,
        dir: Dir,
    ) -> crate::skiplist::iter::SkipKeyKeys<'_, K, crate::skiplist::node::VarNode<K>> {
        self.inner.keys_range_bounds_dir(start, end, dir)
    }

    /// Whether index iteration yields descending key order.
    pub(crate) fn reversed(&self) -> bool {
        self.inner.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()))
    }

    /// Arc-owned prefix/range builder for [`crate::Tx::delete_scope`].
    #[cfg(feature = "armour")]
    pub fn tx_scope<const N: usize>(self: &Arc<Self>, key: &K) -> crate::TxScope<Self, K, N> {
        crate::txn::scope::TxScope::from_parts(
            self.clone(),
            key,
            dir_from(self.reversed()),
            Self::tx_scope_keys,
        )
    }

    fn tx_scope_keys<'a>(
        collection: &'a Self,
        start: Bound<&'a K>,
        end: Bound<&'a K>,
        dir: Dir,
    ) -> Box<dyn Iterator<Item = K> + 'a> {
        Box::new(collection.keys_range_bounds_dir(start, end, dir))
    }

    /// Strict range iterator: surfaces storage/decode errors instead of skipping.
    pub(crate) fn try_range_bounds(
        &self,
        start: Bound<&K>,
        end: Bound<&K>,
    ) -> TryVarTypedIter<'_, K, T, C, H> {
        TryVarTypedIter {
            inner: crate::scope::DirRangeIter::from_storage_match(
                self.inner.try_range_bounds(start, end),
                true,
            ),
            codec: &self.codec,
            failed: false,
        }
    }

    pub(crate) fn try_range_bounds_dir(
        &self,
        start: Bound<&K>,
        end: Bound<&K>,
        dir: Dir,
    ) -> TryVarTypedIter<'_, K, T, C, H> {
        TryVarTypedIter {
            inner: self.inner.try_range_bounds_dir(start, end, dir),
            codec: &self.codec,
            failed: false,
        }
    }

    // -- Info -----------------------------------------------------------------

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

    /// Strict live-key count for read-only inspection (key-only skiplist walk).
    pub(crate) fn try_count_live_keys(&self) -> DbResult<u64> {
        self.inner.try_count_live_keys()
    }

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

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

    /// 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> {
        self.inner.entry_len(key)
    }

    // -- Batch ----------------------------------------------------------------

    /// Batch point-read returning decoded values, aligned to `keys`. **Lossy** —
    /// like single-key [`get`](Self::get), a decode failure yields `None`.
    pub fn get_many(&self, keys: &[K]) -> Vec<Option<T>> {
        self.inner
            .get_many(keys)
            .into_iter()
            .map(|opt| opt.and_then(|b| self.codec.decode_from(&b).ok()))
            .collect()
    }

    /// Batch read-modify-write over decoded values. See [`crate::ConstTree::update_many`].
    ///
    /// **Strict on codec failures**, consistent with single-key [`update`](Self::update):
    /// a codec decode of an old value or encode of a new value that fails aborts the
    /// batch with that `DbError` (the applied prefix stays committed, as for any
    /// mid-batch error). Codec errors are never silently collapsed into
    /// `Keep`/`Kept`. (Read-only [`get_many`](Self::get_many) stays lossy, mirroring
    /// single-key [`get`](Self::get).)
    pub fn update_many<P>(
        &self,
        items: Vec<(K, P)>,
        f: impl Fn(&K, Option<&T>, &P) -> Upsert<T>,
    ) -> DbResult<Vec<(K, Applied<T>)>> {
        use std::cell::RefCell;

        // The inner (raw-bytes) closure cannot return a `Result`, so a codec
        // failure is captured here and surfaced after the batch unwinds. Once an
        // error is recorded, remaining items short-circuit to `Keep` so the batch
        // stops mutating; the recorded error is then returned.
        let codec_err: RefCell<Option<DbError>> = RefCell::new(None);
        let inner_out = self.inner.update_many(items, |k, old_bytes, p| {
            if codec_err.borrow().is_some() {
                return Upsert::Keep;
            }
            let old_t: Option<T> = match old_bytes {
                Some(b) => match self.codec.decode_from(b) {
                    Ok(t) => Some(t),
                    Err(e) => {
                        *codec_err.borrow_mut() = Some(e);
                        return Upsert::Keep;
                    }
                },
                None => None,
            };
            match f(k, old_t.as_ref(), p) {
                Upsert::Put(v) => {
                    let mut buf = Vec::new();
                    match self.codec.encode_to(&v, &mut buf) {
                        Ok(()) => Upsert::Put(ByteView::from_vec(buf)),
                        Err(e) => {
                            *codec_err.borrow_mut() = Some(e);
                            Upsert::Keep
                        }
                    }
                }
                Upsert::Keep => Upsert::Keep,
                Upsert::Delete => Upsert::Delete,
            }
        })?;
        if let Some(e) = codec_err.into_inner() {
            return Err(e);
        }
        // On the success path every old value already decoded inside the closure
        // and every `new` was just encoded here, so these decodes succeed; a
        // failure would be genuine corruption and is propagated, not swallowed.
        let mut out = Vec::with_capacity(inner_out.len());
        for (k, applied) in inner_out {
            let applied_t = match applied {
                Applied::Written { old, new } => Applied::Written {
                    old: match old {
                        Some(b) => Some(self.codec.decode_from(&b)?),
                        None => None,
                    },
                    new: self.codec.decode_from(&new)?,
                },
                Applied::Kept => Applied::Kept,
                Applied::Deleted(b) => Applied::Deleted(self.codec.decode_from(&b)?),
            };
            out.push((k, applied_t));
        }
        Ok(out)
    }

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

    /// Iterate all entries and optionally mutate them. Call once at startup.
    ///
    /// The callback receives each `(key, &T)` and returns `MigrateAction`:
    /// - `Keep` — no change
    /// - `Update(new_value)` — replace value (re-encoded via codec)
    /// - `Delete` — remove entry
    ///
    /// **Strict on codec failures**, like the underlying
    /// [`VarTree::migrate`](crate::VarTree::migrate) value read: an entry that
    /// fails to decode — or a migrated value that fails to re-encode — aborts the
    /// migration with that `DbError`. A migration must not report success while
    /// corrupt rows remain untouched on disk: the caller
    /// ([`run_migration`](crate::armour::Db) path) would commit a version bump
    /// over unmigrated data.
    ///
    /// **Failure is not clean.** Entries are rewritten as the scan walks, so an
    /// `Err` leaves the collection *partially* migrated with the version
    /// un-bumped; the next open re-runs the whole step, which is why migration
    /// callbacks must be idempotent (the standing `run_migration` contract). A
    /// single undecodable entry therefore blocks the collection's open on
    /// **every** subsequent start until it is repaired out-of-band — deliberate:
    /// silently carrying corrupt rows through a version bump is worse.
    ///
    /// Returns the number of mutated entries.
    pub fn migrate(&self, f: impl Fn(&K, &T) -> crate::MigrateAction<T>) -> DbResult<usize> {
        self.migrate_inner(f, true)
    }

    pub(crate) fn migrate_inner(
        &self,
        f: impl Fn(&K, &T) -> crate::MigrateAction<T>,
        fire_init: bool,
    ) -> DbResult<usize> {
        use crate::MigrateAction;
        use std::cell::RefCell;

        // The inner (raw-bytes) closure cannot return a `Result`, so a codec
        // failure is captured here and surfaced after the scan unwinds — same
        // shape as `update_many`. Once an error is recorded the remaining
        // entries short-circuit to `Keep`, so the scan stops mutating and the
        // outcome matches an abort at the failing entry.
        let codec_err: RefCell<Option<DbError>> = RefCell::new(None);
        let count = self.inner.migrate_inner(
            |key, bytes| {
                if codec_err.borrow().is_some() {
                    return MigrateAction::Keep;
                }
                let current = match self.codec.decode_from(bytes) {
                    Ok(v) => v,
                    Err(e) => {
                        *codec_err.borrow_mut() = Some(e);
                        return MigrateAction::Keep;
                    }
                };
                match f(key, &current) {
                    MigrateAction::Keep => MigrateAction::Keep,
                    MigrateAction::Update(new) => {
                        let mut buf = Vec::new();
                        match self.codec.encode_to(&new, &mut buf) {
                            Ok(()) => MigrateAction::Update(ByteView::from_vec(buf)),
                            Err(e) => {
                                *codec_err.borrow_mut() = Some(e);
                                MigrateAction::Keep
                            }
                        }
                    }
                    MigrateAction::Delete => MigrateAction::Delete,
                }
            },
            fire_init,
        )?;
        if let Some(e) = codec_err.into_inner() {
            tracing::error!(error = %e, "var_typed_tree migrate: codec failure — aborting migration step");
            return Err(e);
        }
        Ok(count)
    }

    /// Replay `on_init` for every live entry. Used by Db when no migration ran.
    pub(crate) fn replay_init(&self) {
        self.inner.replay_init();
    }

    pub(crate) fn reconcile_projections(&self) -> DbResult<()> {
        self.as_inner().reconcile_projections()
    }

    pub(crate) fn effective_projection_limits(&self) -> (usize, usize) {
        self.as_inner().effective_projection_limits()
    }
}

impl<K: Key, T: Send + Sync, C: Codec<T> + Clone, H: TypedWriteHook<K, T>> CompactionIndex<K>
    for VarTypedTree<K, T, C, H>
{
    fn update_if_match(&self, key: &K, old_loc: DiskLoc, new_loc: DiskLoc) -> bool {
        self.inner.update_if_match(key, old_loc, new_loc)
    }

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

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

    fn is_live(&self, shard_id: u8, key: &K, loc: DiskLoc) -> bool {
        self.inner.is_live(shard_id, key, loc)
    }
}

#[cfg(feature = "replication")]
impl<K: Key, T: Send + Sync, C: Codec<T> + Clone, H: TypedWriteHook<K, T>>
    crate::replication::ReplicationTarget for VarTypedTree<K, T, C, 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> {
        self.inner.apply_entry(
            shard_inner,
            shard_id,
            file_id,
            entry_offset,
            header,
            key,
            value,
        )
    }

    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> {
        self.inner.try_apply_entry(
            shard_inner,
            shard_id,
            file_id,
            entry_offset,
            header,
            raw_after_header,
        )
    }

    fn key_len(&self) -> usize {
        self.inner.key_len()
    }

    fn mark_replication_follower(&self) {
        self.inner.mark_replication_follower();
    }
}

// ---------------------------------------------------------------------------
// VarTypedIter
// ---------------------------------------------------------------------------

/// Iterator over entries in a [`VarTypedTree`]. Returned by `iter()`, `range()`,
/// `range_bounds()`, and `prefix_iter()`. Wraps [`VarIter`] and decodes each
/// value via the codec. Entries with decode errors are skipped (see `tracing::debug!`).
pub struct VarTypedIter<'a, K: Key, T: Send + Sync, C: Codec<T> + Clone, H: TypedWriteHook<K, T>> {
    inner: VarIter<'a, K, VarTypedHookAdapter<K, T, C, H>>,
    codec: &'a C,
    _marker: PhantomData<fn() -> T>,
}

impl<K: Key, T: Send + Sync, C: Codec<T> + Clone, H: TypedWriteHook<K, T>> Iterator
    for VarTypedIter<'_, K, T, C, H>
{
    type Item = (K, T);

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            let (k, bytes) = self.inner.next()?;
            match self.codec.decode_from(&bytes) {
                Ok(v) => return Some((k, v)),
                Err(_) => {
                    tracing::debug!(
                        value_len = bytes.len(),
                        "var_typed_iter: decode error, skipping entry"
                    );
                }
            }
        }
    }
}

impl<K: Key, T: Send + Sync, C: Codec<T> + Clone, H: TypedWriteHook<K, T>> DoubleEndedIterator
    for VarTypedIter<'_, K, T, C, H>
{
    fn next_back(&mut self) -> Option<Self::Item> {
        while let Some((k, bytes)) = self.inner.next_back() {
            match self.codec.decode_from(&bytes) {
                Ok(v) => return Some((k, v)),
                Err(_) => {
                    tracing::debug!(
                        value_len = bytes.len(),
                        "var_typed_iter: decode error, skipping entry"
                    );
                }
            }
        }
        None
    }
}

/// Direction-aware range iterator over a [`VarTypedTree`].
pub(crate) struct VarTypedDirRangeIter<
    'a,
    K: Key,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
> {
    inner: DirRangeIter<VarIter<'a, K, VarTypedHookAdapter<K, T, C, H>>>,
    codec: &'a C,
    _marker: PhantomData<fn() -> T>,
}

impl<K: Key, T: Send + Sync, C: Codec<T> + Clone, H: TypedWriteHook<K, T>> Iterator
    for VarTypedDirRangeIter<'_, K, T, C, H>
{
    type Item = (K, T);

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            let (k, bytes) = self.inner.next()?;
            match self.codec.decode_from(&bytes) {
                Ok(v) => return Some((k, v)),
                Err(_) => {
                    tracing::debug!(
                        value_len = bytes.len(),
                        "var_typed_iter: decode error, skipping entry"
                    );
                }
            }
        }
    }
}

/// Strict iterator over entries in a [`VarTypedTree`]. Wraps [`TryVarIter`] and
/// decodes each value via the codec, surfacing decode errors.
#[doc(hidden)]
pub struct TryVarTypedIter<'a, K, T, C, H>
where
    K: Key,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
{
    inner: crate::scope::DirRangeIter<TryVarIter<'a, K, VarTypedHookAdapter<K, T, C, H>>>,
    codec: &'a C,
    failed: bool,
}

impl<K, T, C, H> Iterator for TryVarTypedIter<'_, K, T, C, H>
where
    K: Key,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
{
    type Item = DbResult<(K, T)>;

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

// ---------------------------------------------------------------------------
// VarTypedShard
// ---------------------------------------------------------------------------

/// Handle for atomic multi-key operations within a single shard.
/// Obtained via [`VarTypedTree::atomic`]. The shard lock is held for the
/// lifetime of this struct — keep the closure short.
pub struct VarTypedShard<
    'tree,
    K: Key,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
> {
    tree: &'tree VarTypedTree<K, T, C, H>,
    // Type-erased raw pointer to a `VarShard<'_, K, VarTypedHookAdapter<K, T, C, H>>`.
    // Invariance of the inner `'_` lifetime would otherwise force it to `'static`
    // when stored. Only dereferenced inside the enclosing `atomic()` closure.
    inner: *mut (),
    _marker: PhantomData<&'tree mut ()>,
}

// SAFETY: raw pointer dereferenced only while the enclosing `atomic` closure
// is active; `VarShard` itself is `Send` and guards a mutex. `tree` is `Send+Sync`.
unsafe impl<K: Key, T: Send + Sync, C: Codec<T> + Clone, H: TypedWriteHook<K, T>> Send
    for VarTypedShard<'_, K, T, C, H>
{
}

impl<K: Key, T: Send + Sync, C: Codec<T> + Clone, H: TypedWriteHook<K, T>>
    VarTypedShard<'_, K, T, C, H>
{
    fn inner_mut(&mut self) -> &mut VarShard<'_, K, VarTypedHookAdapter<K, T, C, H>> {
        // SAFETY: pointer set in `VarTypedTree::atomic` from a valid `&mut VarShard`,
        // only dereferenced while the closure is live.
        unsafe { &mut *(self.inner as *mut VarShard<'_, K, VarTypedHookAdapter<K, T, C, H>>) }
    }

    fn inner_ref(&self) -> &VarShard<'_, K, VarTypedHookAdapter<K, T, C, H>> {
        // SAFETY: see `inner_mut`.
        unsafe { &*(self.inner as *const VarShard<'_, K, VarTypedHookAdapter<K, T, C, 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: &T) -> DbResult<bool> {
        let mut buf = Vec::new();
        self.tree.codec.encode_to(value, &mut buf)?;
        self.inner_mut().put(key, &buf)
    }

    pub fn insert(&mut self, key: &K, value: &T) -> DbResult<()> {
        let mut buf = Vec::new();
        self.tree.codec.encode_to(value, &mut buf)?;
        self.inner_mut().insert(key, &buf)
    }

    pub fn delete(&mut self, key: &K) -> DbResult<bool> {
        self.inner_mut().delete(key)
    }

    pub fn try_get(&self, key: &K) -> DbResult<Option<T>> {
        match self.inner_ref().try_get(key)? {
            Some(bytes) => self.tree.codec.decode_from(&bytes).map(Some),
            None => Ok(None),
        }
    }

    pub fn get_or_err(&self, key: &K) -> DbResult<T> {
        let bytes = self.inner_ref().get_or_err(key)?;
        self.tree.codec.decode_from(&bytes)
    }

    pub fn try_contains(&self, key: &K) -> DbResult<bool> {
        self.inner_ref().try_contains(key)
    }

    /// Atomically read-modify-write within the shard. Returns `Some(new_value)`
    /// if the key existed, `None` if absent.
    pub fn update(&mut self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
        use std::cell::Cell;
        let tree = self.tree; // &'tree VarTypedTree — Copy, avoids borrowing self in closure
        let out: Cell<Option<T>> = Cell::new(None);
        let result = self.inner_mut().update_inner(
            key,
            |bytes| {
                let current = tree.codec.decode_from(bytes)?;
                let new_val = f(&current);
                let mut buf = Vec::new();
                tree.codec.encode_to(&new_val, &mut buf)?;
                out.set(Some(new_val));
                Ok(Some(ByteView::from_vec(buf)))
            },
            false,
        )?;
        if result.is_none() {
            return Ok(None);
        }
        Ok(out.into_inner())
    }

    /// 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(&T) -> T) -> DbResult<Option<T>> {
        use std::cell::Cell;
        let tree = self.tree; // Copy
        let out: Cell<Option<T>> = Cell::new(None);
        let result = self.inner_mut().update_inner(
            key,
            |bytes| {
                let current = tree.codec.decode_from(bytes)?;
                let new_val = f(&current);
                let mut buf = Vec::new();
                tree.codec.encode_to(&new_val, &mut buf)?;
                out.set(Some(current)); // capture the OLD value
                Ok(Some(ByteView::from_vec(buf)))
            },
            true,
        )?;
        if result.is_none() {
            return Ok(None);
        }
        Ok(out.into_inner())
    }

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

#[cfg(feature = "armour")]
impl<T, C, H> crate::armour::collection::Collection for VarTypedTree<T::SelfId, T, C, H>
where
    T: crate::CollectionMeta + Send + Sync,
    C: Codec<T> + Clone + 'static,
    H: TypedWriteHook<T::SelfId, T>,
    T::SelfId: crate::Key + Ord,
{
    fn name(&self) -> &str {
        T::NAME
    }
    fn len(&self) -> usize {
        self.len()
    }
    fn compact(&self) -> DbResult<usize> {
        self.compact()
    }
    fn flush(&self) -> DbResult<()> {
        crate::projection::collection_flush_fault()?;
        self.flush_buffers()?;
        self.sync_hints()?;
        Ok(())
    }
    fn periodic_flush(&self) -> DbResult<()> {
        self.flush_buffers()
    }
    fn durable_flush(&self) -> DbResult<()> {
        VarTypedTree::durable_flush(self)
    }
    fn clean_shutdown(&self) -> DbResult<()> {
        VarTypedTree::clean_shutdown(self)
    }
    fn sweep_expired(&self, now_ms: u64, budget: usize) -> DbResult<crate::ttl::SweepOutcome> {
        self.sweep_expired(now_ms, budget)
    }
    fn set_ttl_sweep(&self, enabled: bool) -> DbResult<()> {
        self.set_ttl_sweep(enabled)
    }
    fn rebuild_ttl_index(&self) -> DbResult<()> {
        self.rebuild_ttl_index()
    }
    fn reset_counter_dirty_marker(&self) -> bool {
        self.as_inner()
            .counter_dirty
            .get()
            .is_some_and(|d| d.take_marked())
    }
    fn reconcile_counters(&self) -> DbResult<u64> {
        self.as_inner().reconcile_counters()
    }
    fn counter_target_names(&self) -> Vec<&'static str> {
        self.as_inner().counter_target_names()
    }
    fn close_watches(&self) {
        if let Some(channel) = self.inner.hook.watch_channel() {
            channel.close_terminal();
        }
    }
}

#[cfg(feature = "typed-tree")]
impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    T: armour_core::GetType + crate::CollectionMeta + Clone + PartialEq + Send + Sync + 'static,
    C: Codec<T> + Clone + Sync + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::projection::sealed::Sealed for VarTypedTree<K, T, C, H>
{
}

impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    T: armour_core::GetType + crate::CollectionMeta + Clone + PartialEq + Send + Sync + 'static,
    C: Codec<T> + Clone + Sync + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::projection::ProjectableIndex for VarTypedTree<K, T, C, H>
{
    fn collection_name(&self) -> &'static str {
        T::NAME
    }

    fn put_bytes(&self, key: &[u8], value: &[u8]) -> DbResult<()> {
        if key.len() != std::mem::size_of::<K>() {
            return Err(DbError::CorruptedEntry { offset: 0 });
        }
        let k = K::from_bytes(key);
        let v = self.codec.decode_from(value)?;
        self.put(&k, &v)?;
        Ok(())
    }

    fn delete_bytes(&self, key: &[u8]) -> DbResult<()> {
        if key.len() != std::mem::size_of::<K>() {
            return Err(DbError::CorruptedEntry { offset: 0 });
        }
        let k = K::from_bytes(key);
        self.delete(&k)?;
        Ok(())
    }

    fn values_equal(&self, a: &[u8], b: &[u8]) -> DbResult<bool> {
        if a == b {
            return Ok(true);
        }
        let va = self.codec.decode_from(a)?;
        let vb = self.codec.decode_from(b)?;
        Ok(va == vb)
    }

    fn for_each_bytes(&self, f: &mut dyn FnMut(&[u8], &[u8]) -> DbResult<()>) -> DbResult<()> {
        let mut err: Option<DbError> = None;
        for item in self.iter() {
            if err.is_some() {
                break;
            }
            let (k, v) = item;
            let mut buf = Vec::new();
            match self
                .codec
                .encode_to(&v, &mut buf)
                .and_then(|()| f(k.as_bytes(), &buf))
            {
                Ok(()) => {}
                Err(e) => err = Some(e),
            }
        }
        match err {
            Some(e) => Err(e),
            None => Ok(()),
        }
    }

    fn durable_flush(&self) -> DbResult<()> {
        VarTypedTree::durable_flush(self)
    }
}

#[cfg(feature = "typed-tree")]
impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    T: armour_core::GetType + crate::CollectionMeta + Clone + PartialEq + Send + Sync + 'static,
    C: Codec<T> + Clone + Sync + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::projection::ProjectableIndexTx for VarTypedTree<K, T, C, H>
{
    fn as_tx_participant(self: Arc<Self>) -> Arc<dyn crate::txn::participant::TxParticipant> {
        self
    }
}

#[cfg(feature = "typed-tree")]
impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    T: armour_core::GetType + crate::CollectionMeta + Clone + PartialEq + Send + Sync + 'static,
    C: Codec<T> + Clone + Sync + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::projection::IndexKeyed for VarTypedTree<K, T, C, H>
{
    type K = K;
    type V = T;

    fn encode_value(&self, v: &T) -> DbResult<Vec<u8>> {
        let mut buf = Vec::new();
        self.codec.encode_to(v, &mut buf)?;
        Ok(buf)
    }
}

#[cfg(feature = "typed-tree")]
impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    T: armour_core::GetType + crate::CollectionMeta + Clone + PartialEq + Send + Sync + 'static,
    C: Codec<T> + Clone + Sync + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::counter::CounterCollection for VarTypedTree<K, T, C, H>
{
    fn decode_value(&self, bytes: &[u8]) -> DbResult<T> {
        self.codec.decode_from(bytes)
    }

    /// Read path — no shard write lock: `try_get` on the raw byte tree is a
    /// strict, lock-free-on-hit read; unlike `get`/`try_get` on `Self` it
    /// returns the stored bytes directly instead of decode-then-reencode.
    fn read_bytes(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
        if key.len() != std::mem::size_of::<K>() {
            return Err(DbError::CorruptedEntry { offset: 0 });
        }
        let k = K::from_bytes(key);
        Ok(self.inner.try_get(&k)?.map(|bv| bv.as_bytes().to_vec()))
    }

    fn contains_bytes(&self, key: &[u8]) -> DbResult<bool> {
        if key.len() != std::mem::size_of::<K>() {
            return Err(DbError::CorruptedEntry { offset: 0 });
        }
        let k = K::from_bytes(key);
        Ok(self.contains(&k))
    }

    fn upsert_counter(
        &self,
        key: &[u8],
        f: &mut dyn FnMut(Option<&T>) -> Upsert<T>,
    ) -> DbResult<()> {
        if key.len() != std::mem::size_of::<K>() {
            return Err(DbError::CorruptedEntry { offset: 0 });
        }
        let k = K::from_bytes(key);
        self.upsert(&k, f)?;
        Ok(())
    }
}

#[cfg(test)]
#[cfg(feature = "rapira-codec")]
mod tests {
    use super::*;
    use crate::codec::{Codec, RapiraCodec};
    use crate::config::{CacheConfig, Config};
    use crate::hook::TypedWriteHook;
    use rapira::Rapira;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};
    use tempfile::tempdir;

    #[derive(Debug, Clone, PartialEq, Rapira)]
    struct Msg {
        id: u64,
        body: String,
        tags: Vec<String>,
    }

    type MsgTree = VarTypedTree<[u8; 16], Msg, RapiraCodec>;

    fn open(dir: &std::path::Path) -> MsgTree {
        VarTypedTree::open(dir, Config::test(), RapiraCodec).unwrap()
    }

    fn msg(id: u64) -> Msg {
        Msg {
            id,
            body: format!("hello from message #{id}"),
            tags: vec![format!("t{id}"), "bench".into()],
        }
    }

    /// Codec that counts decodes, so tests can assert "a hit does not decode"
    /// directly instead of inferring it from `Arc` identity.
    #[derive(Clone)]
    struct CountingCodec {
        decodes: Arc<AtomicUsize>,
    }

    impl Codec<Msg> for CountingCodec {
        fn encode_to(&self, value: &Msg, buf: &mut Vec<u8>) -> DbResult<()> {
            RapiraCodec.encode_to(value, buf)
        }
        fn decode_from(&self, bytes: &[u8]) -> DbResult<Msg> {
            self.decodes.fetch_add(1, AtomicOrdering::Relaxed);
            RapiraCodec.decode_from(bytes)
        }
    }

    fn open_counting(
        dir: &std::path::Path,
        decoded_cache_bytes: u64,
    ) -> (VarTypedTree<[u8; 16], Msg, CountingCodec>, Arc<AtomicUsize>) {
        let decodes = Arc::new(AtomicUsize::new(0));
        let mut cfg = Config::test();
        cfg.decoded_cache = CacheConfig {
            max_size: decoded_cache_bytes,
            estimated_items: 64,
        };
        let tree = VarTypedTree::open(
            dir,
            cfg,
            CountingCodec {
                decodes: Arc::clone(&decodes),
            },
        )
        .unwrap();
        (tree, decodes)
    }

    #[test]
    fn get_shared_hit_does_not_decode() {
        let dir = tempdir().unwrap();
        let (tree, decodes) = open_counting(dir.path(), 1 << 20);
        let k = [1u8; 16];
        tree.put(&k, &msg(1)).unwrap();
        let after_write = decodes.load(AtomicOrdering::Relaxed);

        let a = tree.get_shared(&k).expect("present");
        let b = tree.get_shared(&k).expect("present");
        assert_eq!(*a, msg(1));
        assert_eq!(*b, msg(1));
        assert_eq!(
            decodes.load(AtomicOrdering::Relaxed) - after_write,
            1,
            "the second get_shared must be served from the decoded cache"
        );
        assert!(Arc::ptr_eq(&a, &b)); // consequence of residency, not a contract
    }

    #[test]
    fn get_shared_works_with_the_cache_disabled() {
        let dir = tempdir().unwrap();
        let (tree, decodes) = open_counting(dir.path(), 0);
        let k = [2u8; 16];
        tree.put(&k, &msg(2)).unwrap();
        let after_write = decodes.load(AtomicOrdering::Relaxed);

        for _ in 0..3 {
            assert_eq!(*tree.get_shared(&k).expect("present"), msg(2));
        }
        assert_eq!(
            decodes.load(AtomicOrdering::Relaxed) - after_write,
            3,
            "a disabled cache must still serve correct values, decoding each time"
        );
    }

    #[test]
    fn shared_reads_absent_key() {
        let dir = tempdir().unwrap();
        let (tree, _) = open_counting(dir.path(), 1 << 20);
        let k = [3u8; 16];
        assert!(tree.get_shared(&k).is_none());
        assert!(
            tree.try_get_shared(&k)
                .expect("absent is Ok(None)")
                .is_none()
        );
    }

    #[test]
    fn shared_reads_undecodable_value() {
        let dir = tempdir().unwrap();
        let (tree, _) = open_counting(dir.path(), 1 << 20);
        let k = [4u8; 16];
        // Write raw bytes that the codec cannot decode, bypassing the wrapper.
        tree.as_inner().put(&k, &[0xffu8; 8]).unwrap();

        assert!(tree.get_shared(&k).is_none(), "lossy: undecodable -> None");
        assert!(
            tree.try_get_shared(&k).is_err(),
            "strict: undecodable -> Err"
        );
    }

    #[test]
    fn get_many_shared_is_aligned_and_lossy() {
        let dir = tempdir().unwrap();
        let (tree, decodes) = open_counting(dir.path(), 1 << 20);
        let keys = [[10u8; 16], [11u8; 16], [12u8; 16]];
        tree.put(&keys[0], &msg(10)).unwrap();
        tree.put(&keys[2], &msg(12)).unwrap();
        let after_writes = decodes.load(AtomicOrdering::Relaxed);

        let out = tree.get_many_shared(&keys);
        assert_eq!(out.len(), 3);
        assert_eq!(out[0].as_deref(), Some(&msg(10)));
        assert!(out[1].is_none(), "absent key is None at its own index");
        assert_eq!(out[2].as_deref(), Some(&msg(12)));
        assert_eq!(decodes.load(AtomicOrdering::Relaxed) - after_writes, 2);

        let again = tree.get_many_shared(&keys);
        assert_eq!(again[0].as_deref(), Some(&msg(10)));
        assert_eq!(
            decodes.load(AtomicOrdering::Relaxed) - after_writes,
            2,
            "a cached batch must not decode again"
        );
    }

    #[test]
    fn shared_reads_do_not_require_clone_on_t() {
        // `T` here is deliberately not `Clone`: this test fails to compile if
        // `#[derive(Clone)]` ever replaces the manual `Clone for DecodedEntry<T>`.
        struct NotClone {
            body: Vec<u8>,
        }
        #[derive(Clone)]
        struct RawCodec;
        impl Codec<NotClone> for RawCodec {
            fn encode_to(&self, value: &NotClone, buf: &mut Vec<u8>) -> DbResult<()> {
                buf.clear();
                buf.extend_from_slice(&value.body);
                Ok(())
            }
            fn decode_from(&self, bytes: &[u8]) -> DbResult<NotClone> {
                Ok(NotClone {
                    body: bytes.to_vec(),
                })
            }
        }

        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.decoded_cache = CacheConfig {
            max_size: 1 << 20,
            estimated_items: 8,
        };
        let tree: VarTypedTree<[u8; 16], NotClone, RawCodec> =
            VarTypedTree::open(dir.path(), cfg, RawCodec).unwrap();

        let k = [5u8; 16];
        tree.put(
            &k,
            &NotClone {
                body: vec![1, 2, 3],
            },
        )
        .unwrap();
        assert_eq!(tree.get_shared(&k).expect("present").body, vec![1, 2, 3]);
        assert_eq!(
            tree.try_get_shared(&k).expect("ok").expect("present").body,
            vec![1, 2, 3]
        );
        assert_eq!(
            tree.get_many_shared(&[k])[0].as_ref().unwrap().body,
            vec![1, 2, 3]
        );
    }

    #[test]
    fn get_shared_absent_after_delete_redecodes_on_reinsert() {
        let dir = tempdir().unwrap();
        let (tree, decodes) = open_counting(dir.path(), 1 << 20);
        let k = [7u8; 16];
        tree.put(&k, &msg(7)).unwrap();

        assert_eq!(*tree.get_shared(&k).expect("present"), msg(7));
        assert_eq!(decodes.load(AtomicOrdering::Relaxed), 1);

        assert!(tree.delete(&k).unwrap());
        assert!(tree.get_shared(&k).is_none());
        assert!(
            tree.try_get_shared(&k)
                .expect("absent is Ok(None)")
                .is_none()
        );

        tree.put(&k, &msg(77)).unwrap();
        let before = decodes.load(AtomicOrdering::Relaxed);
        assert_eq!(*tree.get_shared(&k).expect("reinserted"), msg(77));
        assert_eq!(
            decodes.load(AtomicOrdering::Relaxed) - before,
            1,
            "re-insert must decode again"
        );
    }

    #[test]
    fn get_shared_evicts_under_pressure_and_redecodes() {
        fn big_msg(id: u64, body_len: usize) -> Msg {
            Msg {
                id,
                body: "x".repeat(body_len),
                tags: vec![],
            }
        }

        // Measure the on-disk weight the decoded cache actually charges (`DiskLoc::len`).
        let dir_probe = tempdir().unwrap();
        let (probe, probe_decodes) = open_counting(dir_probe.path(), 1 << 20);
        let body_len = 512;
        let k_probe = [99u8; 16];
        let sample = big_msg(99, body_len);
        probe.put(&k_probe, &sample).unwrap();
        let weight = u64::from(probe.entry_len(&k_probe).expect("on-disk len"));
        assert!(weight > 0, "entry must have non-zero on-disk weight");

        // Control, on the probe's roomy cache: a resident entry is served without
        // decoding. Without it every assertion below would also hold for a cache
        // that never hits at all — three reads, three decodes — so this is what
        // makes the re-decode further down evidence of *eviction*. It cannot run
        // on the one-entry budget instead: there the policy does not keep even
        // the most recent insert resident.
        probe.get_shared(&k_probe).expect("present");
        probe.get_shared(&k_probe).expect("present");
        assert_eq!(
            probe_decodes.load(AtomicOrdering::Relaxed),
            1,
            "control: a resident entry must be served without decoding"
        );

        let dir = tempdir().unwrap();
        // Budget holds exactly one decoded entry; warming a second key evicts the first.
        let (tree, decodes) = open_counting(dir.path(), weight);
        let k0 = [0u8; 16];
        let k1 = [1u8; 16];
        tree.put(&k0, &big_msg(0, body_len)).unwrap();
        tree.put(&k1, &big_msg(1, body_len)).unwrap();
        assert_eq!(
            tree.entry_len(&k0),
            tree.entry_len(&k1),
            "same-shaped payloads must weigh the same in the decoded cache"
        );

        tree.get_shared(&k0).expect("present");
        assert_eq!(decodes.load(AtomicOrdering::Relaxed), 1);

        let before_k1 = decodes.load(AtomicOrdering::Relaxed);
        tree.get_shared(&k1).expect("present");
        assert_eq!(decodes.load(AtomicOrdering::Relaxed) - before_k1, 1);

        let before_reread = decodes.load(AtomicOrdering::Relaxed);
        tree.get_shared(&k0)
            .expect("evicted entry must still read correctly");
        assert_eq!(
            decodes.load(AtomicOrdering::Relaxed) - before_reread,
            1,
            "an evicted entry must re-decode (weight={weight}, budget={weight})"
        );
    }

    #[test]
    fn concurrent_get_shared_on_a_cold_key_is_correct() {
        let dir = tempdir().unwrap();
        let (tree, _) = open_counting(dir.path(), 1 << 20);
        let k = [6u8; 16];
        tree.put(&k, &msg(6)).unwrap();
        let tree = Arc::new(tree);

        // No assertion on decode count or Arc identity: without coalescing,
        // concurrent cold misses may each decode (spec, "Alternatives considered").
        std::thread::scope(|s| {
            for _ in 0..4 {
                let tree = Arc::clone(&tree);
                s.spawn(move || {
                    for _ in 0..32 {
                        assert_eq!(*tree.get_shared(&k).expect("present"), msg(6));
                    }
                });
            }
        });
    }

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

        let k = [1u8; 16];
        let v = msg(1);
        let other = msg(99);
        tree.put(&k, &v).unwrap();

        assert!(matches!(
            tree.compare_delete(&k, &other),
            Err(DbError::CasMismatch)
        ));
        assert_eq!(tree.get(&k), Some(v.clone()));

        assert!(tree.compare_delete(&k, &v).is_ok());
        assert!(tree.get(&k).is_none());

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

    #[test]
    fn entry_len_returns_some_for_existing_key() {
        let tmp = tempdir().expect("tmp");
        let tree: VarTypedTree<[u8; 8], Vec<u8>, RapiraCodec> =
            VarTypedTree::open(tmp.path(), Config::test(), RapiraCodec).expect("open");
        let key = 1u64.to_be_bytes();
        tree.put(&key, &vec![10u8, 20, 30, 40]).expect("put");
        // RapiraCodec encodes Vec<u8> as 4-byte length prefix + data bytes.
        // vec![10, 20, 30, 40] → 4 + 4 = 8 encoded bytes.
        assert_eq!(tree.entry_len(&key), Some(8u32));
    }

    #[test]
    fn entry_len_returns_none_for_missing_key() {
        let tmp = tempdir().expect("tmp");
        let tree: VarTypedTree<[u8; 8], Vec<u8>, RapiraCodec> =
            VarTypedTree::open(tmp.path(), Config::test(), RapiraCodec).expect("open");
        let key = 99u64.to_be_bytes();
        assert_eq!(tree.entry_len(&key), None);
    }

    #[test]
    fn try_range_bounds_surfaces_decode_error_without_skipping() {
        use crate::error::{DbError, DbResult};
        use crate::test_faults::{POISON, PoisonCodec};
        use std::ops::Bound;

        let dir = tempdir().unwrap();
        let tree: VarTypedTree<[u8; 8], u64, PoisonCodec> =
            VarTypedTree::open(dir.path(), Config::test(), PoisonCodec).unwrap();
        tree.put(&1_u64.to_be_bytes(), &1).unwrap();
        tree.put(&2_u64.to_be_bytes(), &POISON).unwrap();
        tree.put(&3_u64.to_be_bytes(), &3).unwrap();

        let rows = tree
            .try_range_bounds(Bound::Unbounded, Bound::Unbounded)
            .collect::<DbResult<Vec<_>>>();

        assert!(matches!(rows, Err(DbError::CorruptedEntry { .. })));
        assert_eq!(
            tree.iter().count(),
            2,
            "the existing iterator remains lossy"
        );
    }

    #[test]
    fn try_range_bounds_reversed_direction_fails_at_poison_from_opposite_edge() {
        use crate::error::DbError;
        use crate::test_faults::{POISON, PoisonCodec};
        use std::ops::Bound;

        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.reversed = false;
        let tree: VarTypedTree<[u8; 8], u64, PoisonCodec> =
            VarTypedTree::open(dir.path(), cfg, PoisonCodec).unwrap();
        tree.put(&1_u64.to_be_bytes(), &1).unwrap();
        tree.put(&2_u64.to_be_bytes(), &POISON).unwrap();
        tree.put(&3_u64.to_be_bytes(), &3).unwrap();

        let k1 = 1_u64.to_be_bytes();
        let mut iter = tree.try_range_bounds(Bound::Unbounded, Bound::Unbounded);
        match iter.next() {
            Some(Ok((k, v))) => {
                assert_eq!(k, k1);
                assert_eq!(v, 1);
            }
            other => panic!("expected Ok((k1, 1)), got {other:?}"),
        }
        assert!(matches!(
            iter.next(),
            Some(Err(DbError::CorruptedEntry { .. }))
        ));
        assert!(iter.next().is_none());
    }

    #[test]
    fn typed_decode_fault_surfaced_on_result_methods_raw_bytes_on_cas() {
        use crate::test_faults::{POISON, PoisonCodec};
        let dir = tempdir().unwrap();
        let tree: VarTypedTree<[u8; 8], u64, PoisonCodec> =
            VarTypedTree::open(dir.path(), Config::test(), PoisonCodec).unwrap();

        // Default reversed=true => first() is the largest key. Put a clean entry and
        // a poisoned boundary entry (key 2 > key 1).
        let k1 = 1u64.to_be_bytes();
        let k2 = 2u64.to_be_bytes();
        tree.put(&k1, &7u64).unwrap();
        tree.put(&k2, &POISON).unwrap();

        // Result methods surface CorruptedEntry, not KeyNotFound / Ok(None).
        assert!(matches!(
            tree.get_or_err(&k2),
            Err(DbError::CorruptedEntry { .. })
        ));
        assert!(matches!(
            tree.try_get(&k2),
            Err(DbError::CorruptedEntry { .. })
        ));
        assert!(matches!(
            tree.update(&k2, |v| v + 1),
            Err(DbError::CorruptedEntry { .. })
        ));
        assert!(matches!(
            tree.fetch_update(&k2, |v| v + 1),
            Err(DbError::CorruptedEntry { .. })
        ));

        // Lossy first() skips the poisoned boundary and returns the next valid entry,
        // matching iter().next().
        assert_eq!(tree.first(), Some((k1, 7u64)));
        assert_eq!(tree.first(), tree.iter().next());
        // Strict try_first surfaces the boundary decode error.
        assert!(matches!(
            tree.try_first(),
            Err(DbError::CorruptedEntry { .. })
        ));

        // cas/compare_delete compare RAW bytes — a decodable-but-poison value is not
        // a read error: mismatch on a different expected, success on the exact bytes.
        assert!(matches!(
            tree.cas(&k2, &123u64, &5u64),
            Err(DbError::CasMismatch)
        ));
        assert!(tree.cas(&k2, &POISON, &5u64).is_ok());
    }

    #[test]
    fn typed_shard_get_or_err_surfaces_decode_fault() {
        use crate::test_faults::{POISON, PoisonCodec};
        let dir = tempdir().unwrap();
        let tree: VarTypedTree<[u8; 8], u64, PoisonCodec> =
            VarTypedTree::open(dir.path(), Config::test(), PoisonCodec).unwrap();

        let k = 1u64.to_be_bytes();
        tree.put(&k, &POISON).unwrap();

        let got = tree.atomic(&k, |shard| shard.get_or_err(&k));
        assert!(matches!(got, Err(DbError::CorruptedEntry { .. })));

        let absent = 999u64.to_be_bytes();
        let got_absent = tree.atomic(&absent, |shard| shard.get_or_err(&absent));
        assert!(matches!(got_absent, Err(DbError::KeyNotFound)));
    }

    #[test]
    fn var_typed_try_get_surfaces_undecodable_rapira_as_err() {
        // A stored value whose bytes are cursor-valid but semantically invalid
        // (here: a `String` field with invalid UTF-8) must surface as
        // Err(CorruptedEntry) on the strict read — never UB / panic / abort.
        // Under the old unchecked codec this constructed an invalid `String`.
        let dir = tempdir().unwrap();
        let tree = open(dir.path());
        let key = [0u8; 16];

        // Msg layout: id:u64 | body:String(u32 len + utf8) | tags:Vec<String>(u32 count).
        // body has len=2 but bytes 0xFF 0xFE (invalid UTF-8); tags empty.
        let bad: &[u8] = &[
            0, 0, 0, 0, 0, 0, 0, 0, // id: u64 = 0
            2, 0, 0, 0, // body String length = 2
            0xFF, 0xFE, // body content — invalid UTF-8
            0, 0, 0, 0, // tags Vec<String> length = 0
        ];
        // NoHook (default) → the write-path adapter does not decode; raw bytes land on disk.
        tree.as_inner().put(&key, bad).unwrap();

        let res = tree.try_get(&key);
        assert!(
            matches!(res, Err(DbError::CorruptedEntry { .. })),
            "undecodable stored value must surface as Err, got {res:?}"
        );
    }

    #[test]
    fn var_typed_tree_migrate_errors_on_undecodable_row() {
        use crate::MigrateAction;
        let dir = tempdir().unwrap();
        let tree = open(dir.path());
        tree.put(&[1u8; 16], &msg(1)).unwrap();
        // Inject a value that cannot decode as `Msg` (same layout as the
        // `try_get` fault test: `body` claims 2 bytes of invalid UTF-8).
        let bad: &[u8] = &[
            0, 0, 0, 0, 0, 0, 0, 0, // id: u64 = 0
            2, 0, 0, 0, // body String length = 2
            0xFF, 0xFE, // body content — invalid UTF-8
            0, 0, 0, 0, // tags Vec<String> length = 0
        ];
        tree.as_inner().put(&[2u8; 16], bad).unwrap();

        // A migration that rewrites every row must surface the corrupt row as an
        // error, not skip it and report success (which would let `run_migration`
        // commit a version bump with unmigrated rows on disk).
        let err = tree
            .migrate(|_k, _v| MigrateAction::Update(msg(9)))
            .unwrap_err();
        assert!(matches!(err, DbError::CorruptedEntry { .. }), "{err:?}");
    }

    #[test]
    fn var_typed_tree_migrate_errors_on_unencodable_value() {
        use crate::MigrateAction;
        let dir = tempdir().unwrap();
        let tree: VarTypedTree<[u8; 8], u64, EncodeFailCodec> =
            VarTypedTree::open(dir.path(), Config::test(), EncodeFailCodec).unwrap();
        tree.put(&1u64.to_be_bytes(), &7u64).unwrap();

        // A migrated value that will not re-encode is a real failure: the row
        // would stay at its old contents while the version moves on.
        let err = tree
            .migrate(|_k, _v| MigrateAction::Update(ENCODE_FAIL))
            .unwrap_err();
        assert!(matches!(err, DbError::CorruptedEntry { .. }), "{err:?}");
    }

    /// Sentinel that [`EncodeFailCodec`] refuses to encode.
    const ENCODE_FAIL: u64 = 0xBAD0_BAD0_BAD0_BAD0;

    /// Mirror of `test_faults::PoisonCodec`: decodes anything 8 bytes wide but
    /// fails to *encode* the [`ENCODE_FAIL`] sentinel.
    #[derive(Clone)]
    struct EncodeFailCodec;

    impl Codec<u64> for EncodeFailCodec {
        fn encode_to(&self, value: &u64, buf: &mut Vec<u8>) -> DbResult<()> {
            if *value == ENCODE_FAIL {
                return Err(DbError::CorruptedEntry { offset: 0 });
            }
            buf.clear();
            buf.extend_from_slice(&value.to_be_bytes());
            Ok(())
        }

        fn decode_from(&self, bytes: &[u8]) -> DbResult<u64> {
            let arr: [u8; 8] = bytes
                .try_into()
                .map_err(|_| DbError::CorruptedEntry { offset: 0 })?;
            Ok(u64::from_be_bytes(arr))
        }
    }

    #[derive(Clone)]
    struct U64Codec;

    impl Codec<u64> for U64Codec {
        fn encode_to(&self, value: &u64, buf: &mut Vec<u8>) -> DbResult<()> {
            buf.clear();
            buf.extend_from_slice(&value.to_be_bytes());
            Ok(())
        }

        fn decode_from(&self, bytes: &[u8]) -> DbResult<u64> {
            let arr: [u8; 8] = bytes
                .try_into()
                .map_err(|_| DbError::CorruptedEntry { offset: 0 })?;
            Ok(u64::from_be_bytes(arr))
        }
    }

    #[derive(Default)]
    struct TRecHookState {
        writes: AtomicUsize,
        last_new: crate::sync::Mutex<Option<u64>>,
    }

    #[derive(Clone, Default)]
    struct TRecHook {
        state: Arc<TRecHookState>,
    }

    impl TypedWriteHook<[u8; 8], u64> for TRecHook {
        fn on_write(&self, _key: &[u8; 8], _old: Option<&u64>, new: Option<&u64>) {
            self.state.writes.fetch_add(1, AtomicOrdering::Relaxed);
            *crate::sync::lock(&self.state.last_new) = new.copied();
        }
    }

    fn open_var_typed_hooked(
        dir: &std::path::Path,
        hook: TRecHook,
    ) -> VarTypedTree<[u8; 8], u64, U64Codec, TRecHook> {
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        VarTypedTree::open_hooked(dir, cfg, U64Codec, hook).expect("open hooked")
    }

    #[test]
    fn var_typed_tree_atomic_fires_typed_hook() {
        let dir = tempdir().unwrap();
        let hook = TRecHook::default();
        let state = hook.state.clone();
        let tree = open_var_typed_hooked(dir.path(), hook);
        let k = 7u64.to_be_bytes();
        tree.atomic(&k, |s| {
            s.put(&k, &42)?;
            Ok(())
        })
        .expect("atomic");
        assert_eq!(state.writes.load(AtomicOrdering::Relaxed), 1);
        assert_eq!(*crate::sync::lock(&state.last_new), Some(42));
    }

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

    #[test]
    fn var_typed_get_many_and_update_many() {
        use crate::{Applied, Upsert};
        let dir = tempdir().unwrap();
        let cfg = Config::test();
        let tree = VarTypedTree::<[u8; 8], String, RapiraCodec>::open(dir.path(), cfg, RapiraCodec)
            .unwrap();
        tree.insert(&1u64.to_be_bytes(), &"a".to_string()).unwrap();
        tree.insert(&2u64.to_be_bytes(), &"b".to_string()).unwrap();

        let keys: Vec<[u8; 8]> = vec![2u64.to_be_bytes(), 9u64.to_be_bytes(), 1u64.to_be_bytes()];
        assert_eq!(
            tree.get_many(&keys),
            vec![Some("b".to_string()), None, Some("a".to_string())]
        );

        let items: Vec<([u8; 8], &str)> =
            vec![(1u64.to_be_bytes(), "a2"), (3u64.to_be_bytes(), "c")];
        let out = tree
            .update_many(items, |_k, _cur, p| Upsert::Put(p.to_string()))
            .unwrap();
        assert_eq!(
            out[0].1,
            Applied::Written {
                old: Some("a".to_string()),
                new: "a2".to_string()
            }
        );
        assert_eq!(
            out[1].1,
            Applied::Written {
                old: None,
                new: "c".to_string()
            }
        );
        assert_eq!(tree.get(&1u64.to_be_bytes()), Some("a2".to_string()));
        assert_eq!(tree.get(&3u64.to_be_bytes()), Some("c".to_string()));
    }

    #[test]
    fn var_typed_update_many_strict_on_codec_fault() {
        use crate::test_faults::{POISON, PoisonCodec};
        let dir = tempdir().unwrap();
        let tree: VarTypedTree<[u8; 8], u64, PoisonCodec> =
            VarTypedTree::open(dir.path(), Config::test(), PoisonCodec).unwrap();
        let k = 1u64.to_be_bytes();
        tree.put(&k, &POISON).unwrap();

        // Reading the poisoned old value during update_many must surface the decode
        // error (strict), not silently collapse to Applied::Kept.
        let res = tree.update_many(vec![(k, 5u64)], |_k, _old, v| Upsert::Put(*v));
        assert!(matches!(res, Err(DbError::CorruptedEntry { .. })));

        // The poisoned key was not overwritten by a hidden Keep path — raw bytes
        // remain, and a strict read still surfaces the corruption.
        assert!(matches!(
            tree.try_get(&k),
            Err(DbError::CorruptedEntry { .. })
        ));
    }

    #[test]
    fn var_typed_tree_shard_put_returns_existed() {
        let dir = tempdir().unwrap();
        let tree: VarTypedTree<[u8; 8], u64, U64Codec> =
            VarTypedTree::open(dir.path(), Config::test(), U64Codec).unwrap();
        let k = 1u64.to_be_bytes();
        let (fresh, overwrite) = tree
            .atomic(&k, |s| {
                let fresh = s.put(&k, &10u64)?;
                let overwrite = s.put(&k, &20u64)?;
                Ok((fresh, overwrite))
            })
            .expect("atomic");
        assert!(!fresh, "fresh insert must return false");
        assert!(overwrite, "overwrite must return true");
    }

    #[test]
    fn var_typed_tree_shard_update_and_fetch_update() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        let tree: VarTypedTree<[u8; 8], u64, U64Codec> =
            VarTypedTree::open(dir.path(), cfg, U64Codec).unwrap();
        let k = 1u64.to_be_bytes();
        tree.put(&k, &100u64).unwrap();

        // After put: k -> 100
        // update: 100 + 1 = 101 stored, returns Some(101) [new]
        // fetch_update: 101 + 1 = 102 stored, returns Some(101) [old]
        // missing key returns None
        let (upd, fetched, missing) = tree
            .atomic(&k, |s| {
                let upd = s.update(&k, |old| old + 1)?;
                let fetched = s.fetch_update(&k, |old| old + 1)?;
                let missing = s.update(&2u64.to_be_bytes(), |old| *old)?;
                Ok((upd, fetched, missing))
            })
            .expect("atomic");

        assert_eq!(upd, Some(101), "update returns new value");
        assert_eq!(fetched, Some(101), "fetch_update returns old value");
        assert_eq!(missing, None, "absent key returns None");
        assert_eq!(
            tree.get(&k),
            Some(102),
            "final stored value after two increments"
        );
    }

    #[test]
    fn var_typed_tree_atomic_update_fires_hook() {
        let dir = tempdir().unwrap();
        let hook = TRecHook::default();
        let state = hook.state.clone();
        let tree = open_var_typed_hooked(dir.path(), hook);
        let k = 7u64.to_be_bytes();

        // Seed the value so update finds the key.
        tree.put(&k, &10u64).unwrap();
        // Reset counter after the initial put.
        state.writes.store(0, AtomicOrdering::Relaxed);

        tree.atomic(&k, |s| s.update(&k, |old| old + 5))
            .expect("atomic update");

        // The hook must have fired once for the update (new = 15).
        assert_eq!(
            state.writes.load(AtomicOrdering::Relaxed),
            1,
            "hook fired once"
        );
        assert_eq!(
            *crate::sync::lock(&state.last_new),
            Some(15),
            "hook saw the new value"
        );
    }

    #[test]
    fn keys_skip_poison_decode_on_var_typed_tree() {
        use crate::test_faults::{POISON, PoisonCodec};

        let dir = tempdir().unwrap();
        let tree: VarTypedTree<[u8; 8], u64, PoisonCodec> =
            VarTypedTree::open(dir.path(), Config::test(), PoisonCodec).unwrap();
        let good_key = 1_u64.to_be_bytes();
        let poison_key = 2_u64.to_be_bytes();
        tree.put(&good_key, &1).unwrap();
        tree.put(&poison_key, &POISON).unwrap();

        assert_eq!(tree.iter().count(), 1, "iter skips poison");

        let keys: Vec<_> = tree.scope::<0>(&[0u8; 8]).keys().collect();
        assert!(keys.contains(&good_key));
        assert!(
            keys.contains(&poison_key),
            "keys must include the poison row's key without decoding"
        );
    }

    #[test]
    fn keys_perform_zero_disk_value_reads() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.direct_io = true;
        let tree: VarTypedTree<[u8; 8], Vec<u8>, RapiraCodec> =
            VarTypedTree::open(dir.path(), cfg, RapiraCodec).unwrap();

        for i in 1..=5u64 {
            let k = i.to_be_bytes();
            let v: Vec<u8> = (0..5000).map(|j| ((i * 17 + j) % 251) as u8).collect();
            tree.put(&k, &v).unwrap();
        }
        tree.flush_buffers().unwrap();

        let scope_key = [0u8; 8];
        let before = crate::var_tree::test_disk_value_reads();
        let n = tree.scope::<0>(&scope_key).keys().count();
        assert!(n > 0);
        assert_eq!(
            crate::var_tree::test_disk_value_reads(),
            before,
            "keys() must not read values from disk"
        );

        let before_iter = crate::var_tree::test_disk_value_reads();
        let iter_count = tree.scope::<0>(&scope_key).iter().count();
        assert!(iter_count > 0);
        assert!(
            crate::var_tree::test_disk_value_reads() > before_iter,
            "iter() must read values from disk (counter sanity check)"
        );
    }
}