armdb 0.5.3

pub mod iter;
pub mod node;

use std::ptr;

#[cfg(feature = "loom")]
use loom::sync::atomic::{AtomicUsize, Ordering};
#[cfg(not(feature = "loom"))]
use std::sync::atomic::{AtomicUsize, Ordering};

use crate::sync::{self, Mutex};
use seize::Collector;
#[cfg(not(feature = "loom"))]
use seize::Guard;

pub use self::node::SkipNode;

/// Identity pass-through over a tower pointer. Mark state lives in a
/// dedicated `AtomicBool` on the node; tower pointers no longer carry
/// tag bits. Kept under this name as a compatibility shim for traversal
/// call sites.
#[inline(always)]
pub(crate) fn strip_mark<T>(ptr: *mut T) -> *mut T {
    ptr
}

/// Result of an insert operation.
pub enum InsertResult<'g, N> {
    /// The key was newly inserted.
    Inserted,
    /// The key already existed. Returns a reference to the existing node.
    Exists(&'g N),
}

/// Search frame: predecessor and raw pointer to its successor at each level,
/// plus the first key-equal non-marked node (null if the key is absent).
///
/// NOTE: no derive(Clone, Copy) — that would add a bound `N: Copy` which
/// nodes do not satisfy; frame is passed move/borrow everywhere.
struct Frame<N> {
    preds: [*mut N; node::MAX_HEIGHT],
    succs: [*mut N; node::MAX_HEIGHT],
    found: *mut N,
}

impl<N> Frame<N> {
    fn empty() -> Self {
        Self {
            preds: [ptr::null_mut(); node::MAX_HEIGHT],
            succs: [ptr::null_mut(); node::MAX_HEIGHT],
            found: ptr::null_mut(),
        }
    }
}

/// A concurrent skip list using `seize` for memory reclamation.
///
/// **Concurrency model:**
/// - Reads (`get`, iterators) are lock-free, protected by `seize::Guard`.
/// - Writes (`insert`, `remove`) use an optimistic pre-search OUTSIDE the
///   internal `write_lock`, then re-validate and link under the lock; a
///   stale frame falls back to a locked search with helping. The lock is
///   contended when different shards insert new keys concurrently, but the
///   critical section is O(node_height) validation + linking, not the
///   full O(log n) search.
/// - Tower slots are mutated ONLY by the `write_lock` holder (plain-store
///   linking + helping CAS) — the single-linker invariant. Pre-search and
///   readers never write to towers.
/// - Same-key operations need no external serialization: `Exists`/`None`
///   fast paths linearize at the pre-search observation; duplicate
///   same-key inserts are caught by a guaranteed level-0 validation
///   failure and routed through the locked fallback.
pub struct SkipList<N: SkipNode> {
    /// Sentinel head node. Has MAX_HEIGHT, never contains real data, never removed.
    head: *mut N,
    collector: Collector,
    len: AtomicUsize,
    height: AtomicUsize,
    write_lock: Mutex<()>,
    reversed: bool,
}

// SAFETY: SkipList is designed for concurrent access. Head is a stable allocation.
// All node access is protected by seize guards.
unsafe impl<N: SkipNode> Send for SkipList<N> {}
unsafe impl<N: SkipNode> Sync for SkipList<N> {}

impl<N: SkipNode> SkipList<N> {
    pub fn new(reversed: bool) -> Self {
        let head = N::alloc_head();
        Self {
            head,
            collector: Collector::new(),
            len: AtomicUsize::new(0),
            height: AtomicUsize::new(1),
            write_lock: Mutex::new(()),
            reversed,
        }
    }

    /// Compare keys respecting the `reversed` flag.
    #[inline(always)]
    fn key_cmp(&self, a: &[u8], b: &[u8]) -> std::cmp::Ordering {
        if self.reversed { b.cmp(a) } else { a.cmp(b) }
    }

    /// Search with helping: unlinks marked nodes via CAS along the way.
    ///
    /// MUST be called strictly under `write_lock`: helping-CAS and plain
    /// store by the linker must not race (invariant "one linker" — only the
    /// lock holder mutates tower slots).
    ///
    /// When a key-equal node is found at some level, the descent continues to
    /// level 0 performing helping-CAS at each lower level. This is intentional:
    /// it mirrors the remove path's need for correct preds/succs at all levels,
    /// and matches the old insert behavior of collecting the full frame before
    /// linking.
    fn search_locked(&self, key: &[u8], search_height: usize) -> Frame<N> {
        let mut frame = Frame::empty();
        let mut current = self.head;
        for level in (0..search_height).rev() {
            loop {
                let next = strip_mark(unsafe { (*current).tower(level).load(Ordering::Acquire) });
                if next.is_null() {
                    break;
                }
                let next_ref = unsafe { &*next };
                if next_ref.is_marked() {
                    // Help unlink marked nodes (allowed: we hold write_lock)
                    let after = strip_mark(unsafe { (*next).tower(level).load(Ordering::Acquire) });
                    let _ = unsafe {
                        (*current).tower(level).compare_exchange(
                            next,
                            after,
                            Ordering::AcqRel,
                            Ordering::Relaxed,
                        )
                    };
                    continue;
                }
                match self.key_cmp(next_ref.key_bytes(), key) {
                    std::cmp::Ordering::Less => current = next,
                    std::cmp::Ordering::Equal => {
                        frame.found = next;
                        break;
                    }
                    std::cmp::Ordering::Greater => break,
                }
            }
            frame.preds[level] = current;
            // This post-loop re-read is safe ONLY because the whole search runs under
            // `write_lock` (world frozen); `presearch` must NOT do this — see the
            // `level_succ` comment there.  Unifying these two paths would re-introduce
            // the ordering bug fixed in c791565.
            frame.succs[level] =
                strip_mark(unsafe { (*current).tower(level).load(Ordering::Acquire) });
        }
        frame
    }

    /// Optimistic read-only search: the same top-down descent as `search_locked`,
    /// but with NO writes to tower slots — no helping-CAS. Marked nodes are skipped
    /// by following their tower pointers (same as `get`).
    ///
    /// Safe to call without `write_lock`. The returned frame is a snapshot: by the
    /// time the lock is acquired it may be stale and MUST be re-validated with
    /// `validate_link`/`validate_unlink` before linking.
    ///
    /// `frame.found` is the first key-equal node observed as non-marked during the
    /// descent; fast-path `Exists`/`None` decisions linearize at that observation
    /// point. A concurrent remover may mark it after presearch returns — the
    /// locked validation step (`validate_link`/`validate_unlink`) re-checks this.
    ///
    /// Helping-CAS outside the lock would race with the plain store performed by the
    /// lock holder (two writers to one slot → lost node). presearch is strictly
    /// read-only.
    fn presearch(&self, key: &[u8], search_height: usize) -> Frame<N> {
        let mut frame = Frame::empty();
        // SAFETY: descent starts at head (stable allocation) and follows only
        // tower pointers observed under the caller's seize guard — nodes may be
        // unlinked but not freed until all guards exit. Tower pointers of marked
        // nodes remain valid: unlink does not clear outgoing pointers.
        let mut current = self.head;
        for level in (0..search_height).rev() {
            // `level_succ` captures the successor pointer at the exact point
            // where the descent stops. Using this value (rather than re-reading
            // (*current).tower(level) after the loop) ensures preds[level] and
            // succs[level] form a consistent snapshot: both come from the same
            // sequence of loads. A post-loop re-read is a separate atomic
            // operation that may observe a value written by a concurrent insert
            // that raced between the loop's break and the re-read. That would
            // produce an internally inconsistent frame — preds chosen on the old
            // successor, succs pointing at the newly-inserted node — which passes
            // validate_link but corrupts list order (e.g. 5 linked before 4).
            let mut level_succ: *mut N = ptr::null_mut();
            loop {
                let mut next =
                    strip_mark(unsafe { (*current).tower(level).load(Ordering::Acquire) });
                // Read-only skip of marked nodes: helping here would race with
                // the plain store by the write_lock holder (two writers, one slot).
                while !next.is_null() && unsafe { &*next }.is_marked() {
                    next = strip_mark(unsafe { (*next).tower(level).load(Ordering::Acquire) });
                }
                if next.is_null() {
                    // level_succ stays null_mut
                    break;
                }
                let next_ref = unsafe { &*next };
                match self.key_cmp(next_ref.key_bytes(), key) {
                    std::cmp::Ordering::Less => current = next,
                    std::cmp::Ordering::Equal => {
                        frame.found = next;
                        level_succ = next;
                        break;
                    }
                    std::cmp::Ordering::Greater => {
                        level_succ = next;
                        break;
                    }
                }
            }
            frame.preds[level] = current;
            frame.succs[level] = level_succ;
        }
        frame
    }

    /// Re-validates an optimistic frame before linking a new node of height
    /// `node_height`. MUST be called under `write_lock`: no other mutator can
    /// run concurrently, so a successful result cannot be invalidated before
    /// the caller completes the link.
    ///
    /// A null successor is valid — it means the node is being inserted at the
    /// tail of that level (R1 from the spec review). The mark check is only
    /// applied to non-null successors to avoid UB from a null dereference.
    ///
    /// Caller guarantees `frame.preds` and `frame.succs` are populated for
    /// levels `0..node_height` (frame built by a `presearch`/`search_locked`
    /// with `search_height >= node_height`).
    fn validate_link(frame: &Frame<N>, node_height: usize) -> bool {
        for level in 0..node_height {
            let pred = frame.preds[level];
            let succ = frame.succs[level];
            // SAFETY: pred/succ were reachable during presearch; the caller
            // holds a seize guard — nodes cannot be freed even if unlinked.
            unsafe {
                if (*pred).is_marked() {
                    return false;
                }
                if !succ.is_null() && (*succ).is_marked() {
                    return false;
                }
                if strip_mark((*pred).tower(level).load(Ordering::Acquire)) != succ {
                    return false;
                }
            }
        }
        true
    }

    /// Re-validates an optimistic frame before unlinking `found` of height
    /// `node_height`. Same locking contract as `validate_link`.
    ///
    /// Caller guarantees `frame.preds` is populated for levels `0..node_height`
    /// (frame built by a `presearch`/`search_locked` with `search_height >= node_height`).
    fn validate_unlink(frame: &Frame<N>, found: *mut N, node_height: usize) -> bool {
        for level in 0..node_height {
            let pred = frame.preds[level];
            // SAFETY: `pred` was reachable from head during presearch; the caller's
            // seize guard keeps it alive even if it has since been unlinked. We do
            // not dereference `succ` here — only `pred` is loaded and compared.
            unsafe {
                if (*pred).is_marked() {
                    return false;
                }
                if strip_mark((*pred).tower(level).load(Ordering::Acquire)) != found {
                    return false;
                }
            }
        }
        true
    }

    /// Collector getter (for creating guards externally).
    pub fn collector(&self) -> &Collector {
        &self.collector
    }

    /// Head sentinel pointer (for iteration from the beginning).
    pub fn head_ptr(&self) -> *mut N {
        self.head
    }

    /// Returns the approximate number of entries in the skip list.
    ///
    /// The count is updated with `Relaxed` ordering and may be stale
    /// under concurrent inserts or removes.
    pub fn len(&self) -> usize {
        self.len.load(Ordering::Relaxed)
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Lock-free lookup. Returns a reference valid for the lifetime of the guard.
    pub fn get<'g>(&self, key: &[u8], guard: &'g seize::LocalGuard<'_>) -> Option<&'g N> {
        let _ = guard; // lifetime anchor
        let mut current = self.head;
        let h = self.height.load(Ordering::Relaxed);

        // Traverse from the top level down
        for level in (0..h).rev() {
            loop {
                let mut next =
                    strip_mark(unsafe { (*current).tower(level).load(Ordering::Acquire) });
                while !next.is_null() && unsafe { &*next }.is_marked() {
                    next = strip_mark(unsafe { (*next).tower(level).load(Ordering::Acquire) });
                }
                if next.is_null() {
                    break;
                }
                let next_ref = unsafe { &*next };
                match self.key_cmp(next_ref.key_bytes(), key) {
                    std::cmp::Ordering::Less => current = next,
                    std::cmp::Ordering::Equal => {
                        return Some(next_ref);
                    }
                    std::cmp::Ordering::Greater => break,
                }
            }
        }
        None
    }

    /// Insert a node into the skip list.
    ///
    /// Optimistic path: predecessors are located WITHOUT `write_lock` (read-only
    /// pre-search); only the validation and linking step runs under the lock.
    /// On a stale frame, falls back to a locked search with helping.
    ///
    /// If the key already exists (not marked) at the time of the pre-search
    /// observation, `Exists` is returned without acquiring the lock — linearized
    /// at that observation point.
    ///
    /// Concurrent inserts of the same key cannot produce a duplicate: the loser's
    /// level-0 validation is guaranteed to fail (the winner changed the
    /// preds[0]→succs[0] chain), the fallback locked search then finds the winner
    /// and returns `Exists`.
    ///
    /// `node` must be a valid pointer from `SkipNode::alloc`, exclusively owned
    /// by the caller until inserted. The module is `pub` only for loom/miri/
    /// bench-internals builds; collections uphold this contract internally.
    #[allow(clippy::not_unsafe_ptr_arg_deref)]
    pub fn insert<'g>(
        &self,
        node: *mut N,
        guard: &'g seize::LocalGuard<'_>,
    ) -> InsertResult<'g, N> {
        let key = unsafe { (*node).key_bytes() };
        let node_height = unsafe { (*node).height() } as usize;

        // Height bump — atomic CAS loop, safe outside the lock: readers carry
        // height without linked nodes (head tower slots remain null for new levels
        // until a node is actually linked there).
        let mut current_height = self.height.load(Ordering::Relaxed);
        while node_height > current_height {
            match self.height.compare_exchange_weak(
                current_height,
                node_height,
                Ordering::Relaxed,
                Ordering::Relaxed,
            ) {
                Ok(_) => break,
                Err(h) => current_height = h,
            }
        }
        let search_height = self.height.load(Ordering::Relaxed);

        // Optimistic pre-search outside the lock (read-only, no helping).
        let frame = self.presearch(key, search_height);
        if !frame.found.is_null() {
            return InsertResult::Exists(unsafe { &*frame.found });
        }

        self.insert_with_frame(node, node_height, search_height, frame, guard)
    }

    /// Locked phase of `insert`: validate the optimistic frame, fall back to a
    /// locked search on mismatch, then link the node.
    ///
    /// Exposed as a separate method so tests can supply a deterministically stale
    /// frame to exercise the fallback path.
    ///
    /// Under `write_lock` no other mutator runs concurrently, so a frame that
    /// passes `validate_link` cannot become stale before the linking below
    /// completes. For a same-key race, the loser's level-0 validation
    /// deterministically fails (winner already changed preds[0]→succs[0]);
    /// the fallback locked search then finds the winner and returns `Exists`.
    ///
    /// Guard contract: the frame must have been collected under the SAME guard
    /// (or one still alive) — the guard keeps every frame pointer alive across
    /// the lock wait.
    ///
    /// Caller guarantees `frame.found` is null — a frame with a non-null `found`
    /// must take the `Exists` fast path instead.
    fn insert_with_frame<'g>(
        &self,
        node: *mut N,
        node_height: usize,
        search_height: usize,
        frame: Frame<N>,
        guard: &'g seize::LocalGuard<'_>,
    ) -> InsertResult<'g, N> {
        debug_assert!(
            frame.found.is_null(),
            "insert_with_frame requires a frame with no found node"
        );
        let _ = guard;
        let key = unsafe { (*node).key_bytes() };
        let _lock = sync::lock(&self.write_lock);

        // Under the global write_lock no other mutator can run: a successful
        // validate_link result cannot be invalidated before the linking below.
        let frame = if Self::validate_link(&frame, node_height) {
            frame
        } else {
            let locked = self.search_locked(key, search_height);
            if !locked.found.is_null() {
                // A same-key insert won the race between pre-search and lock acquisition.
                return InsertResult::Exists(unsafe { &*locked.found });
            }
            locked
        };

        // Link the new node bottom-up: set node's outgoing pointers first (Relaxed),
        // then publish by updating predecessors (Release so the node is visible).
        #[allow(clippy::needless_range_loop)]
        for level in 0..node_height {
            unsafe {
                (*node)
                    .tower(level)
                    .store(frame.succs[level], Ordering::Relaxed);
            }
        }
        #[allow(clippy::needless_range_loop)]
        for level in 0..node_height {
            unsafe {
                (*frame.preds[level])
                    .tower(level)
                    .store(node, Ordering::Release);
            }
        }

        self.len.fetch_add(1, Ordering::Relaxed);
        InsertResult::Inserted
    }

    /// Remove a node by key. Returns the removed node pointer (for value extraction).
    ///
    /// Optimistic path mirrors `insert`: read-only pre-search without the lock;
    /// a miss returns `None` without ever acquiring the lock (linearized at the
    /// pre-search observation point — the key was absent or already logically
    /// deleted).
    pub fn remove(&self, key: &[u8], guard: &seize::LocalGuard<'_>) -> Option<*mut N> {
        let search_height = self.height.load(Ordering::Relaxed);

        let frame = self.presearch(key, search_height);
        if frame.found.is_null() {
            return None;
        }

        self.remove_with_frame(key, search_height, frame, guard)
    }

    /// Locked phase of `remove`: validate the optimistic frame, fall back to a
    /// locked search on mismatch, then mark → unlink → retire.
    ///
    /// Exposed as a separate method so tests can supply a deterministically stale
    /// frame to exercise the fallback path.
    ///
    /// Guard contract: the frame must have been collected under the SAME guard
    /// (or one still alive) — the guard keeps every frame pointer alive across
    /// the lock wait.
    fn remove_with_frame(
        &self,
        key: &[u8],
        search_height: usize,
        mut frame: Frame<N>,
        guard: &seize::LocalGuard<'_>,
    ) -> Option<*mut N> {
        let _ = guard;
        let _lock = sync::lock(&self.write_lock);

        // SAFETY: frame.found was observed reachable during presearch; the caller
        // holds a seize guard, so the node cannot be freed even if a concurrent
        // remover unlinked and retired it while we waited for the lock.
        let found_height = unsafe { (*frame.found).height() } as usize;
        // A concurrent taller same-key insert may have completed between the
        // pre-lock `search_height` load and our presearch: `found` then has
        // levels the frame never populated (preds above `search_height` are
        // null — validating them would be a null deref). Treat such a frame
        // as stale (PR #43 review, R1).
        if found_height > search_height || !Self::validate_unlink(&frame, frame.found, found_height)
        {
            // Re-load the height UNDER the lock: a linked node's height bump
            // precedes its linker's unlock (Release), so our lock acquire sees
            // a height >= the height of every node `search_locked` can find.
            // The fallback frame therefore covers every level of `found`,
            // unlike the pre-lock `search_height`.
            let locked_height = self.height.load(Ordering::Relaxed);
            let locked = self.search_locked(key, locked_height);
            if locked.found.is_null() {
                // The observed node was removed while we waited for the lock.
                return None;
            }
            debug_assert!(
                unsafe { (*locked.found).height() } as usize <= locked_height,
                "under the lock the tree height must cover every linked node"
            );
            frame = locked;
        }
        let found = frame.found;

        // Logical deletion.
        let node_ref = unsafe { &*found };
        if !node_ref.mark() {
            // Defensive invariant: under the write lock the only marker is the
            // current remove; a same-key concurrent remove is filtered out earlier
            // by validate/fallback, so reaching here unmarked should not happen in
            // practice.
            return None;
        }

        // Physical unlink top-down. After a successful validation every CAS is
        // guaranteed to succeed; after a fallback the preds are fresh. A failing
        // CAS simply leaves the node for future helping.
        let node_h = node_ref.height() as usize;
        for level in (0..node_h).rev() {
            debug_assert!(
                !frame.preds[level].is_null(),
                "unlink pred must be populated for every level of the node"
            );
            let next = strip_mark(unsafe { (*found).tower(level).load(Ordering::Acquire) });
            let _ = unsafe {
                (*frame.preds[level]).tower(level).compare_exchange(
                    found,
                    next,
                    Ordering::AcqRel,
                    Ordering::Relaxed,
                )
            };
        }

        // Retire via seize (no-op under loom: leak is acceptable)
        #[cfg(not(feature = "loom"))]
        unsafe {
            guard.defer_retire(found, N::reclaim);
        }

        self.len.fetch_sub(1, Ordering::Relaxed);
        Some(found)
    }

    /// Find the first node whose key is >= `start_key`. Used by iterators.
    pub(crate) fn find_first_ge(&self, start_key: &[u8], guard: &seize::LocalGuard<'_>) -> *mut N {
        let _ = guard;
        let mut current = self.head;
        let h = self.height.load(Ordering::Relaxed);

        for level in (0..h).rev() {
            loop {
                let mut next =
                    strip_mark(unsafe { (*current).tower(level).load(Ordering::Acquire) });
                while !next.is_null() && unsafe { &*next }.is_marked() {
                    next = strip_mark(unsafe { (*next).tower(level).load(Ordering::Acquire) });
                }
                if next.is_null() {
                    break;
                }
                let next_ref = unsafe { &*next };
                if self.key_cmp(next_ref.key_bytes(), start_key) == std::cmp::Ordering::Less {
                    current = next;
                } else {
                    break;
                }
            }
        }

        let mut result = strip_mark(unsafe { (*current).tower(0).load(Ordering::Acquire) });
        while !result.is_null() && unsafe { &*result }.is_marked() {
            result = strip_mark(unsafe { (*result).tower(0).load(Ordering::Acquire) });
        }
        result
    }

    /// Find the last node whose key is strictly less than `key` in list order.
    /// Returns null if no such node exists.
    /// Caller must hold a seize guard.
    pub(crate) fn find_last_lt(&self, key: &[u8], _guard: &seize::LocalGuard<'_>) -> *mut N {
        let mut current = self.head;
        let h = self.height.load(Ordering::Relaxed);

        for level in (0..h).rev() {
            loop {
                let mut next =
                    strip_mark(unsafe { (*current).tower(level).load(Ordering::Acquire) });
                while !next.is_null() && unsafe { &*next }.is_marked() {
                    next = strip_mark(unsafe { (*next).tower(level).load(Ordering::Acquire) });
                }
                if next.is_null() {
                    break;
                }
                let next_ref = unsafe { &*next };
                if self.key_cmp(next_ref.key_bytes(), key) == std::cmp::Ordering::Less {
                    current = next;
                } else {
                    break;
                }
            }
        }

        if current == self.head {
            ptr::null_mut()
        } else {
            current
        }
    }

    /// Find the last node in list order. Returns null if the list is empty.
    /// Caller must hold a seize guard.
    pub(crate) fn find_last(&self, _guard: &seize::LocalGuard<'_>) -> *mut N {
        let mut current = self.head;
        let h = self.height.load(Ordering::Relaxed);

        for level in (0..h).rev() {
            loop {
                let mut next =
                    strip_mark(unsafe { (*current).tower(level).load(Ordering::Acquire) });
                while !next.is_null() && unsafe { &*next }.is_marked() {
                    next = strip_mark(unsafe { (*next).tower(level).load(Ordering::Acquire) });
                }
                if next.is_null() {
                    break;
                }
                current = next;
            }
        }

        if current == self.head {
            ptr::null_mut()
        } else {
            current
        }
    }
}

/// Integration-test shims (`miri_skiplist`); lib unit tests call `find_*` directly.
#[cfg(any(test, miri))]
#[allow(dead_code)]
impl<N: SkipNode> SkipList<N> {
    pub fn test_find_first_ge(&self, start_key: &[u8], guard: &seize::LocalGuard<'_>) -> *mut N {
        self.find_first_ge(start_key, guard)
    }

    pub fn test_find_last_lt(&self, key: &[u8], guard: &seize::LocalGuard<'_>) -> *mut N {
        self.find_last_lt(key, guard)
    }

    pub fn test_find_last(&self, guard: &seize::LocalGuard<'_>) -> *mut N {
        self.find_last(guard)
    }
}

impl<N: SkipNode> Drop for SkipList<N> {
    fn drop(&mut self) {
        // Walk level 0 and free all VLA-allocated nodes (including head)
        let mut current = self.head;
        while !current.is_null() {
            let next = strip_mark(unsafe { (*current).tower(0).load(Ordering::Relaxed) });
            unsafe {
                N::dealloc_node(current);
            }
            current = next;
        }
    }
}

#[cfg(test)]
mod s1_marked_traversal_tests {
    use super::*;
    use crate::DiskLoc;
    use crate::skiplist::node::ConstNode;

    type TestNode = ConstNode<[u8; 4], 8>;
    type TestList = SkipList<TestNode>;

    fn disk() -> DiskLoc {
        DiskLoc::new(0, 0, 0)
    }

    fn alloc_node_h(key: [u8; 4], value: [u8; 8], height: u8) -> *mut TestNode {
        TestNode::alloc(key, value, disk(), height)
    }

    #[test]
    fn single_get_skips_marked_node() {
        let list = TestList::new(false);
        let guard = list.collector().enter();

        let n3 = alloc_node_h([0, 0, 0, 3], [3u8; 8], 1);
        let n6 = alloc_node_h([0, 0, 0, 6], [6u8; 8], 2);
        list.insert(n3, &guard);
        list.insert(n6, &guard);

        let node6 = list.get(&[0, 0, 0, 6], &guard).unwrap();
        node6.mark();

        let found = list.get(&[0, 0, 0, 3], &guard);
        assert!(
            found.is_some(),
            "get(3) must not be blocked by marked node 6"
        );
        assert_eq!(found.unwrap().read_value(), [3u8; 8]);
    }

    #[test]
    fn single_find_first_ge_skips_marked_node() {
        let list = TestList::new(false);
        let guard = list.collector().enter();

        let n3 = alloc_node_h([0, 0, 0, 3], [3u8; 8], 1);
        let n6 = alloc_node_h([0, 0, 0, 6], [6u8; 8], 2);
        list.insert(n3, &guard);
        list.insert(n6, &guard);

        let node6 = list.get(&[0, 0, 0, 6], &guard).unwrap();
        node6.mark();

        let found = list.find_first_ge(&[0, 0, 0, 3], &guard);
        assert!(!found.is_null(), "find_first_ge(3) must find node 3");
        assert_eq!(unsafe { &*found }.key_bytes(), &[0, 0, 0, 3]);
    }

    #[test]
    fn single_find_last_lt_skips_marked_node() {
        let list = TestList::new(false);
        let guard = list.collector().enter();

        let n5 = alloc_node_h([0, 0, 0, 5], [5u8; 8], 1);
        let n6 = alloc_node_h([0, 0, 0, 6], [6u8; 8], 2);
        list.insert(n5, &guard);
        list.insert(n6, &guard);

        let node6 = list.get(&[0, 0, 0, 6], &guard).unwrap();
        node6.mark();

        let found = list.find_last_lt(&[0, 0, 0, 7], &guard);
        assert!(!found.is_null(), "find_last_lt(7) must find node 5");
        let found_ref = unsafe { &*found };
        assert!(
            !found_ref.is_marked(),
            "find_last_lt must not return a marked node"
        );
        assert_eq!(found_ref.key_bytes(), &[0, 0, 0, 5]);
    }

    #[test]
    fn single_find_last_skips_marked_node() {
        let list = TestList::new(false);
        let guard = list.collector().enter();

        let n3 = alloc_node_h([0, 0, 0, 3], [3u8; 8], 1);
        let n5 = alloc_node_h([0, 0, 0, 5], [5u8; 8], 1);
        let n8 = alloc_node_h([0, 0, 0, 8], [8u8; 8], 2);
        list.insert(n3, &guard);
        list.insert(n5, &guard);
        list.insert(n8, &guard);

        let node8 = list.get(&[0, 0, 0, 8], &guard).unwrap();
        node8.mark();

        let found = list.find_last(&guard);
        assert!(!found.is_null(), "find_last must find a live node");
        let found_ref = unsafe { &*found };
        assert!(
            !found_ref.is_marked(),
            "find_last must not return a marked node"
        );
        assert_eq!(found_ref.key_bytes(), &[0, 0, 0, 5]);
    }

    /// S-1 regression: find_first_ge must not return a marked node from the
    /// return path. Setup: A(1) → B(3, marked) → C(5, live).
    /// find_first_ge(5) traversal sets current=A (skipping B), then returns
    /// A.tower(0) which is B — a marked node with key < start_key.
    #[test]
    fn find_first_ge_return_path_skips_marked_node() {
        let list = TestList::new(false);
        let guard = list.collector().enter();

        let na = alloc_node_h([0, 0, 0, 1], [1u8; 8], 1);
        let nb = alloc_node_h([0, 0, 0, 3], [3u8; 8], 1);
        let nc = alloc_node_h([0, 0, 0, 5], [5u8; 8], 1);
        list.insert(na, &guard);
        list.insert(nb, &guard);
        list.insert(nc, &guard);

        list.get(&[0, 0, 0, 3], &guard).unwrap().mark();

        let found = list.find_first_ge(&[0, 0, 0, 5], &guard);
        assert!(!found.is_null(), "find_first_ge(5) must find node 5");
        let found_ref = unsafe { &*found };
        assert!(
            !found_ref.is_marked(),
            "find_first_ge must not return a marked node"
        );
        assert_eq!(found_ref.key_bytes(), &[0, 0, 0, 5]);
    }

    /// S-1 regression: chain of consecutive marked nodes must be skipped entirely.
    /// Setup: key 2 (height 1, live), key 4 (height 2, marked), key 6 (height 2, marked),
    /// key 8 (height 1, live). find_last_lt(10) must return node 8, not a marked node.
    #[test]
    fn find_last_lt_skips_consecutive_marked_chain() {
        let list = TestList::new(false);
        let guard = list.collector().enter();

        let n2 = alloc_node_h([0, 0, 0, 2], [2u8; 8], 1);
        let n4 = alloc_node_h([0, 0, 0, 4], [4u8; 8], 2);
        let n6 = alloc_node_h([0, 0, 0, 6], [6u8; 8], 2);
        let n8 = alloc_node_h([0, 0, 0, 8], [8u8; 8], 1);
        list.insert(n2, &guard);
        list.insert(n4, &guard);
        list.insert(n6, &guard);
        list.insert(n8, &guard);

        list.get(&[0, 0, 0, 4], &guard).unwrap().mark();
        list.get(&[0, 0, 0, 6], &guard).unwrap().mark();

        let found = list.find_last_lt(&[0, 0, 0, 10], &guard);
        assert!(!found.is_null());
        let found_ref = unsafe { &*found };
        assert!(!found_ref.is_marked());
        assert_eq!(found_ref.key_bytes(), &[0, 0, 0, 8]);
    }
}

#[cfg(test)]
mod optimistic_tests {
    use super::*;
    use crate::DiskLoc;
    use crate::skiplist::node::ConstNode;

    type TestNode = ConstNode<[u8; 4], 8>;
    type TestList = SkipList<TestNode>;

    fn disk() -> DiskLoc {
        DiskLoc::new(0, 0, 0)
    }

    fn alloc_node_h(key: [u8; 4], value: [u8; 8], height: u8) -> *mut TestNode {
        TestNode::alloc(key, value, disk(), height)
    }

    fn search_height(list: &TestList) -> usize {
        list.height.load(Ordering::Relaxed)
    }

    #[test]
    fn presearch_collects_preds_succs_and_found() {
        let list = TestList::new(false);
        let guard = list.collector().enter();
        let n1 = alloc_node_h([0, 0, 0, 1], [1u8; 8], 1);
        let n5 = alloc_node_h([0, 0, 0, 5], [5u8; 8], 1);
        list.insert(n1, &guard);
        list.insert(n5, &guard);
        let h = search_height(&list);

        // Missing key between two nodes
        let frame = list.presearch(&[0, 0, 0, 3], h);
        assert!(frame.found.is_null());
        assert_eq!(frame.preds[0], n1);
        assert_eq!(frame.succs[0], n5);

        // Existing key
        let frame = list.presearch(&[0, 0, 0, 5], h);
        assert_eq!(frame.found, n5);

        // Key past the tail: succ == null
        let frame = list.presearch(&[0, 0, 0, 9], h);
        assert!(frame.found.is_null());
        assert_eq!(frame.preds[0], n5);
        assert!(frame.succs[0].is_null());
    }

    #[test]
    fn presearch_skips_marked_without_helping() {
        // 1 -> 3(marked) -> 5: presearch(4) must skip 3 as a reader,
        // but must NOT unlink it (read-only) — 1.tower(0) must remain == 3.
        let list = TestList::new(false);
        let guard = list.collector().enter();
        let n1 = alloc_node_h([0, 0, 0, 1], [1u8; 8], 1);
        let n3 = alloc_node_h([0, 0, 0, 3], [3u8; 8], 1);
        let n5 = alloc_node_h([0, 0, 0, 5], [5u8; 8], 1);
        list.insert(n1, &guard);
        list.insert(n3, &guard);
        list.insert(n5, &guard);
        list.get(&[0, 0, 0, 3], &guard).unwrap().mark();
        let h = search_height(&list);

        let frame = list.presearch(&[0, 0, 0, 4], h);
        assert!(frame.found.is_null());
        assert_eq!(
            frame.preds[0], n1,
            "pred must be the last non-marked node < key"
        );
        // succs[0] is the first live (non-marked) successor: n5, not the
        // raw marked n3. The key invariant is that n1.tower(0) still points
        // at n3 — presearch never wrote to it.
        assert_eq!(frame.succs[0], n5, "succs must be the first live successor");
        assert_eq!(
            unsafe { (*n1).tower(0).load(Ordering::Acquire) },
            n3,
            "presearch must not unlink marked nodes"
        );
    }

    #[test]
    fn presearch_treats_marked_equal_as_absent() {
        // Key exists but node is marked → found == null (logically deleted).
        let list = TestList::new(false);
        let guard = list.collector().enter();
        let n3 = alloc_node_h([0, 0, 0, 3], [3u8; 8], 1);
        list.insert(n3, &guard);
        list.get(&[0, 0, 0, 3], &guard).unwrap().mark();
        let h = search_height(&list);

        let frame = list.presearch(&[0, 0, 0, 3], h);
        assert!(frame.found.is_null());
    }

    #[test]
    fn validate_link_accepts_fresh_frame_and_null_succ() {
        // R1 from spec review: null succ (insert at tail of level) is valid, not an error.
        let list = TestList::new(false);
        let guard = list.collector().enter();
        let n1 = alloc_node_h([0, 0, 0, 1], [1u8; 8], 1);
        list.insert(n1, &guard);
        let h = search_height(&list);

        let frame = list.presearch(&[0, 0, 0, 5], h);
        assert!(frame.succs[0].is_null());
        assert!(TestList::validate_link(&frame, 1));
    }

    #[test]
    fn validate_link_detects_stale_pred_tower() {
        // Frame for key 3 in {1, 5}; then insert 2 — 1.tower(0)
        // changed, validation must fail.
        let list = TestList::new(false);
        let guard = list.collector().enter();
        let n1 = alloc_node_h([0, 0, 0, 1], [1u8; 8], 1);
        let n5 = alloc_node_h([0, 0, 0, 5], [5u8; 8], 1);
        list.insert(n1, &guard);
        list.insert(n5, &guard);
        let h = search_height(&list);

        let frame = list.presearch(&[0, 0, 0, 3], h);
        assert!(TestList::validate_link(&frame, 1), "fresh frame is valid");

        let n2 = alloc_node_h([0, 0, 0, 2], [2u8; 8], 1);
        list.insert(n2, &guard);
        assert!(!TestList::validate_link(&frame, 1), "stale frame must fail");
    }

    #[test]
    fn validate_link_detects_marked_pred_and_marked_succ() {
        let list = TestList::new(false);
        let guard = list.collector().enter();
        let n1 = alloc_node_h([0, 0, 0, 1], [1u8; 8], 1);
        let n5 = alloc_node_h([0, 0, 0, 5], [5u8; 8], 1);
        list.insert(n1, &guard);
        list.insert(n5, &guard);
        let h = search_height(&list);

        // marked succ: frame(3) over {1,5}, then mark(5)
        let frame = list.presearch(&[0, 0, 0, 3], h);
        list.get(&[0, 0, 0, 5], &guard).unwrap().mark();
        assert!(!TestList::validate_link(&frame, 1));

        // marked pred: build frame directly — pred = n5 (already marked)
        let frame2 = Frame {
            preds: {
                let mut p = [std::ptr::null_mut(); node::MAX_HEIGHT];
                p[0] = n5;
                p
            },
            succs: [std::ptr::null_mut(); node::MAX_HEIGHT],
            found: std::ptr::null_mut(),
        };
        assert!(
            !TestList::validate_link(&frame2, 1),
            "marked pred must fail"
        );
    }

    #[test]
    fn validate_unlink_detects_stale_and_accepts_fresh() {
        let list = TestList::new(false);
        let guard = list.collector().enter();
        let n1 = alloc_node_h([0, 0, 0, 1], [1u8; 8], 1);
        let n3 = alloc_node_h([0, 0, 0, 3], [3u8; 8], 1);
        list.insert(n1, &guard);
        list.insert(n3, &guard);
        let h = search_height(&list);

        let frame = list.presearch(&[0, 0, 0, 3], h);
        assert_eq!(frame.found, n3);
        assert!(TestList::validate_unlink(&frame, frame.found, 1));

        // Insert 2 between pred(1) and found(3) → stale
        let n2 = alloc_node_h([0, 0, 0, 2], [2u8; 8], 1);
        list.insert(n2, &guard);
        assert!(!TestList::validate_unlink(&frame, frame.found, 1));
    }

    #[test]
    fn insert_with_stale_frame_falls_back_and_links() {
        // Deterministic race simulation: frame is collected, then the list
        // changes (insert 2), then the locked phase receives a stale frame.
        let list = TestList::new(false);
        let guard = list.collector().enter();
        let n1 = alloc_node_h([0, 0, 0, 1], [1u8; 8], 1);
        let n5 = alloc_node_h([0, 0, 0, 5], [5u8; 8], 1);
        list.insert(n1, &guard);
        list.insert(n5, &guard);
        let h = search_height(&list);

        let frame = list.presearch(&[0, 0, 0, 3], h);
        let n2 = alloc_node_h([0, 0, 0, 2], [2u8; 8], 1);
        list.insert(n2, &guard); // makes frame stale

        let n3 = alloc_node_h([0, 0, 0, 3], [3u8; 8], 1);
        let res = list.insert_with_frame(n3, 1, h, frame, &guard);
        assert!(matches!(res, InsertResult::Inserted));

        // Chain is correct: 1 -> 2 -> 3 -> 5
        for k in [1u8, 2, 3, 5] {
            assert!(list.get(&[0, 0, 0, k], &guard).is_some(), "key {k} lost");
        }
        assert_eq!(list.len(), 4);
    }

    #[test]
    fn insert_with_stale_frame_detects_same_key_winner() {
        // Duplicate protection: while the frame was waiting, someone inserted
        // THE SAME key. Level-0 validation is guaranteed to fail, fallback finds
        // the winner → Exists.
        let list = TestList::new(false);
        let guard = list.collector().enter();
        let n1 = alloc_node_h([0, 0, 0, 1], [1u8; 8], 1);
        list.insert(n1, &guard);
        let h = search_height(&list);

        let frame = list.presearch(&[0, 0, 0, 3], h);
        let winner = alloc_node_h([0, 0, 0, 3], [7u8; 8], 1);
        list.insert(winner, &guard);

        let loser = alloc_node_h([0, 0, 0, 3], [9u8; 8], 1);
        let res = list.insert_with_frame(loser, 1, h, frame, &guard);
        match res {
            InsertResult::Exists(existing) => {
                assert_eq!(existing.read_value(), [7u8; 8], "winner is preserved");
                // The loser node is not linked — test must free it
                unsafe { TestNode::dealloc_node(loser) };
            }
            InsertResult::Inserted => panic!("duplicate must not be linked"),
        }
        assert_eq!(list.len(), 2);
    }

    #[test]
    fn insert_optimistic_tail_path_end_to_end() {
        // R1: insert at tail (null succ) goes through the optimistic path via
        // the public API.
        let list = TestList::new(false);
        let guard = list.collector().enter();
        let n1 = alloc_node_h([0, 0, 0, 1], [1u8; 8], 1);
        list.insert(n1, &guard);
        let n5 = alloc_node_h([0, 0, 0, 5], [5u8; 8], 1);
        assert!(matches!(list.insert(n5, &guard), InsertResult::Inserted));
        assert!(list.get(&[0, 0, 0, 5], &guard).is_some());
        assert_eq!(list.len(), 2);
    }

    #[test]
    fn remove_with_stale_frame_falls_back() {
        let list = TestList::new(false);
        let guard = list.collector().enter();
        let n1 = alloc_node_h([0, 0, 0, 1], [1u8; 8], 1);
        let n3 = alloc_node_h([0, 0, 0, 3], [3u8; 8], 1);
        let n5 = alloc_node_h([0, 0, 0, 5], [5u8; 8], 1);
        list.insert(n1, &guard);
        list.insert(n3, &guard);
        list.insert(n5, &guard);
        let h = search_height(&list);

        let frame = list.presearch(&[0, 0, 0, 3], h);
        assert_eq!(frame.found, n3);

        // Делает frame stale: 2 вклинивается между pred(1) и found(3)
        let n2 = alloc_node_h([0, 0, 0, 2], [2u8; 8], 1);
        list.insert(n2, &guard);

        let removed = list.remove_with_frame(&[0, 0, 0, 3], h, frame, &guard);
        assert_eq!(removed, Some(n3));
        assert!(list.get(&[0, 0, 0, 3], &guard).is_none());
        for k in [1u8, 2, 5] {
            assert!(list.get(&[0, 0, 0, k], &guard).is_some(), "key {k} lost");
        }
        assert_eq!(list.len(), 3);
    }

    #[test]
    fn remove_with_frame_returns_none_if_concurrently_removed() {
        // Узел наблюдался, но «пока ждали лок» его удалил другой remove.
        let list = TestList::new(false);
        let guard = list.collector().enter();
        let n3 = alloc_node_h([0, 0, 0, 3], [3u8; 8], 1);
        list.insert(n3, &guard);
        let h = search_height(&list);

        let frame = list.presearch(&[0, 0, 0, 3], h);
        assert_eq!(frame.found, n3);

        // Конкурент удаляет узел обычным путём
        assert!(list.remove(&[0, 0, 0, 3], &guard).is_some());

        // Наш stale frame: валидация провалится, fallback не найдёт ключ
        let removed = list.remove_with_frame(&[0, 0, 0, 3], h, frame, &guard);
        assert_eq!(removed, None);
        assert_eq!(list.len(), 0);
    }

    #[test]
    fn remove_missing_key_returns_none() {
        let list = TestList::new(false);
        let guard = list.collector().enter();
        assert!(list.remove(&[0, 0, 0, 9], &guard).is_none());
    }

    /// PR #43 review (R1): a concurrent taller same-key insert can complete
    /// between remove's pre-lock `search_height` load and its presearch. The
    /// found node then has more levels than the frame ever populated — the
    /// frame must be treated as stale (validating it would deref null preds),
    /// and the fallback must search with a height that covers the found node.
    #[test]
    fn remove_with_frame_handles_taller_found_node() {
        let list = TestList::new(false);
        let guard = list.collector().enter();
        let n1 = alloc_node_h([0, 0, 0, 1], [1u8; 8], 1);
        list.insert(n1, &guard);

        // Stale view: tree height is 1 at this point.
        let stale_h = search_height(&list);
        assert_eq!(stale_h, 1);

        // "Concurrent" taller same-key insert completes: bumps height to 3.
        let n3 = alloc_node_h([0, 0, 0, 3], [3u8; 8], 3);
        list.insert(n3, &guard);
        assert_eq!(search_height(&list), 3);

        // Presearch with the stale height finds the node but only fills
        // preds/succs for level 0.
        let frame = list.presearch(&[0, 0, 0, 3], stale_h);
        assert_eq!(frame.found, n3);
        assert!(frame.preds[1].is_null());

        // Must take the fallback (found_height > search_height) and unlink
        // the node at ALL of its levels without dereferencing null preds.
        let removed = list.remove_with_frame(&[0, 0, 0, 3], stale_h, frame, &guard);
        assert_eq!(removed, Some(n3));
        assert!(list.get(&[0, 0, 0, 3], &guard).is_none());
        assert!(list.get(&[0, 0, 0, 1], &guard).is_some());
        assert_eq!(list.len(), 1);
    }

    /// Verifies that `validate_link` checks each level independently.
    ///
    /// Scenario: list {1(h2), 9(h2)}. Capture a fresh frame for key 5 at
    /// search_height=2: preds=[n1,n1], succs=[n9,n9].
    ///
    /// Then insert 3(h2). After the insert:
    ///   n1.tower(0) = n3,  n1.tower(1) = n3   (both levels of n1 now point at n3)
    ///
    /// Build a hybrid frame: level 0 from a fresh presearch (preds[0]=n3, succs[0]=n9),
    /// level 1 from the old stale frame (preds[1]=n1, succs[1]=n9).
    ///
    /// Expected:
    ///   validate_link(hybrid, 1) → true   (only level 0 checked; n3.tower(0)==n9 ✓)
    ///   validate_link(hybrid, 2) → false  (level 1 checked; n1.tower(1)==n3 ≠ n9 ✗)
    #[test]
    fn validate_link_checks_each_level_independently() {
        let list = TestList::new(false);
        let guard = list.collector().enter();

        let n1 = alloc_node_h([0, 0, 0, 1], [1u8; 8], 2);
        let n9 = alloc_node_h([0, 0, 0, 9], [9u8; 8], 2);
        list.insert(n1, &guard);
        list.insert(n9, &guard);

        let h = search_height(&list);
        assert!(
            h >= 2,
            "list height must be >= 2 after inserting two h=2 nodes"
        );

        // Capture stale frame for key 5 before inserting 3.
        // At this point: preds=[n1,n1], succs=[n9,n9] at both levels.
        let stale = list.presearch(&[0, 0, 0, 5], h);
        assert!(
            TestList::validate_link(&stale, 2),
            "stale frame is still fresh before inserting 3"
        );

        // Insert 3(h2): both n1.tower(0) and n1.tower(1) now point at n3.
        let n3 = alloc_node_h([0, 0, 0, 3], [3u8; 8], 2);
        list.insert(n3, &guard);

        // Fresh presearch after inserting 3.
        // preds=[n3,n3], succs=[n9,n9] at both levels.
        let fresh = list.presearch(&[0, 0, 0, 5], h);
        assert!(
            TestList::validate_link(&fresh, 2),
            "fully fresh frame must be valid at height 2"
        );

        // Build a hybrid frame: level 0 is fresh, level 1 is from the stale frame.
        //   hybrid.preds[0] = n3 (fresh), succs[0] = n9  → n3.tower(0) == n9 ✓
        //   hybrid.preds[1] = n1 (stale), succs[1] = n9  → n1.tower(1) == n3 ≠ n9 ✗
        let mut hybrid = Frame {
            preds: fresh.preds,
            succs: fresh.succs,
            found: std::ptr::null_mut(),
        };
        hybrid.preds[1] = stale.preds[1]; // n1
        hybrid.succs[1] = stale.succs[1]; // n9

        assert!(
            TestList::validate_link(&hybrid, 1),
            "height=1 only checks level 0 (fresh) — must pass"
        );
        assert!(
            !TestList::validate_link(&hybrid, 2),
            "height=2 checks level 1 (stale: n1.tower(1)==n3 ≠ n9) — must fail"
        );
    }
}