armdb 0.9.0

use crate::Key;
use crate::batch::Applied;
use crate::codec::Codec;
use crate::compaction::{CompactionIndex, compact_shard};
use crate::config::Config;
use crate::ctx::{Ctx, Family, Held, Owned, ShardGuards};
use crate::disk_loc::DiskLoc;
use crate::engine::Engine;
use crate::error::{DbError, DbResult};
use crate::hook::{NoHook, TypedWriteHook};
use crate::map_index::{Dir, dir_from};
use crate::recovery::recover_typed_tree;
use crate::scope::DirRangeIter;
use crate::shard::ShardInner;
use crate::skiplist::iter::{SkipKeyIter, SkipKeyKeys};
use crate::skiplist::node::{SkipNode, TypedData, TypedNode, random_height};
use crate::skiplist::{InsertResult, SkipList};
use crate::sync::{self, Mutex};
use crate::ttl::{SweepOutcome, TtlSlots};
use crate::upsert::Upsert;
use std::mem::size_of;
use std::ops::Bound;
use std::sync::{Arc, OnceLock, Weak};

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

/// Guard-protected reference to a typed value inside a `TypedTree`.
///
/// Dereferences to `T`. The `seize` guard prevents the referenced data
/// from being reclaimed while this handle exists. No `Clone` required on `T`.
///
/// `TypedRef` is `!Send` because it holds a `seize::LocalGuard` which is
/// thread-local. Use `.clone_value()` or copy the data out if you need to
/// send the value across threads.
///
/// ```compile_fail,E0277
/// fn assert_send<T: Send>() {}
/// // TypedRef must NOT be Send — LocalGuard is !Send.
/// assert_send::<armdb::TypedRef<'_, u64>>();
/// ```
pub struct TypedRef<'a, T> {
    _guard: seize::LocalGuard<'a>,
    data: *const TypedData<T>,
    _marker: std::marker::PhantomData<&'a T>,
}

impl<'a, T> TypedRef<'a, T> {
    pub(crate) fn new(guard: seize::LocalGuard<'a>, data: *const TypedData<T>) -> Self {
        Self {
            _guard: guard,
            data,
            _marker: std::marker::PhantomData,
        }
    }
}

impl<T> std::ops::Deref for TypedRef<'_, T> {
    type Target = T;
    fn deref(&self) -> &T {
        // SAFETY: data pointer is valid while guard is alive
        unsafe { &(*self.data).value }
    }
}

impl<T: Clone> TypedRef<'_, T> {
    /// Clone the referenced value.
    pub fn clone_value(&self) -> T {
        (**self).clone()
    }
}

/// Deferred hook/retire work returned by _locked methods.
/// Caller drops shard lock, fires hook, then retires pointers.
pub(crate) struct MutationResult<T> {
    /// Raw pointer for TypedRef return. None if key was new / not found.
    ref_ptr: Option<*const TypedData<T>>,
    /// Hook args: (old_value_ptr, new_value_ptr). None = don't fire hook.
    hook: Option<(Option<*const T>, Option<*const T>)>,
    /// Pointer to retire via seize after hook. None = nothing to retire.
    retire: Option<*mut TypedData<T>>,
}

/// Pointers produced by an `update_locked_inner`. `old`/`retire` point at the
/// same `TypedData<T>` that was swapped out; `new` is the freshly allocated one.
struct UpdatePtrs<T> {
    old: *const TypedData<T>,
    new: *const TypedData<T>,
    retire: *mut TypedData<T>,
}

/// A tree with fixed-size keys and typed values `T`. Values are encoded via
/// a [`Codec`] for disk persistence but stored as `T` in memory — reads never
/// touch disk and return [`TypedRef<T>`](TypedRef) (guard-protected reference).
///
/// Each `TypedTree` owns its storage engine — one tree = one database directory.
///
/// # Clone-free design
///
/// Unlike `ConstTree` which copies `[u8; V]` values (they are `Copy`), `TypedTree`
/// returns [`TypedRef<T>`](TypedRef) — a guard-protected reference. The `seize` guard
/// inside `TypedRef` prevents reclamation of the old data while the reference exists.
/// This means `put()`, `delete()`, and `update()` can return the old value **without
/// requiring `T: Clone`**.
///
/// The only method that requires `T: Clone` is [`compact()`](Self::compact) — compaction
/// must copy the value into a new `TypedData` with an updated disk location.
///
/// # Write hooks
///
/// Uses [`TypedWriteHook<K, T>`] instead of `WriteHook<K>`. The hook receives `&T`
/// directly (not encoded bytes). `on_write` fires on `put`/`insert`/`delete`/`cas`/`compare_delete`/`update`/`fetch_update`
/// and on each mutation inside `atomic()` (replayed after the lock is released, in application order).
/// Old value is always provided (it lives in
/// memory) — `NEEDS_OLD_VALUE` is ignored.
///
/// `on_init` fires once per live entry during `migrate()` or `replay_init()`
/// (enable via `NEEDS_INIT = true`).
///
/// # Usage
///
/// ```ignore
/// let tree = TypedTree::<16, MyValue, RapiraCodec>::open(
///     "data/users",
///     Config::balanced().build(),
///     RapiraCodec,
/// )?;
/// tree.put(&key, value)?;
/// if let Some(r) = tree.get(&key) {
///     println!("{:?}", &*r);  // TypedRef<MyValue> derefs to &MyValue
/// }
/// tree.close()?;
/// ```
///
/// # Iteration
///
/// `iter()`, `range()`, and `prefix_iter()` all return [`TypedIter`] which
/// implements `Iterator + DoubleEndedIterator` with `Item = (K, TypedRef<T>)`.
/// Lock-free, zero disk I/O. [`TypedRef`] derefs to `&T`.
///
/// ```ignore
/// for (key, value) in tree.iter() { }
/// let latest = tree.prefix_iter(&user_id).take(20).collect::<Vec<_>>();
/// let oldest = tree.iter().rev().take(5);  // DoubleEndedIterator
/// ```
///
/// A scan that does not keep its rows should use
/// [`TypedIter::for_each_ref`] instead — it borrows `&T` for the
/// call and skips the per-row guard, which is ~3× faster.
///
/// # Features
///
/// Requires the `typed-tree` feature. Codec implementations:
/// - `rapira-codec` — [`RapiraCodec`](crate::RapiraCodec) for `T: rapira::Rapira`
/// - `bitcode-codec` — [`BitcodeCodec`](crate::BitcodeCodec) for `T: bitcode::Encode + 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). Both require
/// `T: Clone` — the node owns a live `T` and the codec is not on the read path.
/// See `docs/superpowers/specs/26-06-24-armdb-batch-tree-ops.md`.
pub struct TypedTree<K: Key, T: Send + Sync, C: Codec<T>, H: TypedWriteHook<K, T> = NoHook> {
    index: SkipList<TypedNode<K, T>>,
    engine: Engine,
    codec: C,
    compaction_threshold: f64,
    shard_prefix_bits: usize,
    reversed: bool,
    hook: H,
    encode_bufs: Vec<Mutex<Vec<u8>>>,
    /// 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>,
}

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

impl<K: Key, T: Send + Sync, C: Codec<T> + Sync, H: TypedWriteHook<K, T>> TypedTree<K, T, C, H> {
    /// Open or create a `TypedTree` 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,
        codec: C,
        hook: H,
    ) -> DbResult<Self> {
        let config = config.with_resolved_hints(false);
        Self::open_inner(path, config, codec, hook)
    }

    fn open_inner(
        path: impl AsRef<std::path::Path>,
        config: Config,
        codec: C,
        hook: H,
    ) -> DbResult<Self> {
        let compaction_threshold = config.compaction_threshold;
        let shard_prefix_bits = config.shard_prefix_bits;
        let reversed = config.reversed;
        let engine = Engine::open(path, config)?;

        let shard_count = engine.shards().len();
        let encode_bufs = (0..shard_count).map(|_| Mutex::new(Vec::new())).collect();

        let tree = Self {
            index: SkipList::new(reversed),
            engine,
            codec,
            compaction_threshold,
            shard_prefix_bits,
            reversed,
            hook,
            encode_bufs,
            tx_self: OnceLock::new(),
            ttl: TtlSlots::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_typed_tree::<K, T, C>(
            &shard_dir_refs,
            &shard_ids,
            tree.index(),
            &tree.codec,
            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);
        }
        // The transaction replay rule compares a reserved GSN against this
        // watermark, so it must exist in every build — not only with
        // `replication` on.
        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(),
            "typed_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()?;
        }
        self.engine.flush()
    }

    /// 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.
    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, T: Clone + Send + Sync, C: Codec<T> + Sync, H: TypedWriteHook<K, T>> CompactionIndex<K>
    for TypedTree<K, T, C, 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) {
            let current_data = node.load_data();
            if current_data.disk == old_loc {
                let new_data = Box::into_raw(Box::new(TypedData {
                    disk: new_loc,
                    value: current_data.value.clone(),
                }));
                let old_data_ptr = node.swap_data(new_data);
                unsafe {
                    self.index
                        .collector()
                        .retire(old_data_ptr, seize::reclaim::boxed::<TypedData<T>>);
                }
                return true;
            }
        }
        false
    }

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

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

impl<K: Key, T: Send + Sync, C: Codec<T> + Sync, H: TypedWriteHook<K, T>> Family
    for TypedTree<K, T, C, H>
{
    type Locked<'a>
        = sync::MutexGuard<'a, ShardInner>
    where
        Self: 'a;
    type Event<'v> = (K, MutationResult<T>);
    type Owned = (K, MutationResult<T>);

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

    fn fire(&self, ev: Self::Event<'_>) {
        let (k, mr) = ev;
        self.fire_hook_and_retire(&k, &mr);
    }

    fn own(ev: Self::Event<'_>) -> Self::Owned {
        ev
    }

    fn fire_owned(&self, ev: &Self::Owned) {
        self.fire_hook_and_retire(&ev.0, &ev.1);
    }

    fn apply_projection_for_event(&self, ev: &Self::Event<'_>) -> DbResult<()> {
        self.apply_projection_for_mutation(ev)
    }

    fn apply_projection_for_owned(&self, ev: &Self::Owned) -> DbResult<()> {
        self.apply_projection_for_mutation(ev)
    }
}

impl<K: Key, T: Send + Sync, C: Codec<T> + Sync, H: TypedWriteHook<K, T>> TypedTree<K, T, C, H> {
    /// Build per-shard TTL indexes by walking live in-memory entries.
    #[doc(hidden)]
    pub fn build_ttl_index(&self) -> DbResult<()> {
        if !H::EXPIRES {
            return Ok(());
        }
        let slots = self.ttl.install_and_clear(self.engine.shards().len());

        let _guard = self.index.collector().enter();
        let mut current = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        while !current.is_null() {
            let node = unsafe { &*current };
            current = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
            if node.is_marked() {
                continue;
            }
            let sid = self.shard_for(&node.key);
            // Shard lock before reading value — concurrent put under the same
            // shard cannot race a stale TypedData into the rebuild record.
            let _shard = self.engine.shards()[sid].lock();
            let data = node.load_data();
            let deadline = self.hook.deadline(&node.key, &data.value);
            sync::lock(&slots[sid]).record(&node.key, deadline);
        }
        Ok(())
    }

    /// Record a deadline when `H::EXPIRES` and the index is built.
    /// Caller must hold the engine shard lock for `shard_id` first.
    #[inline]
    fn record_ttl(&self, shard_id: usize, key: &K, expires_at: Option<u64>) {
        if !H::EXPIRES {
            return;
        }
        self.ttl.record(shard_id, key, expires_at);
    }

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

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

    /// Sweep expired rows across shards (round-robin) until `budget` is spent or no
    /// validated due entries remain. See [`Collection::sweep_expired`].
    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::new();
            let mut locked = self.lock_shard(sid);
            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 value = &node.load_data().value;
                        match crate::ttl::catch_deadline(|| self.hook.deadline(&key, value)) {
                            Some(Some(dl)) if dl <= now_ms => {
                                SweepAction::Delete(key, candidate_dl)
                            }
                            Some(Some(dl)) => SweepAction::Reschedule(key, dl),
                            Some(None) => SweepAction::Drop,
                            None => SweepAction::Retry(key),
                        }
                    }
                    None => SweepAction::Drop,
                };

                match action {
                    SweepAction::Delete(key, candidate_dl) => {
                        drop(idx);
                        match self.delete_locked_inner::<true>(sid, &mut locked, &guard, &key) {
                            Ok(mr) => {
                                idx = sync::lock(&slots[sid]);
                                if mr.hook.is_some() {
                                    shard_events.push((key, mr));
                                    outcome.deleted += 1;
                                }
                            }
                            Err(e) => {
                                // Row still live — put it back so later sweeps see it.
                                sync::lock(&slots[sid]).record(&key, Some(candidate_dl));
                                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);
                                let _ = crate::projection::replay_owned_events(
                                    self,
                                    &shard_events,
                                    false,
                                    sid,
                                );
                                return Err(e);
                            }
                        }
                    }
                    SweepAction::Reschedule(key, dl) => {
                        idx.record(&key, Some(dl));
                    }
                    SweepAction::Retry(key) => {
                        outcome.failed += 1;
                        idx.record(&key, Some(now_ms + crate::ttl::TTL_RETRY_BACKOFF_MS));
                    }
                    SweepAction::Drop => {}
                }
            }

            drop(idx);
            locked.release();
            crate::projection::replay_owned_events(self, &shard_events, false, sid)?;
        }

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

    /// Enable or disable TTL sweep for this collection (blocks in-flight passes when disabled).
    pub fn set_ttl_sweep(&self, enabled: bool) -> DbResult<()> {
        if H::EXPIRES {
            self.ttl.set_enabled(enabled)
        } else {
            Ok(())
        }
    }

    /// Disable sweep and require [`Self::rebuild_ttl_index`] before re-enabling after promotion.
    pub fn mark_replication_follower(&self) {
        if H::EXPIRES {
            self.ttl.mark_replication_follower();
        }
    }

    /// Clear and rebuild the in-memory TTL index from live rows.
    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(())
    }

    #[doc(hidden)]
    pub fn test_hold_sweep_pass(&self) -> Option<crate::ttl::TtlSweepPass<'_>> {
        if H::EXPIRES {
            self.ttl.begin_pass()
        } else {
            None
        }
    }
}

impl<K: Key, T: Send + Sync, C: Codec<T> + Sync, H: TypedWriteHook<K, T>> TypedTree<K, T, C, H> {
    /// Trigger a compaction pass across all shards.
    pub fn compact(&self) -> DbResult<usize>
    where
        T: Clone,
    {
        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 guard-protected reference to a value by key. Lock-free, zero disk I/O.
    ///
    /// **O(log n)** — SkipList point lookup.
    pub fn get(&self, key: &K) -> Option<TypedRef<'_, T>> {
        metrics::counter!("armdb.ops", "op" => "get", "tree" => "typed_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("typed_tree.get");
        let guard = self.index.collector().enter();
        let node = self.index.get(key.as_bytes(), &guard)?;
        // SAFETY: load_data returns a pointer valid while guard is alive.
        // We store the pointer (not a reference) so the borrow of guard is released,
        // allowing us to move guard into TypedRef.
        let data = node.load_data() as *const TypedData<T>;
        Some(TypedRef::new(guard, data))
    }

    /// Get a guard-protected reference to a value by key, returning `Err(KeyNotFound)` if absent.
    pub fn get_or_err(&self, key: &K) -> DbResult<TypedRef<'_, T>> {
        self.get(key).ok_or(DbError::KeyNotFound)
    }

    /// `Held`-only: buffer the event for replay, or — when hooks are off and no
    /// event will be buffered — retire the old value immediately.
    ///
    /// The facade (`Owned`) must **not** call this: it always returns the event so
    /// `Family::fire` → `fire_hook_and_retire` runs *after* the shard lock is
    /// released. `retire` can allocate, scan every thread's reservation and run
    /// `T::drop` in a batch, none of which belongs in the critical section — and
    /// `hook.rs` documents that hooks observe an unlocked shard.
    #[inline]
    fn defer_or_retire_event(
        &self,
        key: &K,
        mr: MutationResult<T>,
    ) -> Option<(K, MutationResult<T>)> {
        if H::NEEDS_WRITE {
            Some((*key, mr))
        } else {
            if let Some(ptr) = mr.retire {
                unsafe {
                    self.index
                        .collector()
                        .retire(ptr, seize::reclaim::boxed::<TypedData<T>>);
                }
            }
            None
        }
    }

    /// Insert or update a key-value pair. Returns a `TypedRef` to the old value
    /// if the key existed (valid while the guard lives).
    pub fn put(&self, key: &K, value: T) -> DbResult<Option<TypedRef<'_, T>>> {
        metrics::counter!("armdb.ops", "op" => "put", "tree" => "typed_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("typed_tree.put");
        let expires_at = if H::EXPIRES {
            self.hook.deadline(key, &value)
        } else {
            None
        };
        let shard_id = self.shard_for(key);
        let guard = self.index.collector().enter();
        let mut ctx = Owned;
        let ref_ptr = Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let mr = self.put_locked_inner::<true>(
                shard_id,
                locked,
                &guard,
                key,
                value,
                crate::ttl::TtlAt::Precomputed(expires_at),
            )?;
            let ref_ptr = mr.ref_ptr;
            *ev_out = Some((*key, mr));
            Ok(ref_ptr)
        })?;
        Ok(ref_ptr.map(|p| TypedRef::new(guard, p)))
    }

    /// 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<()> {
        metrics::counter!("armdb.ops", "op" => "insert", "tree" => "typed_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("typed_tree.insert");
        let expires_at = if H::EXPIRES {
            self.hook.deadline(key, &value)
        } else {
            None
        };
        let shard_id = self.shard_for(key);
        let guard = self.index.collector().enter();
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let mr =
                self.insert_locked::<true>(shard_id, locked, &guard, key, value, expires_at)?;
            *ev_out = Some((*key, mr));
            Ok(())
        })
    }

    /// Delete a key. Returns a `TypedRef` to the old value if the key existed.
    pub fn delete(&self, key: &K) -> DbResult<Option<TypedRef<'_, T>>> {
        metrics::counter!("armdb.ops", "op" => "delete", "tree" => "typed_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("typed_tree.delete");
        let shard_id = self.shard_for(key);
        let guard = self.index.collector().enter();
        let mut ctx = Owned;
        let ref_ptr = Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let mr = self.delete_locked_inner::<true>(shard_id, locked, &guard, key)?;
            let ref_ptr = mr.ref_ptr;
            *ev_out = Some((*key, mr));
            Ok(ref_ptr)
        })?;
        Ok(ref_ptr.map(|p| TypedRef::new(guard, p)))
    }

    /// 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 TypedShard<'_, K, T, C, H>) -> DbResult<R>,
    ) -> DbResult<R> {
        let _atomic_depth = crate::AtomicDepthGuard::try_enter()?;
        let shard_id = self.shard_for(shard_key);
        let mut shard = self.tx_shard(shard_id);
        let result = f(&mut shard);
        let (events, guard) = Self::tx_release_shard(shard);
        let result =
            crate::projection::finish_atomic_teardown(self, shard_id, result, false, &events);
        drop(guard);
        result
    }

    /// Fire deferred on_write and retire old data. Called after lock release.
    ///
    /// SAFETY: caller must hold a live `seize::LocalGuard` that was entered
    /// before the mutation. This guarantees old/new value pointers are valid
    /// (not yet reclaimed) even if a reentrant hook retires them.
    pub(crate) fn fire_hook_and_retire(&self, key: &K, mr: &MutationResult<T>) {
        if let Some((old, new)) = mr.hook {
            unsafe {
                self.hook.on_write(key, old.map(|p| &*p), new.map(|p| &*p));
            }
        }
        if let Some(ptr) = mr.retire {
            unsafe {
                self.index
                    .collector()
                    .retire(ptr, seize::reclaim::boxed::<TypedData<T>>);
            }
        }
    }

    fn apply_projection_for_mutation(&self, ev: &(K, MutationResult<T>)) -> DbResult<()> {
        if !H::PROJECTS {
            return Ok(());
        }
        let Some(targets) = self.hook.projection_targets() else {
            return Ok(());
        };
        let (key, mr) = ev;
        let Some((old_ptr, new_ptr)) = mr.hook else {
            return Ok(());
        };
        let old = old_ptr.map(|p| unsafe { &*p });
        let new = new_ptr.map(|p| unsafe { &*p });
        let (max_rows, max_bytes) = self.effective_projection_limits();
        let plan = crate::projection::diff_projection(
            &self.hook, targets, max_rows, max_bytes, key, old, new,
        )?;
        plan.apply_or_incomplete()
    }

    #[allow(clippy::too_many_arguments)]
    fn put_locked_inner<const HOOKS: bool>(
        &self,
        shard_id: usize,
        inner: &mut ShardInner,
        guard: &seize::LocalGuard<'_>,
        key: &K,
        value: T,
        expires_at: crate::ttl::TtlAt,
    ) -> DbResult<MutationResult<T>> {
        // Resolve before `value` is moved into the index below.
        let expires_at = match expires_at {
            crate::ttl::TtlAt::Precomputed(at) => at,
            crate::ttl::TtlAt::Derive if H::EXPIRES => self.hook.deadline(key, &value),
            crate::ttl::TtlAt::Derive => None,
        };
        let mut buf = sync::lock(&self.encode_bufs[shard_id]);
        self.codec.encode_to(&value, &mut buf)?;
        let (disk_loc, _gsn) = inner.append_entry(shard_id as u8, key.as_bytes(), &buf, false)?;

        // Fast path: key exists — atomic swap
        if let Some(existing) = self.index.get(key.as_bytes(), guard) {
            let old_data = existing.load_data() as *const TypedData<T>;
            let new_data = Box::new(TypedData {
                disk: disk_loc,
                value,
            });
            let new_data_ptr = Box::into_raw(new_data);

            // Commit: swap pointer in index
            let old_data_ptr = existing.swap_data(new_data_ptr);
            let old_disk = unsafe { (*old_data_ptr).disk };
            inner.add_dead_bytes(
                old_disk.file_id,
                crate::entry::entry_size(size_of::<K>(), old_disk.len),
            );

            if HOOKS {
                self.record_ttl(shard_id, key, expires_at);
                return Ok(MutationResult {
                    ref_ptr: Some(old_data),
                    hook: Some((
                        Some(unsafe { &(*old_data).value as *const T }),
                        Some(unsafe { &(*new_data_ptr).value as *const T }),
                    )),
                    retire: Some(old_data_ptr),
                });
            }
            self.record_ttl(shard_id, key, expires_at);
            unsafe {
                self.index
                    .collector()
                    .retire(old_data_ptr, seize::reclaim::boxed::<TypedData<T>>);
            }
            return Ok(MutationResult {
                ref_ptr: Some(old_data),
                hook: None,
                retire: None,
            });
        }

        // Slow path: new key — allocate node + insert
        let height = random_height();
        let node_ptr = TypedNode::alloc(*key, value, disk_loc, height);

        match self.index.insert(node_ptr, guard) {
            InsertResult::Inserted => {
                self.record_ttl(shard_id, key, expires_at);
                let hook = if HOOKS {
                    let data = unsafe { &*node_ptr }.load_data();
                    Some((None, Some(&data.value as *const T)))
                } else {
                    None
                };
                Ok(MutationResult {
                    ref_ptr: None,
                    hook,
                    retire: None,
                })
            }
            InsertResult::Exists(existing) => {
                // Race: another thread inserted same key between get and insert
                let old_data = existing.load_data() as *const TypedData<T>;
                let new_data_ptr = unsafe {
                    let ptr = (*node_ptr).data.load(std::sync::atomic::Ordering::Relaxed);
                    (*node_ptr)
                        .data
                        .store(std::ptr::null_mut(), std::sync::atomic::Ordering::Relaxed);
                    ptr
                };
                let old_data_ptr = existing.swap_data(new_data_ptr);
                let old_disk = unsafe { (*old_data_ptr).disk };
                inner.add_dead_bytes(
                    old_disk.file_id,
                    crate::entry::entry_size(size_of::<K>(), old_disk.len),
                );
                // Dealloc the unused node (not the data — data was moved to existing)
                unsafe {
                    TypedNode::<K, T>::dealloc_node(node_ptr);
                }

                if HOOKS {
                    self.record_ttl(shard_id, key, expires_at);
                    Ok(MutationResult {
                        ref_ptr: Some(old_data),
                        hook: Some((
                            Some(unsafe { &(*old_data).value as *const T }),
                            Some(unsafe { &(*new_data_ptr).value as *const T }),
                        )),
                        retire: Some(old_data_ptr),
                    })
                } else {
                    self.record_ttl(shard_id, key, expires_at);
                    unsafe {
                        self.index
                            .collector()
                            .retire(old_data_ptr, seize::reclaim::boxed::<TypedData<T>>);
                    }
                    Ok(MutationResult {
                        ref_ptr: Some(old_data),
                        hook: None,
                        retire: None,
                    })
                }
            }
        }
    }

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

        let mut buf = sync::lock(&self.encode_bufs[shard_id]);
        self.codec.encode_to(&value, &mut buf)?;
        let (disk_loc, _gsn) = inner.append_entry(shard_id as u8, key.as_bytes(), &buf, false)?;
        let height = random_height();
        let node_ptr = TypedNode::alloc(*key, value, disk_loc, height);
        self.index.insert(node_ptr, guard);

        let hook = if HOOKS {
            let data = unsafe { &*node_ptr }.load_data();
            Some((None, Some(&data.value as *const T)))
        } else {
            None
        };
        self.record_ttl(shard_id, key, expires_at);
        Ok(MutationResult {
            ref_ptr: None,
            hook,
            retire: None,
        })
    }

    fn delete_locked_inner<const HOOKS: bool>(
        &self,
        shard_id: usize,
        inner: &mut ShardInner,
        guard: &seize::LocalGuard<'_>,
        key: &K,
    ) -> DbResult<MutationResult<T>> {
        let node = match self.index.get(key.as_bytes(), guard) {
            Some(n) => n,
            None => {
                return Ok(MutationResult {
                    ref_ptr: None,
                    hook: None,
                    retire: None,
                });
            }
        };
        let data = node.load_data() as *const TypedData<T>;

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

        let disk = unsafe { (*data).disk };
        inner.add_dead_bytes(
            disk.file_id,
            crate::entry::entry_size(size_of::<K>(), disk.len),
        );

        // Never retire data separately here — index.remove() already retired
        // the node, and TypedNode::reclaim will free the data inside it.
        // The guard keeps the node (and its data) alive until dropped.
        let hook = if HOOKS {
            Some((Some(unsafe { &(*data).value as *const T }), None))
        } else {
            None
        };
        self.record_ttl(shard_id, key, None);
        Ok(MutationResult {
            ref_ptr: Some(data),
            hook,
            retire: None,
        })
    }

    /// Compare-and-swap: if current value == expected, replace with new_value.
    /// Returns `Ok(())` on success, `Err(CasMismatch)` if current != expected,
    /// `Err(KeyNotFound)` if key doesn't exist.
    pub fn cas(&self, key: &K, expected: &T, new_value: T) -> DbResult<()>
    where
        T: PartialEq,
    {
        metrics::counter!("armdb.ops", "op" => "cas", "tree" => "typed_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("typed_tree.cas");
        let expires_at = if H::EXPIRES {
            self.hook.deadline(key, &new_value)
        } else {
            None
        };
        let shard_id = self.shard_for(key);
        let guard = self.index.collector().enter();
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let existing = self
                .index
                .get(key.as_bytes(), &guard)
                .ok_or(DbError::KeyNotFound)?;

            let current_data = existing.load_data();
            if current_data.value != *expected {
                return Err(DbError::CasMismatch);
            }

            let mut buf = sync::lock(&self.encode_bufs[shard_id]);
            self.codec.encode_to(&new_value, &mut buf)?;
            let (disk_loc, _gsn) =
                locked.append_entry(shard_id as u8, key.as_bytes(), &buf, false)?;

            let new_data_ptr = Box::into_raw(Box::new(TypedData {
                disk: disk_loc,
                value: new_value,
            }));
            let old_data = existing.swap_data(new_data_ptr);
            let old_disk = unsafe { (*old_data).disk };
            locked.add_dead_bytes(
                old_disk.file_id,
                crate::entry::entry_size(size_of::<K>(), old_disk.len),
            );

            self.record_ttl(shard_id, key, expires_at);

            let mr = MutationResult {
                ref_ptr: None,
                hook: Some((
                    Some(&current_data.value as *const T),
                    Some(unsafe { &(*new_data_ptr).value as *const T }),
                )),
                retire: Some(old_data),
            };
            *ev_out = Some((*key, mr));
            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.
    pub fn compare_delete(&self, key: &K, expected: &T) -> DbResult<()>
    where
        T: PartialEq,
    {
        metrics::counter!("armdb.ops", "op" => "compare_delete", "tree" => "typed_tree")
            .increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("typed_tree.compare_delete");
        let shard_id = self.shard_for(key);
        let guard = self.index.collector().enter();
        let mut ctx = Owned;
        Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let existing = self
                .index
                .get(key.as_bytes(), &guard)
                .ok_or(DbError::KeyNotFound)?;
            if existing.load_data().value != *expected {
                return Err(DbError::CasMismatch);
            }

            let mr = self.delete_locked_inner::<true>(shard_id, locked, &guard, key)?;
            *ev_out = Some((*key, mr));
            Ok(())
        })
    }

    /// Atomically read-modify-write. Returns `Some(TypedRef)` to the **new** value
    /// if key existed, `None` otherwise.
    /// The closure must not be heavy (shard lock is held).
    pub fn update(&self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<TypedRef<'_, T>>> {
        self.update_inner(key, f, false)
    }

    /// Like [`update()`](Self::update), but returns `Some(TypedRef)` to the **old** value.
    pub fn fetch_update(
        &self,
        key: &K,
        f: impl FnOnce(&T) -> T,
    ) -> DbResult<Option<TypedRef<'_, T>>> {
        self.update_inner(key, f, true)
    }

    /// 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> {
        metrics::counter!("armdb.ops", "op" => "upsert", "tree" => "typed_tree").increment(1);
        self.atomic(key, |s| crate::upsert::upsert_via_held(s, key, f))
    }

    fn update_locked_inner(
        &self,
        shard_id: usize,
        inner: &mut ShardInner,
        guard: &seize::LocalGuard<'_>,
        key: &K,
        f: impl FnOnce(&T) -> T,
    ) -> DbResult<Option<UpdatePtrs<T>>> {
        let existing = match self.index.get(key.as_bytes(), guard) {
            Some(n) => n,
            None => return Ok(None),
        };
        let old_data = existing.load_data();
        let new_value = f(&old_data.value);

        let mut buf = sync::lock(&self.encode_bufs[shard_id]);
        self.codec.encode_to(&new_value, &mut buf)?;
        let (disk_loc, _gsn) = inner.append_entry(shard_id as u8, key.as_bytes(), &buf, false)?;
        drop(buf);

        let new_data_ptr = Box::into_raw(Box::new(TypedData {
            disk: disk_loc,
            value: new_value,
        }));
        let old_data_ptr = existing.swap_data(new_data_ptr);
        let old_disk = unsafe { (*old_data_ptr).disk };
        inner.add_dead_bytes(
            old_disk.file_id,
            crate::entry::entry_size(size_of::<K>(), old_disk.len),
        );

        let expires_at = if H::EXPIRES {
            self.hook.deadline(key, unsafe { &(*new_data_ptr).value })
        } else {
            None
        };
        self.record_ttl(shard_id, key, expires_at);

        Ok(Some(UpdatePtrs {
            old: old_data as *const TypedData<T>,
            new: new_data_ptr as *const TypedData<T>,
            retire: old_data_ptr,
        }))
    }

    fn update_inner(
        &self,
        key: &K,
        f: impl FnOnce(&T) -> T,
        return_old: bool,
    ) -> DbResult<Option<TypedRef<'_, T>>> {
        metrics::counter!("armdb.ops", "op" => "update", "tree" => "typed_tree").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("typed_tree.update");
        let shard_id = self.shard_for(key);
        let guard = self.index.collector().enter();
        let mut ctx = Owned;
        let ref_ptr = Ctx::with_shard(&mut ctx, self, shard_id, |locked, ev_out| {
            let ptrs = match self.update_locked_inner(shard_id, locked, &guard, key, f)? {
                Some(p) => p,
                None => return Ok(None),
            };
            let ref_ptr = if return_old { ptrs.old } else { ptrs.new };
            let mr = MutationResult {
                ref_ptr: None,
                hook: Some((
                    Some(unsafe { &(*ptrs.old).value as *const T }),
                    Some(unsafe { &(*ptrs.new).value as *const T }),
                )),
                retire: Some(ptrs.retire),
            };
            *ev_out = Some((*key, mr));
            Ok(Some(ref_ptr))
        })?;
        Ok(ref_ptr.map(|p| TypedRef::new(guard, p)))
    }

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

    /// Return the first entry in index order, or `None` if empty.
    /// With `reversed=true` (default): the entry with the largest key.
    /// O(1) — follows head's level-0 pointer, skipping marked nodes.
    pub fn first(&self) -> Option<(K, TypedRef<'_, T>)> {
        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 data = node.load_data() as *const TypedData<T>;
                return Some((node.key, TypedRef::new(guard, data)));
            }
            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.
    pub fn last(&self) -> Option<(K, TypedRef<'_, T>)> {
        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() {
                let data = node.load_data() as *const TypedData<T>;
                return Some((node.key, TypedRef::new(guard, data)));
            }
            ptr = self.index.find_last_lt(node.key_bytes(), &guard);
        }
        None
    }

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

    fn resolve_front_asc(
        &self,
        bound: &Bound<&K>,
        guard: &seize::LocalGuard<'_>,
    ) -> *mut TypedNode<K, T> {
        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 resolve_front_rev(
        &self,
        bound: &Bound<&K>,
        guard: &seize::LocalGuard<'_>,
    ) -> *mut TypedNode<K, T> {
        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]) -> Option<(K, Bound<K>)> {
        if prefix.len() > size_of::<K>() {
            return None;
        }
        if self.reversed {
            let mut search = K::zeroed();
            for b in search.as_bytes_mut().iter_mut() {
                *b = 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);
            Some((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);
            Some((search, end))
        }
    }

    /// Iterate entries whose keys start with `prefix`.
    ///
    /// `reversed=true` (default): yields matching keys in DESC order.
    /// `next()` is O(1), `next_back()` is O(log n).
    pub fn prefix_iter(&self, prefix: &[u8]) -> TypedIter<'_, K, T> {
        let guard = self.index.collector().enter();
        let Some((search_key, end)) = self.prefix_bounds(prefix) else {
            return TypedIter {
                list: &self.index,
                front: std::ptr::null_mut(),
                back: None,
                end: Bound::Unbounded,
                start: Bound::Unbounded,
                reversed: self.reversed,
                done: true,
                _guard: guard,
            };
        };
        let front = self.index.find_first_ge(search_key.as_bytes(), &guard);
        TypedIter {
            list: &self.index,
            front,
            back: None,
            end,
            start: Bound::Included(search_key),
            reversed: self.reversed,
            done: false,
            _guard: guard,
        }
    }

    /// Iterate all entries in index order.
    ///
    /// `reversed=true` (default): DESC. `reversed=false`: ASC.
    /// `next()` is O(1), `next_back()` is O(log n).
    pub fn iter(&self) -> TypedIter<'_, K, T> {
        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)
        });
        TypedIter {
            list: &self.index,
            front,
            back: None,
            end: Bound::Unbounded,
            start: Bound::Unbounded,
            reversed: self.reversed,
            done: false,
            _guard: guard,
        }
    }

    /// Iterate entries in `[start, end)` — start inclusive, end exclusive.
    ///
    /// `reversed=true` (default): DESC within range. `reversed=false`: ASC.
    /// `next()` is O(1), `next_back()` is O(log n).
    pub fn range(&self, start: &K, end: &K) -> TypedIter<'_, K, T> {
        self.range_bounds(Bound::Included(start), Bound::Excluded(end))
    }

    /// Iterate entries in range defined by `start` and `end` bounds.
    ///
    /// Unlike [`range()`](Self::range), allows `Included`, `Excluded`, or `Unbounded`
    /// for each bound independently.
    ///
    /// `reversed=true` (default): DESC within range. `reversed=false`: ASC.
    /// `next()` is O(1), `next_back()` is O(log n).
    pub fn range_bounds(&self, start: Bound<&K>, end: Bound<&K>) -> TypedIter<'_, K, T> {
        let guard = self.index.collector().enter();
        if self.reversed {
            let front = self.resolve_front_rev(&end, &guard);
            TypedIter {
                list: &self.index,
                front,
                back: None,
                end: bound_owned(&start),
                start: bound_owned(&end),
                reversed: true,
                done: false,
                _guard: guard,
            }
        } else {
            let front = self.resolve_front_asc(&start, &guard);
            TypedIter {
                list: &self.index,
                front,
                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, TypedNode<K, T>> {
        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_rev(&end, &guard);
            (front, bound_owned(&start), bound_owned(&end), true)
        } else {
            let front = self.resolve_front_asc(&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, each step uses
    /// [`DoubleEndedIterator::next_back`] on the storage-order iterator (`find_last_lt` on the
    /// skiplist) — **O(log n) per row** vs O(1) along storage order.
    pub(crate) fn range_bounds_dir(
        &self,
        start: Bound<&K>,
        end: Bound<&K>,
        dir: Dir,
    ) -> DirRangeIter<TypedIter<'_, K, T>> {
        let iter = self.range_bounds(start, end);
        DirRangeIter::from_storage_match(iter, dir == dir_from(self.reversed))
    }

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

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

    /// Iterate all entries and optionally mutate them. Call once at startup.
    ///
    /// **Must run before any other writer obtains a handle to this collection.**
    /// Concurrent writes during migration produce undefined results.
    ///
    /// The callback receives each (key, &T) and returns `MigrateAction`:
    /// - `Keep` — no change (fires `on_init` if `NEEDS_INIT`)
    /// - `Update(new_value)` — replace value (hook-free write, fires `on_init`)
    /// - `Delete` — remove entry (hook-free tombstone)
    ///
    /// 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;

        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 data = node.load_data();
            match f(&node.key, &data.value) {
                MigrateAction::Keep => {
                    if fire_init && H::NEEDS_INIT {
                        self.hook.on_init(&node.key, &data.value);
                    }
                }
                MigrateAction::Update(value) => {
                    let shard_id = self.shard_for(&node.key);
                    let init_ptr = {
                        let mut inner = self.engine.shards()[shard_id].lock();
                        let g = self.index.collector().enter();
                        // `Derive`, not `Precomputed(None)`: a migration rewrites the
                        // value, so the row's deadline must be recomputed from it.
                        // Passing `None` here would clear the deadline of every
                        // migrated row — invisible until the row failed to expire.
                        self.put_locked_inner::<false>(
                            shard_id,
                            &mut inner,
                            &g,
                            &node.key,
                            value,
                            crate::ttl::TtlAt::Derive,
                        )?;
                        (fire_init && H::NEEDS_INIT)
                            .then(|| node.load_data() as *const TypedData<T>)
                    };
                    if let Some(ptr) = init_ptr {
                        let data = unsafe { &*ptr };
                        self.hook.on_init(&node.key, &data.value);
                    }
                    count += 1;
                }
                MigrateAction::Delete => {
                    let shard_id = self.shard_for(&node.key);
                    let mut inner = self.engine.shards()[shard_id].lock();
                    let g = self.index.collector().enter();
                    self.delete_locked_inner::<false>(shard_id, &mut inner, &g, &node.key)?;
                    count += 1;
                }
            }
        }

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

    /// Replay `on_init` for every live entry. Used when no migration runs.
    pub(crate) fn replay_init(&self) {
        if !H::NEEDS_INIT {
            return;
        }
        let _guard = self.index.collector().enter();
        let mut current = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        while !current.is_null() {
            let node = unsafe { &*current };
            current = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
            if node.is_marked() {
                continue;
            }
            let data = node.load_data();
            self.hook.on_init(&node.key, &data.value);
        }
    }

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

    pub(crate) fn reconcile_projections(&self) -> DbResult<()> {
        if !H::PROJECTS {
            return Ok(());
        }
        let Some(targets) = self.hook.projection_targets() else {
            return Ok(());
        };
        let (max_rows, max_bytes) = self.effective_projection_limits();
        crate::projection::reconcile_carrier_projections(
            targets,
            |emit| {
                let mut err: Option<DbError> = None;
                let _guard = self.index.collector().enter();
                let mut current = crate::skiplist::strip_mark(unsafe {
                    (*self.index.head_ptr())
                        .tower(0)
                        .load(std::sync::atomic::Ordering::Acquire)
                });
                while !current.is_null() {
                    let node = unsafe { &*current };
                    current = crate::skiplist::strip_mark(
                        node.tower(0).load(std::sync::atomic::Ordering::Acquire),
                    );
                    if node.is_marked() {
                        continue;
                    }
                    if err.is_some() {
                        break;
                    }
                    let data = node.load_data();
                    let mut sink =
                        crate::projection::ProjectionSink::new(targets, max_rows, max_bytes);
                    match self
                        .hook
                        .project(&node.key, &data.value, &mut sink)
                        .and_then(|()| {
                            for row in sink.rows() {
                                emit(row.target, &row.key, &row.value)?;
                            }
                            Ok(())
                        }) {
                        Ok(()) => {}
                        Err(e) => err = Some(e),
                    }
                }
                err.map_or(Ok(()), Err)
            },
            true,
        )
    }

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

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

    pub(crate) fn codec(&self) -> &C {
        &self.codec
    }

    /// 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(
                "TypedTree::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)?;
        }
        Ok(())
    }

    pub(crate) fn tx_preplan_projection_targets(
        &self,
    ) -> DbResult<Vec<Arc<dyn crate::txn::participant::TxParticipant>>> {
        if !H::PROJECTS {
            return Ok(Vec::new());
        }
        let Some(targets) = self.hook.projection_targets() else {
            return Ok(Vec::new());
        };
        Ok(targets
            .erased_targets()
            .iter()
            .filter_map(|h| targets.resolve_tx_participant(h.name))
            .collect())
    }

    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 old_t = match old {
            Some(b) => Some(self.codec().decode_from(b)?),
            None => None,
        };
        let new_t = match new {
            Some(b) => Some(self.codec().decode_from(b)?),
            None => None,
        };
        let (max_rows, max_bytes) = self.effective_projection_limits();
        let plan = crate::projection::diff_projection(
            &self.hook,
            targets,
            max_rows,
            max_bytes,
            &key_typed,
            old_t.as_ref(),
            new_t.as_ref(),
        )?;
        plan.into_tx_derived(targets)
    }

    /// 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) -> TypedShard<'_, K, T, C, H> {
        let locked = self.lock_shard(sid);
        let guard = self.index.collector().enter();
        TypedShard {
            tree: self,
            held: Held {
                locked,
                sid,
                events: Vec::new(),
            },
            guard,
        }
    }

    /// Release a shard built by [`tx_shard`], returning its buffered events.
    /// The `seize` guard is dropped by the caller's `Deferred` owner, not here.
    pub(crate) fn tx_release_shard(
        shard: TypedShard<'_, K, T, C, H>,
    ) -> (Vec<(K, MutationResult<T>)>, seize::LocalGuard<'_>) {
        let TypedShard { held, guard, .. } = shard;
        let Held { locked, events, .. } = held;
        locked.release();
        (events, 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()
    }

    pub fn shard_for(&self, key: &K) -> usize {
        if self.shard_prefix_bits == 0 || self.shard_prefix_bits >= size_of::<K>() * 8 {
            let hash = xxhash_rust::xxh3::xxh3_64(key.as_bytes());
            return (hash as usize) % self.engine.shards().len();
        }

        let full_bytes = self.shard_prefix_bits / 8;
        let extra_bits = self.shard_prefix_bits % 8;

        let hash = if extra_bits == 0 {
            xxhash_rust::xxh3::xxh3_64(&key.as_bytes()[..full_bytes])
        } else {
            let mut buf = K::zeroed();
            buf.as_bytes_mut()[..full_bytes + 1].copy_from_slice(&key.as_bytes()[..full_bytes + 1]);
            let mask = !((1u8 << (8 - extra_bits)) - 1);
            buf.as_bytes_mut()[full_bytes] = key.as_bytes()[full_bytes] & mask;
            xxhash_rust::xxh3::xxh3_64(&buf.as_bytes()[..full_bytes + 1])
        };

        (hash as usize) % self.engine.shards().len()
    }

    /// Batch point-read returning cloned values, aligned to `keys`. Lock-free,
    /// zero disk I/O (values live in memory). Requires `T: Clone` because the node
    /// owns a live `T` and the codec is not on the read path.
    ///
    /// Result is aligned to `keys`: position `i` is `Some(value)` if `keys[i]`
    /// exists, else `None`. Not snapshot-consistent — a concurrent writer may be
    /// observed mid-walk. Without concurrent writers, duplicate keys yield the
    /// same value at each position.
    pub fn get_many(&self, keys: &[K]) -> Vec<Option<T>>
    where
        T: Clone,
    {
        let n = keys.len();
        let mut out: Vec<Option<T>> = (0..n).map(|_| None).collect();
        if n == 0 {
            return out;
        }
        // Visit keys in list order so the finger only moves forward.
        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] = Some(node.load_data().value.clone());
            }
        }
        out
    }

    /// Batch read-modify-write (tier-2, shard-grouped). The closure decides per key
    /// from `Option<&T>` (`None` = absent). Keys are grouped by shard and each
    /// shard is processed under its own lock, one at a time. Within a shard, items
    /// are applied in input order, so a later duplicate of a key sees the prior
    /// in-batch result (natural fold). Per-shard isolation only — no rollback, no
    /// crash-atomicity. On a real I/O `Err` the batch stops; the applied prefix is
    /// committed and its hooks fire. Result is aligned to `items`.
    ///
    /// Requires `T: Clone` because old/new values for `Applied` and the closure's
    /// `Option<&T>` come from cloning the live in-node value.
    pub fn update_many<P>(
        &self,
        items: Vec<(K, P)>,
        f: impl Fn(&K, Option<&T>, &P) -> Upsert<T>,
    ) -> DbResult<Vec<(K, Applied<T>)>>
    where
        T: Clone,
    {
        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<T>)>> = (0..n).map(|_| None).collect();
        let mut err: Option<DbError> = None;

        'shards: for (shard_id, idxs) in by_shard {
            let guard = self.index.collector().enter();
            let mut events: Vec<(K, MutationResult<T>)> = Vec::new();
            {
                let shard = &self.engine.shards()[shard_id];
                let mut inner = shard.lock();
                for &i in &idxs {
                    let (k, p) = &items[i];
                    let cur: Option<T> = self
                        .index
                        .get(k.as_bytes(), &guard)
                        .map(|node| node.load_data().value.clone());
                    let applied = match f(k, cur.as_ref(), p) {
                        Upsert::Put(v) => {
                            let new_clone = v.clone();
                            match self.put_locked_inner::<true>(
                                shard_id,
                                &mut inner,
                                &guard,
                                k,
                                v,
                                crate::ttl::TtlAt::Derive,
                            ) {
                                Ok(mr) => {
                                    if H::NEEDS_WRITE {
                                        events.push((*k, mr));
                                    } else if let Some(ptr) = mr.retire {
                                        unsafe {
                                            self.index
                                                .collector()
                                                .retire(ptr, seize::reclaim::boxed::<TypedData<T>>);
                                        }
                                    }
                                    Applied::Written {
                                        old: cur,
                                        new: new_clone,
                                    }
                                }
                                Err(e) => {
                                    err = Some(e);
                                    break;
                                }
                            }
                        }
                        Upsert::Keep => Applied::Kept,
                        Upsert::Delete => {
                            match self.delete_locked_inner::<true>(shard_id, &mut inner, &guard, k)
                            {
                                Ok(mr) => match cur {
                                    Some(old) => {
                                        if H::NEEDS_WRITE {
                                            events.push((*k, mr));
                                        }
                                        // delete_locked_inner retires via the node itself;
                                        // mr.retire is always None for delete — nothing extra.
                                        Applied::Deleted(old)
                                    }
                                    None => Applied::Kept, // delete-miss: no write, no hook
                                },
                                Err(e) => {
                                    err = Some(e);
                                    break;
                                }
                            }
                        }
                    };
                    out[i] = Some((*k, applied));
                }
                // shard lock (`inner`) released here
            }
            // Hooks + retire after unlock, while this shard's guard is still alive.
            if H::NEEDS_WRITE {
                for (k, mr) in &events {
                    self.fire_hook_and_retire(k, mr);
                }
            }
            drop(guard);
            if err.is_some() {
                break 'shards;
            }
        }

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

/// Handle for atomic multi-key operations within a single shard.
/// Obtained via [`TypedTree::atomic`]. The shard lock is held for the
/// lifetime of this struct — keep the closure short.
pub struct TypedShard<'a, K: Key, T: Send + Sync, C: Codec<T>, H: TypedWriteHook<K, T> = NoHook> {
    tree: &'a TypedTree<K, T, C, H>,
    held: Held<'a, TypedTree<K, T, C, H>>,
    guard: seize::LocalGuard<'a>,
}

impl<K: Key, T: Send + Sync, C: Codec<T> + Sync, H: TypedWriteHook<K, T>>
    TypedShard<'_, K, T, C, H>
{
    pub fn put(&mut self, key: &K, value: T) -> DbResult<Option<TypedRef<'_, T>>> {
        self.check_shard(key)?;
        let tree = self.tree;
        let sid = self.held.sid;
        let ref_ptr = {
            let guard = &self.guard;
            Ctx::with_shard(&mut self.held, tree, sid, |locked, ev_out| {
                let mr = tree.put_locked_inner::<true>(
                    sid,
                    locked,
                    guard,
                    key,
                    value,
                    crate::ttl::TtlAt::Derive,
                )?;
                let ref_ptr = mr.ref_ptr;
                *ev_out = tree.defer_or_retire_event(key, mr);
                Ok(ref_ptr)
            })?
        };
        let guard = self.tree.index.collector().enter();
        Ok(ref_ptr.map(|p| TypedRef::new(guard, p)))
    }

    pub fn insert(&mut self, key: &K, value: T) -> DbResult<()> {
        self.check_shard(key)?;
        let tree = self.tree;
        let sid = self.held.sid;
        let expires_at = if H::EXPIRES {
            tree.hook.deadline(key, &value)
        } else {
            None
        };
        let guard = &self.guard;
        Ctx::with_shard(&mut self.held, tree, sid, |locked, ev_out| {
            let mr = tree.insert_locked::<true>(sid, locked, guard, key, value, expires_at)?;
            *ev_out = tree.defer_or_retire_event(key, mr);
            Ok(())
        })
    }

    pub fn delete(&mut self, key: &K) -> DbResult<Option<TypedRef<'_, T>>> {
        self.check_shard(key)?;
        let tree = self.tree;
        let sid = self.held.sid;
        let ref_ptr = {
            let guard = &self.guard;
            Ctx::with_shard(&mut self.held, tree, sid, |locked, ev_out| {
                let mr = tree.delete_locked_inner::<true>(sid, locked, guard, key)?;
                let ref_ptr = mr.ref_ptr;
                *ev_out = tree.defer_or_retire_event(key, mr);
                Ok(ref_ptr)
            })?
        };
        let guard = self.tree.index.collector().enter();
        Ok(ref_ptr.map(|p| TypedRef::new(guard, p)))
    }

    pub fn try_get(&self, key: &K) -> DbResult<Option<&T>> {
        self.check_shard(key)?;
        match self.tree.index.get(key.as_bytes(), &self.guard) {
            Some(node) => Ok(Some(&node.load_data().value)),
            None => Ok(None),
        }
    }

    pub fn get_or_err(&self, key: &K) -> DbResult<&T> {
        self.try_get(key)?.ok_or(DbError::KeyNotFound)
    }

    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 the value for `key`. Returns `Some(TypedRef)`
    /// to the **new** value if the key existed, `None` otherwise.
    pub fn update(
        &mut self,
        key: &K,
        f: impl FnOnce(&T) -> T,
    ) -> DbResult<Option<TypedRef<'_, T>>> {
        self.update_inner(key, f, false)
    }

    /// Like [`update`](Self::update), but returns `Some(TypedRef)` to the **old** value.
    pub fn fetch_update(
        &mut self,
        key: &K,
        f: impl FnOnce(&T) -> T,
    ) -> DbResult<Option<TypedRef<'_, T>>> {
        self.update_inner(key, f, true)
    }

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

    fn update_inner(
        &mut self,
        key: &K,
        f: impl FnOnce(&T) -> T,
        return_old: bool,
    ) -> DbResult<Option<TypedRef<'_, T>>> {
        self.check_shard(key)?;
        let tree = self.tree;
        let sid = self.held.sid;
        let ref_ptr = {
            let guard = &self.guard;
            Ctx::with_shard(&mut self.held, tree, sid, |locked, ev_out| {
                let ptrs = match tree.update_locked_inner(sid, locked, guard, key, f)? {
                    Some(p) => p,
                    None => return Ok(None),
                };
                let ref_ptr = if return_old { ptrs.old } else { ptrs.new };
                let mr = MutationResult {
                    ref_ptr: None,
                    hook: Some((
                        Some(unsafe { &(*ptrs.old).value as *const T }),
                        Some(unsafe { &(*ptrs.new).value as *const T }),
                    )),
                    retire: Some(ptrs.retire),
                };
                *ev_out = tree.defer_or_retire_event(key, mr);
                Ok(Some(ref_ptr))
            })?
        };
        let guard = self.tree.index.collector().enter();
        Ok(ref_ptr.map(|p| TypedRef::new(guard, p)))
    }

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

impl<K: Key, T: Send + Sync, C: Codec<T> + Sync, H: TypedWriteHook<K, T>>
    TypedShard<'_, K, T, C, 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 `TypedTree`. Returned by `iter()`, `range()`, and `prefix_iter()`.
///
/// Weakly-consistent: concurrent inserts/updates may be visible during iteration.
/// Deleted entries are skipped.
///
/// `Item` is `(K, TypedRef<'a, T>)`: the iterator's own guard protects the walk
/// over the skiplist, and **each yielded row carries a second guard of its own**,
/// so a row collected out of the iterator keeps its value alive after the
/// iterator is dropped. Yielding a bare `&'a T` (tied to the collection borrow
/// rather than to a guard) let safe code read reclaimed memory — issue #123.
/// [`TypedRef`] derefs to `&T` and is `!Send`; use `clone_value()` to move a
/// value across threads.
pub struct TypedIter<'a, K: Key, T: Send + Sync> {
    list: &'a SkipList<TypedNode<K, T>>,
    front: *mut TypedNode<K, T>,
    /// `None` = not yet resolved (lazy). Computed on first `next_back()` call.
    back: Option<*mut TypedNode<K, T>>,
    end: Bound<K>,
    start: Bound<K>,
    reversed: bool,
    done: bool,
    _guard: seize::LocalGuard<'a>,
}

impl<'a, K: Key, T: Send + Sync> Iterator for TypedIter<'a, K, T> {
    type Item = (K, TypedRef<'a, T>);

    fn next(&mut self) -> Option<Self::Item> {
        let node = self.step_front()?;
        // SAFETY: `step_front` only yields live nodes, pinned by the guard.
        let node = unsafe { &*node };
        Some((node.key, self.row_ref(node)))
    }
}

impl<K: Key, T: Send + Sync> DoubleEndedIterator for TypedIter<'_, K, T> {
    fn next_back(&mut self) -> Option<Self::Item> {
        let node = self.step_back()?;
        // SAFETY: `step_back` only yields live nodes, pinned by the guard.
        let node = unsafe { &*node };
        Some((node.key, self.row_ref(node)))
    }
}

impl<K: Key, T: Send + Sync> TypedIter<'_, K, T> {
    /// Advance the front cursor to the next live, in-bounds node.
    ///
    /// Returns a raw pointer rather than a reference so callers can borrow
    /// `self` again (to take a guard, or to keep walking) while holding it —
    /// the node stays alive because the iterator's guard does.
    fn step_front(&mut self) -> Option<*const TypedNode<K, T>> {
        loop {
            if self.done || self.front.is_null() {
                return None;
            }
            let node = unsafe { &*self.front };
            let converged = self.back.is_some_and(|back| std::ptr::eq(self.front, back));
            self.front = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
            if converged {
                self.done = true;
            }
            if node.is_marked() {
                if converged {
                    return None;
                }
                continue;
            }
            if !self.check_end(&node.key) {
                self.done = true;
                return None;
            }
            return Some(node as *const TypedNode<K, T>);
        }
    }

    /// Advance the back cursor to the previous live, in-bounds node.
    fn step_back(&mut self) -> Option<*const TypedNode<K, T>> {
        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.list.find_last_lt(key.as_bytes(), &self._guard));
            if converged {
                self.done = true;
            }
            if node.is_marked() {
                if converged {
                    return None;
                }
                continue;
            }
            if !self.check_start(&key) {
                self.done = true;
                return None;
            }
            return Some(node as *const TypedNode<K, T>);
        }
    }

    /// Visit every remaining row **without** building a [`TypedRef`] per row.
    ///
    /// `f` borrows the value for the duration of the call only; the iterator's
    /// guard pins it, so no per-row guard is needed. That makes this the cheap
    /// scan path — a plain `for` loop over `Item = (K, TypedRef<T>)` pays one
    /// `seize` guard per row (issue #123), which this avoids.
    ///
    /// Return [`ControlFlow::Break`] to stop early. Early exit lives inside the
    /// closure because `Iterator` adapters (`take`, `filter`, …) wrap the
    /// iterator in a type that no longer has this method.
    ///
    /// Walks in the iterator's forward direction. For a reverse scan use
    /// `scope::<N>(…).desc().iter().for_each_ref(…)`, which threads the
    /// direction through the scan itself.
    ///
    /// ```ignore
    /// let mut total = 0;
    /// tree.range(&start, &end).for_each_ref(|_key, value| {
    ///     total += value.amount;
    ///     ControlFlow::Continue(())
    /// });
    /// ```
    pub fn for_each_ref(mut self, mut f: impl FnMut(K, &T) -> std::ops::ControlFlow<()>) {
        while let Some(node) = self.step_front() {
            // SAFETY: the node is live and pinned by this iterator's guard,
            // which outlives the `&T` handed to `f`.
            let node = unsafe { &*node };
            if f(node.key, &node.load_data().value).is_break() {
                return;
            }
        }
    }

    /// [`for_each_ref`](Self::for_each_ref) walking backwards — the seam's
    /// against-storage direction.
    pub(crate) fn for_each_ref_back(
        mut self,
        mut f: impl FnMut(K, &T) -> std::ops::ControlFlow<()>,
    ) {
        while let Some(node) = self.step_back() {
            // SAFETY: as in `for_each_ref`.
            let node = unsafe { &*node };
            if f(node.key, &node.load_data().value).is_break() {
                return;
            }
        }
    }
}

impl<'a, K: Key, T: Send + Sync> TypedIter<'a, K, T> {
    /// Wrap a node's value in a [`TypedRef`] carrying **its own** guard.
    ///
    /// The iterator's own guard keeps `node` alive while this runs, and the
    /// fresh guard extends that protection to the returned row — so a row that
    /// outlives the iterator still pins its `TypedData<T>` (issue #123).
    fn row_ref(&self, node: &TypedNode<K, T>) -> TypedRef<'a, T> {
        let guard = self.list.collector().enter();
        TypedRef::new(guard, node.load_data() as *const TypedData<T>)
    }
}

impl<K: Key, T: Send + Sync> TypedIter<'_, K, T> {
    /// Lazily resolve the back pointer for DoubleEndedIterator.
    fn resolve_back(&self) -> *mut TypedNode<K, T> {
        match &self.end {
            Bound::Unbounded => self.list.find_last(&self._guard),
            Bound::Excluded(k) => self.list.find_last_lt(k.as_bytes(), &self._guard),
            Bound::Included(k) => {
                let ge = self.list.find_first_ge(k.as_bytes(), &self._guard);
                if !ge.is_null()
                    && !unsafe { &*ge }.is_marked()
                    && unsafe { &*ge }.key_bytes() == k.as_bytes()
                {
                    ge
                } else {
                    self.list.find_last_lt(k.as_bytes(), &self._guard)
                }
            }
        }
    }

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

    /// Check if key is within the start bound (backward direction).
    #[inline(always)]
    fn check_start(&self, key: &K) -> bool {
        match &self.start {
            Bound::Unbounded => true,
            Bound::Excluded(s) => {
                if self.reversed {
                    key.as_bytes() < s.as_bytes()
                } else {
                    key.as_bytes() > s.as_bytes()
                }
            }
            Bound::Included(s) => {
                if self.reversed {
                    key.as_bytes() <= s.as_bytes()
                } else {
                    key.as_bytes() >= s.as_bytes()
                }
            }
        }
    }
}

impl<'a, K: Key, T: Send + Sync> TypedIter<'a, K, T> {
    /// Collect all remaining entries. Convenience for backward compatibility.
    pub fn collect_vec(&mut self) -> Vec<(K, TypedRef<'a, T>)> {
        self.collect()
    }
}

#[cfg(feature = "armour")]
impl<T, C, H> crate::armour::collection::Collection for TypedTree<T::SelfId, T, C, H>
where
    T: crate::CollectionMeta + Clone + Send + Sync,
    C: crate::Codec<T> + Sync,
    H: crate::hook::TypedWriteHook<T::SelfId, T>,
    T::SelfId: crate::Key + Ord,
{
    fn name(&self) -> &str {
        T::NAME
    }
    fn len(&self) -> usize {
        self.len()
    }
    fn compact(&self) -> crate::DbResult<usize> {
        self.compact()
    }
    fn flush(&self) -> crate::DbResult<()> {
        self.flush_buffers()?;
        if self.engine.hints() {
            self.sync_hints()?;
        }
        Ok(())
    }
    fn periodic_flush(&self) -> crate::DbResult<()> {
        self.flush_buffers()
    }
    fn durable_flush(&self) -> crate::DbResult<()> {
        TypedTree::durable_flush(self)
    }
    fn clean_shutdown(&self) -> crate::DbResult<()> {
        TypedTree::clean_shutdown(self)
    }
    fn sweep_expired(
        &self,
        now_ms: u64,
        budget: usize,
    ) -> crate::DbResult<crate::ttl::SweepOutcome> {
        self.sweep_expired(now_ms, budget)
    }
    fn set_ttl_sweep(&self, enabled: bool) -> crate::DbResult<()> {
        self.set_ttl_sweep(enabled)
    }
    fn rebuild_ttl_index(&self) -> crate::DbResult<()> {
        self.rebuild_ttl_index()
    }
}

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

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

    fn put_bytes(&self, key: &[u8], value: &[u8]) -> DbResult<()> {
        if key.len() != 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() != 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;
        let _guard = self.index.collector().enter();
        let mut current = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        while !current.is_null() {
            let node = unsafe { &*current };
            current = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
            if node.is_marked() {
                continue;
            }
            if err.is_some() {
                break;
            }
            let data = node.load_data();
            let mut buf = Vec::new();
            match self
                .codec()
                .encode_to(&data.value, &mut buf)
                .and_then(|()| f(node.key.as_bytes(), &buf))
            {
                Ok(()) => {}
                Err(e) => err = Some(e),
            }
        }
        match err {
            Some(e) => Err(e),
            None => Ok(()),
        }
    }

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

#[cfg(feature = "typed-tree")]
impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    T: armour_core::GetType
        + crate::tree_meta::CollectionMeta
        + Clone
        + PartialEq
        + Send
        + Sync
        + 'static,
    C: Codec<T> + Sync + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::projection::ProjectableIndexTx for TypedTree<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::tree_meta::CollectionMeta
        + Clone
        + PartialEq
        + Send
        + Sync
        + 'static,
    C: Codec<T> + Sync + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::projection::IndexKeyed for TypedTree<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(test)]
mod tests {
    use super::*;
    use crate::codec::Codec;
    use crate::config::Config;
    use crate::hook::TypedWriteHook;
    use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};

    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 TRecHook {
        writes: AtomicUsize,
        #[allow(clippy::type_complexity)]
        seq: crate::sync::Mutex<Vec<(u64, Option<u64>, Option<u64>)>>,
    }

    impl TypedWriteHook<[u8; 8], u64> for TRecHook {
        fn on_write(&self, key: &[u8; 8], old: Option<&u64>, new: Option<&u64>) {
            self.writes.fetch_add(1, AtomicOrdering::Relaxed);
            crate::sync::lock(&self.seq).push((
                u64::from_be_bytes(*key),
                old.copied(),
                new.copied(),
            ));
        }
    }

    fn open_typed_hooked<H: TypedWriteHook<[u8; 8], u64>>(
        dir: &std::path::Path,
        cfg: Config,
        hook: H,
    ) -> TypedTree<[u8; 8], u64, U64Codec, H> {
        TypedTree::open_hooked(dir, cfg, U64Codec, hook).expect("open hooked")
    }

    #[test]
    fn typed_tree_atomic_fires_hooks_in_order() {
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        let tree = open_typed_hooked(dir.path(), cfg, TRecHook::default());
        let k = 5u64.to_be_bytes();
        tree.atomic(&k, |s| {
            s.put(&k, 10)?; // new key
            s.put(&k, 20)?; // update, old=10
            s.delete(&k)?; // delete, old=20
            Ok(())
        })
        .expect("atomic");
        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 3);
        let seq = crate::sync::lock(&tree.hook.seq).clone();
        assert_eq!(seq[0], (5, None, Some(10)));
        assert_eq!(seq[1], (5, Some(10), Some(20)));
        assert_eq!(seq[2], (5, Some(20), None));
    }

    #[test]
    fn typed_tree_atomic_fires_for_applied_on_err() {
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        let tree = open_typed_hooked(dir.path(), cfg, TRecHook::default());
        let k = 1u64.to_be_bytes();
        let r: DbResult<()> = tree.atomic(&k, |s| {
            s.put(&k, 99)?;
            Err(DbError::KeyNotFound)
        });
        assert!(r.is_err());
        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 1);
    }

    #[test]
    fn typed_get_many_clone_aligned() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Config::test();
        let tree = TypedTree::<[u8; 8], String, crate::RapiraCodec>::open(
            dir.path(),
            cfg,
            crate::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(),
            2u64.to_be_bytes(),
        ];
        let got = tree.get_many(&keys);
        assert_eq!(
            got,
            vec![
                Some("b".to_string()),
                None,
                Some("a".to_string()),
                Some("b".to_string())
            ]
        );
    }

    #[test]
    fn typed_update_many_set_keep_delete() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Config::test();
        let tree = TypedTree::<[u8; 8], String, crate::RapiraCodec>::open(
            dir.path(),
            cfg,
            crate::RapiraCodec,
        )
        .unwrap();
        tree.insert(&1u64.to_be_bytes(), "x".to_string()).unwrap();
        tree.insert(&2u64.to_be_bytes(), "y".to_string()).unwrap();

        let items: Vec<([u8; 8], &str)> = vec![
            (1u64.to_be_bytes(), "x2"),
            (2u64.to_be_bytes(), ""),    // delete
            (3u64.to_be_bytes(), "new"), // insert
            (4u64.to_be_bytes(), ""),    // 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 {
                    let _ = cur;
                    Upsert::Put(p.to_string())
                }
            })
            .unwrap();
        assert_eq!(
            out[0].1,
            Applied::Written {
                old: Some("x".to_string()),
                new: "x2".to_string()
            }
        );
        assert_eq!(out[1].1, Applied::Deleted("y".to_string()));
        assert_eq!(
            out[2].1,
            Applied::Written {
                old: None,
                new: "new".to_string()
            }
        );
        assert_eq!(out[3].1, Applied::Kept);

        assert_eq!(
            tree.get(&1u64.to_be_bytes()).as_deref().cloned(),
            Some("x2".to_string())
        );
        assert!(tree.get(&2u64.to_be_bytes()).is_none());
        assert_eq!(
            tree.get(&3u64.to_be_bytes()).as_deref().cloned(),
            Some("new".to_string())
        );
    }

    #[test]
    fn typed_tree_shard_update_and_fetch_update() {
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        let tree: TypedTree<[u8; 8], u64, U64Codec> =
            TypedTree::open(dir.path(), cfg, U64Codec).unwrap();
        let key = 1u64.to_be_bytes();
        tree.put(&key, 10).unwrap();

        let (new_v, old_v, missing) = tree
            .atomic(&key, |s| {
                let new_v = s.update(&key, |old| old + 1)?.as_deref().copied(); // new = 11
                let old_v = s.fetch_update(&key, |old| old + 1)?.as_deref().copied(); // old = 11, writes 12
                let missing = s.update(&2u64.to_be_bytes(), |old| *old)?.is_none();
                Ok((new_v, old_v, missing))
            })
            .expect("atomic");
        assert_eq!(new_v, Some(11));
        assert_eq!(old_v, Some(11));
        assert!(missing);
        assert_eq!(tree.get(&key).as_deref().copied(), Some(12));
    }

    #[test]
    fn typed_tree_atomic_update_fires_hook() {
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        let tree = open_typed_hooked(dir.path(), cfg, TRecHook::default());
        let key = 1u64.to_be_bytes();
        tree.put(&key, 10).unwrap();

        tree.atomic(&key, |s| {
            s.update(&key, |old| old + 1)?; // 10 -> 11
            Ok(())
        })
        .expect("atomic");

        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 2); // put + update
        let seq = crate::sync::lock(&tree.hook.seq).clone();
        // First write was the outer put (key=1, old=None, new=10)
        assert_eq!(seq[0], (1, None, Some(10)));
        // Second write was the atomic update (key=1, old=10, new=11)
        assert_eq!(seq[1], (1, Some(10), Some(11)));
        assert_eq!(tree.get(&key).as_deref().copied(), Some(11));
    }
}