// Copyright 2026 PingCAP, Inc.

//! Crossbeam-backed memtable ordered map.
//!
//! `SkipMap` stores one entry per inner key. Each entry owns a `VersionChain`
//! whose immutable `ValueNode`s run newest to oldest from an atomic head.
//! Writes are serialized per skiplist; point reads and iterators remain
//! concurrent and own the nodes backing any returned raw `Value` view.
//!
//! `size()` preserves deterministic submitted-work accounting for memtable
//! switching. `memory_usage()` separately tracks retained objects until active
//! readers and Crossbeam epoch reclamation release them.

use std::{
    mem::{self, size_of},
    ops::{Bound::Included, Deref},
    ptr,
    sync::{
        Arc, Mutex,
        atomic::{AtomicBool, AtomicPtr, AtomicU64, AtomicUsize, Ordering},
    },
};

use bytes::Bytes;
use crossbeam_skiplist::{SkipMap, map::Entry};

use super::{WriteBatch, WriteBatchEntry};
use crate::{
    SnapAccess,
    table::{
        InnerKey, is_deleted,
        table::{Iterator, Value},
    },
};

// Keep submitted-size accounting aligned with the Arena implementation. For
// retained-memory accounting this is also the estimated average Crossbeam
// tower/node cost; allocator metadata and size-class rounding are excluded.
const AVG_NODE_SIZE: usize = 56;

// One immutable MVCC version. `encoded` always owns metadata and owns ordinary
// payloads. Oversized payloads stay in `separate_value` so the write batch and
// this node can share one `Bytes` allocation. `older` retains the chain suffix.
struct ValueNode {
    encoded: Bytes,
    separate_value: Option<Box<Bytes>>,
    version: u64,
    older: Option<Arc<ValueNode>>,
    memory_tracker: Arc<MemoryTracker>,
}

const ARC_ALLOCATION_OVERHEAD: usize = 2 * size_of::<usize>();
const VALUE_NODE_OVERHEAD: usize = size_of::<ValueNode>() + ARC_ALLOCATION_OVERHEAD;
const MAP_ENTRY_OVERHEAD: usize = AVG_NODE_SIZE + size_of::<Bytes>() + size_of::<VersionChain>();

#[derive(Default)]
// Shared by the map and value nodes so charges remain valid when readers
// outlive a removed key. Relaxed atomics suffice because this is accounting,
// not a publication or lifetime synchronization primitive.
struct MemoryTracker(AtomicU64);

impl MemoryTracker {
    fn add(&self, size: usize) {
        self.0.fetch_add(size as u64, Ordering::Relaxed);
    }

    fn remove(&self, size: usize) {
        self.0.fetch_sub(size as u64, Ordering::Relaxed);
    }

    fn usage(&self) -> u64 {
        self.0.load(Ordering::Relaxed)
    }
}

impl Drop for ValueNode {
    fn drop(&mut self) {
        self.memory_tracker.remove(self.allocation_size());
        // Detach uniquely owned tails iteratively to avoid recursive `Arc`
        // drops. A tail held by another reader continues reclamation later.
        let mut older = self.older.take();
        while let Some(node) = older {
            match Arc::try_unwrap(node) {
                Ok(mut node) => older = node.older.take(),
                Err(_) => break,
            }
        }
    }
}

impl ValueNode {
    fn new(
        batch: &WriteBatch,
        entry: &WriteBatchEntry,
        older: Option<Arc<ValueNode>>,
        memory_tracker: Arc<MemoryTracker>,
    ) -> Self {
        let separate_value = batch.oversized_value(entry).cloned().map(Box::new);
        let separate_value_len = separate_value.as_ref().map_or(0, |value| value.len());
        let mut encoded = Vec::with_capacity(entry.encoded_val_size() - separate_value_len);
        encoded.push(entry.meta);
        encoded.push(entry.user_meta_len);
        encoded.extend_from_slice(&entry.version.to_le_bytes());
        encoded.extend_from_slice(entry.user_meta(&batch.buf));
        if separate_value.is_none() {
            encoded.extend_from_slice(batch.value(entry));
        }
        let node = Self {
            encoded: encoded.into(),
            separate_value,
            version: entry.version,
            older,
            memory_tracker,
        };
        node.memory_tracker.add(node.allocation_size());
        node
    }

    fn allocation_size(&self) -> usize {
        let separate_value_len = self.separate_value.as_ref().map_or(0, |value| value.len());
        let boxed_bytes_overhead = usize::from(self.separate_value.is_some())
            * (size_of::<Bytes>() + ARC_ALLOCATION_OVERHEAD);
        self.encoded.len() + separate_value_len + VALUE_NODE_OVERHEAD + boxed_bytes_overhead
    }

    fn value(&self) -> Value {
        match &self.separate_value {
            Some(value) => Value::decode_with_separate_value(&self.encoded, value),
            None => Value::decode(&self.encoded),
        }
    }
}

// Per-key MVCC state. The single writer publishes an immutable head while
// readers retain whichever head or suffix they observed. `map_entry_size`
// remains charged until Crossbeam actually destroys this map value.
struct VersionChain {
    // Logically owns exactly one strong reference created by `Arc::into_raw`.
    // Readers must clone that ownership with `load_head`; they must never take
    // this reference directly. Every replacement must retain the preceding head
    // through `ValueNode::older` before publication. See `load_head` for the
    // invariants that make its load-then-increment sequence safe.
    head: AtomicPtr<ValueNode>,
    map_entry_size: usize,
}

impl VersionChain {
    fn new(
        batch: &WriteBatch,
        entry: &WriteBatchEntry,
        key_len: usize,
        memory_tracker: Arc<MemoryTracker>,
    ) -> Self {
        let map_entry_size = key_len + MAP_ENTRY_OVERHEAD;
        let head = Arc::new(ValueNode::new(batch, entry, None, memory_tracker.clone()));
        memory_tracker.add(map_entry_size);
        Self {
            head: AtomicPtr::new(Arc::into_raw(head).cast_mut()),
            map_entry_size,
        }
    }

    fn put(&self, batch: &WriteBatch, entry: &WriteBatchEntry) {
        let older = self.load_head();
        if entry.version <= older.version {
            // Restore can contain duplicate or out-of-order versions.
            return;
        }
        let memory_tracker = older.memory_tracker.clone();
        let node = Arc::new(ValueNode::new(batch, entry, Some(older), memory_tracker));
        self.store_head(node);
    }

    fn get(&self, version: u64) -> Option<Arc<ValueNode>> {
        let mut node = self.load_head();
        loop {
            if node.version <= version {
                return Some(node);
            }
            node = node.older.clone()?;
        }
    }

    fn load_head(&self) -> Arc<ValueNode> {
        let head = self.head.load(Ordering::Acquire);
        // SAFETY: loading the pointer and incrementing its strong count are not
        // one atomic operation. The pointer cannot be freed between them because:
        //
        // 1. `self` stays alive for this call. Callers obtain it from a Crossbeam
        //    `Entry` (directly or through `OwnedEntry`), which at the pinned Crossbeam
        //    revision retains the map node even after removal.
        // 2. Head replacements are serialized by `write_lock`. Before swapping, `put`
        //    makes the new head own the previous head through `older`; only then does
        //    `store_head` release the atomic's previous reference.
        //
        // The atomic logically keeps its own strong reference. Incrementing first
        // creates a separate reference for this reader; `from_raw` adopts that
        // newly created reference. Calling `from_raw` without the increment would
        // steal the atomic's reference and allow a reader to release it.
        unsafe {
            Arc::increment_strong_count(head);
            Arc::from_raw(head)
        }
    }

    fn head_ptr(&self) -> *const ValueNode {
        self.head.load(Ordering::Acquire)
    }

    fn store_head(&self, head: Arc<ValueNode>) {
        // Writes to this chain are serialized by `SkipListCore::write_lock`.
        let head = Arc::into_raw(head).cast_mut();
        let old = self.head.swap(head, Ordering::AcqRel);
        // SAFETY: `old` is the strong reference previously owned by the atomic
        // pointer, and each swap consumes that ownership exactly once.
        unsafe { drop(Arc::from_raw(old)) };
    }
}

impl Drop for VersionChain {
    fn drop(&mut self) {
        let head = *self.head.get_mut();
        // SAFETY: exclusive access prevents concurrent loads or swaps, and the
        // pointer still owns exactly one strong reference.
        let head = unsafe { Arc::from_raw(head) };
        head.memory_tracker.remove(self.map_entry_size);
    }
}

// `Value` is a borrowed raw view; `_node` owns its backing until the caller
// materializes the value into an outer owner or drops this guard.
pub(super) struct ValueGuard {
    _node: Option<Arc<ValueNode>>,
    value: Value,
}

impl ValueGuard {
    fn new(node: Option<Arc<ValueNode>>) -> Self {
        let value = node.as_ref().map_or_else(Value::new, |node| node.value());
        Self { _node: node, value }
    }

    pub(super) fn value(&self) -> &Value {
        &self.value
    }

    #[cfg(test)]
    fn is_empty(&self) -> bool {
        self.value.is_empty()
    }

    #[cfg(test)]
    fn is_valid(&self) -> bool {
        self.value.is_valid()
    }

    #[cfg(test)]
    fn is_blob_ref(&self) -> bool {
        self.value.is_blob_ref()
    }

    #[cfg(test)]
    fn get_value(&self) -> &[u8] {
        self.value.get_value()
    }
}

type MapEntry = Entry<'static, Bytes, VersionChain>;
type ValueHeadGuard = Arc<ValueNode>;

// Owns the map entry together with the map that its forged lifetime refers to.
// Its narrow API prevents the embedded `'static` references from escaping.
struct OwnedEntry {
    entry: Option<MapEntry>,
    _list: SkipList,
}

impl OwnedEntry {
    fn new(list: &SkipList, entry: Entry<'_, Bytes, VersionChain>) -> Self {
        let list = list.clone();
        // SAFETY: at pinned Crossbeam revision c4abd1b9, Entry owns an atomic
        // reference to its node independently of epoch guards. The cloned list
        // keeps the borrowed map alive, and Drop releases entry before list.
        // The forged lifetime stays encapsulated behind methods that borrow
        // OwnedEntry. Re-audit these invariants before changing the revision.
        let entry = unsafe { mem::transmute::<Entry<'_, Bytes, VersionChain>, MapEntry>(entry) };
        Self {
            entry: Some(entry),
            _list: list,
        }
    }

    fn key(&self) -> &Bytes {
        self.entry.as_ref().unwrap().key()
    }

    fn value(&self) -> &VersionChain {
        self.entry.as_ref().unwrap().value()
    }

    fn move_next(&mut self) -> bool {
        self.entry.as_mut().unwrap().move_next()
    }

    fn move_prev(&mut self) -> bool {
        self.entry.as_mut().unwrap().move_prev()
    }
}

// SAFETY: at pinned Crossbeam revision c4abd1b9, map::Entry contains only a map
// reference and an atomically reference-counted node pointer, not an epoch
// Guard or LocalHandle. Navigation and Drop pin whichever thread calls them;
// key and value are Send + Sync; and OwnedEntry's cloned map outlives entry.
// Re-audit Entry/RefEntry, navigation, release, and these tests before changing
// the pinned Crossbeam revision.
unsafe impl Send for OwnedEntry {}

impl Drop for OwnedEntry {
    fn drop(&mut self) {
        self.entry.take();
    }
}

#[derive(Clone)]
pub struct SkipList {
    core: Arc<SkipListCore>,
}

impl Deref for SkipList {
    type Target = SkipListCore;

    fn deref(&self) -> &Self::Target {
        &self.core
    }
}

impl Default for SkipList {
    fn default() -> Self {
        Self::new()
    }
}

impl SkipList {
    pub fn new() -> Self {
        Self {
            core: Arc::new(SkipListCore::new()),
        }
    }

    pub fn new_iterator(&self, reversed: bool) -> SkIterator {
        SkIterator {
            entry: None,
            value_head: None,
            older_value_node: None,
            list: self.clone(),
            reversed,
        }
    }

    /// Creates a forward iterator for flushing this permanently sealed map.
    ///
    /// Unlike the concurrent iterator, this path borrows the immutable version
    /// chain and does not clone an `Arc` for every key or older version.
    pub fn new_flush_iterator(&self) -> FlushIterator<'_> {
        assert!(
            self.is_sealed(),
            "cannot flush an unsealed Crossbeam memtable"
        );
        FlushIterator::new(&self.core)
    }

    /// Permanently prevents further writes to this map.
    pub(crate) fn seal(&self) {
        self.core.seal();
    }

    fn is_sealed(&self) -> bool {
        self.core.sealed.load(Ordering::Acquire)
    }

    pub fn has_over_bound_data(&self, start: InnerKey<'_>, end: InnerKey<'_>) -> bool {
        let mut iter = self.new_iterator(false);
        iter.rewind();
        if iter.valid() && iter.key() < start {
            return true;
        }
        let mut reverse = self.new_iterator(true);
        reverse.rewind();
        reverse.valid() && reverse.key() >= end
    }
}

pub struct SkipListCore {
    data: SkipMap<Bytes, VersionChain>,
    // Sealing is permanent and synchronized with the last writer. It allows a
    // flush iterator to borrow map values and version nodes without ownership
    // traffic while this map remains alive.
    sealed: AtomicBool,
    // Monotonic upper bound used to reject `get_newer` without a map lookup.
    data_max_ts: AtomicU64,
    // Writes are serialized so publishing a new version needs one atomic swap.
    write_lock: Mutex<()>,
    // Estimated memory retained by the map, active readers, and epoch-retired
    // nodes. Allocator metadata and allocator caches are not included.
    memory_tracker: Arc<MemoryTracker>,
    // Deterministic cumulative submitted-work counters. They intentionally do
    // not decrease on removal and stay aligned with Arena switch accounting.
    user_data_size: AtomicU64,
    user_data_entries: AtomicUsize,
}

impl SkipListCore {
    fn new() -> Self {
        Self {
            data: SkipMap::new(),
            sealed: AtomicBool::new(false),
            data_max_ts: AtomicU64::new(0),
            write_lock: Mutex::new(()),
            memory_tracker: Arc::new(MemoryTracker::default()),
            user_data_size: AtomicU64::new(0),
            user_data_entries: AtomicUsize::new(0),
        }
    }

    fn seal(&self) {
        let _write_guard = self.write_lock.lock().unwrap();
        self.sealed.store(true, Ordering::Release);
    }

    pub fn put_batch(&self, batch: &mut WriteBatch, snap: Option<&SnapAccess>, cf: usize) {
        self.put_batch_impl(batch, snap, cf, false);
    }

    /// Inserts a remotely reconstructed batch without eliding tombstones.
    pub fn put_batch_preserve_tombstones(&self, batch: &mut WriteBatch, cf: usize) {
        self.put_batch_impl(batch, None, cf, true);
    }

    fn put_batch_impl(
        &self,
        batch: &mut WriteBatch,
        snap: Option<&SnapAccess>,
        cf: usize,
        preserve_tombstones: bool,
    ) {
        let _write_guard = self.write_lock.lock().unwrap();
        assert!(
            !self.sealed.load(Ordering::Acquire),
            "cannot write a sealed Crossbeam memtable"
        );
        let data_max_ts = self.data_max_ts.load(Ordering::Acquire);
        let mut batch_max_ts = 0;
        for entry in &batch.entries {
            if is_deleted(entry.meta) {
                let key = entry.key(&batch.buf);
                // Retain a tombstone when an older source may contain the key;
                // otherwise remove it because there is nothing left to mask.
                // Async tables conservatively retain it because checking older
                // tables here would require expensive asynchronous execution.
                if preserve_tombstones
                    || snap
                        .map(|snap| {
                            !snap.is_cf_sync(cf)
                                || snap.contains_in_older_table(InnerKey::from_inner_buf(key), cf)
                        })
                        .unwrap_or_default()
                {
                    batch_max_ts = batch_max_ts.max(entry.version);
                    self.put_entry(batch, entry);
                } else {
                    self.delete(key);
                }
            } else {
                batch_max_ts = batch_max_ts.max(entry.version);
                self.put_entry(batch, entry);
            }
        }

        let data_size = batch.buffered_data_size() + batch.entries.len() * AVG_NODE_SIZE;
        let entries = batch.entries.len();
        // user_data_size remains approximate when the batch contains deletes.
        self.user_data_size
            .fetch_add(data_size as u64, Ordering::AcqRel);
        self.user_data_entries.fetch_add(entries, Ordering::AcqRel);
        if batch_max_ts > data_max_ts {
            self.data_max_ts.store(batch_max_ts, Ordering::Release);
        }
    }

    fn put_entry(&self, batch: &WriteBatch, entry: &WriteBatchEntry) {
        let key = entry.key(&batch.buf);
        let mut inserted = false;
        // write_lock prevents the closure from being discarded after a
        // competing insertion, so closure execution identifies a new key.
        let map_entry = self
            .data
            .get_or_insert_with(Bytes::copy_from_slice(key), || {
                inserted = true;
                VersionChain::new(batch, entry, key.len(), self.memory_tracker.clone())
            });
        if !inserted {
            map_entry.value().put(batch, entry);
        }
    }

    fn delete(&self, key: &[u8]) -> bool {
        // Removing the map's ownership does not invalidate readers. Their entry
        // and value guards keep data and memory charges alive until reclamation.
        let Some(_entry) = self.data.remove(key) else {
            return false;
        };
        true
    }

    pub fn get(&self, key: &[u8], version: u64) -> ValueGuard {
        ValueGuard::new(
            self.data
                .get(key)
                .and_then(|entry| entry.value().get(version)),
        )
    }

    pub fn get_newer(&self, key: &[u8], version: u64) -> ValueGuard {
        if self.data_max_ts.load(Ordering::Acquire) <= version {
            return ValueGuard::new(None);
        }
        let value = self.get(key, u64::MAX);
        if value.value.version > version {
            return value;
        }
        ValueGuard::new(None)
    }

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

    /// Returns deterministic cumulative submitted bytes used for switching.
    ///
    /// This does not decrease on removal; use `memory_usage()` for retained
    /// allocation accounting.
    pub fn size(&self) -> u64 {
        self.user_data_size.load(Ordering::Acquire)
    }

    /// Returns estimated retained bytes charged to local memory limiters.
    ///
    /// Charges remain until readers and epoch reclamation release objects.
    /// Allocator metadata, caches, and retained RSS pages are excluded.
    pub fn memory_usage(&self) -> u64 {
        self.memory_tracker.usage()
    }

    /// Returns cumulative submitted entries, not current map cardinality.
    pub fn entries(&self) -> usize {
        self.user_data_entries.load(Ordering::Acquire)
    }

    pub(crate) fn data_max_ts(&self) -> u64 {
        self.data_max_ts.load(Ordering::Acquire)
    }
}

// Owns all backing for the current key and value. Both the latest version and
// older-version traversal hold an Arc for the selected suffix. Value guards
// must drop before `OwnedEntry`.
pub struct SkIterator {
    entry: Option<OwnedEntry>,
    value_head: Option<ValueHeadGuard>,
    older_value_node: Option<Arc<ValueNode>>,
    list: SkipList,
    reversed: bool,
}

impl Drop for SkIterator {
    fn drop(&mut self) {
        // Release value views before the entry and map that provide backing.
        self.older_value_node.take();
        self.value_head.take();
        self.entry.take();
    }
}

impl SkIterator {
    fn load_entry(&mut self, entry: Option<OwnedEntry>) {
        self.older_value_node = None;
        self.value_head = None;
        self.entry = entry;
        self.value_head = self.entry.as_ref().map(|entry| entry.value().load_head());
    }

    fn current_value_node(&self) -> Option<&ValueNode> {
        self.older_value_node
            .as_deref()
            .or_else(|| self.value_head.as_ref().map(|head| head.as_ref()))
    }

    fn seek_to_first(&mut self) {
        let entry = self
            .list
            .data
            .front()
            .map(|entry| OwnedEntry::new(&self.list, entry));
        self.load_entry(entry);
    }

    fn seek_to_last(&mut self) {
        let entry = self
            .list
            .data
            .back()
            .map(|entry| OwnedEntry::new(&self.list, entry));
        self.load_entry(entry);
    }

    fn seek_for_next(&mut self, key: &[u8]) {
        let entry = self
            .list
            .data
            .lower_bound(Included(key))
            .map(|entry| OwnedEntry::new(&self.list, entry));
        self.load_entry(entry);
    }

    fn seek_for_prev(&mut self, key: &[u8]) {
        let entry = self
            .list
            .data
            .upper_bound(Included(key))
            .map(|entry| OwnedEntry::new(&self.list, entry));
        self.load_entry(entry);
    }

    fn next_forward(&mut self) {
        self.move_entry(OwnedEntry::move_next);
    }

    fn next_backward(&mut self) {
        self.move_entry(OwnedEntry::move_prev);
    }

    fn move_entry(&mut self, move_to_next: impl FnOnce(&mut OwnedEntry) -> bool) {
        self.older_value_node = None;
        self.value_head = None;
        let Some(entry) = self.entry.as_mut() else {
            return;
        };
        if !move_to_next(entry) {
            self.entry = None;
            return;
        }
        self.value_head = Some(entry.value().load_head());
    }
}

impl Iterator for SkIterator {
    fn next(&mut self) {
        if self.reversed {
            self.next_backward()
        } else {
            self.next_forward()
        }
    }

    fn next_version(&mut self) -> bool {
        if let Some(older) = self
            .current_value_node()
            .and_then(|value| value.older.clone())
        {
            self.older_value_node = Some(older);
            true
        } else {
            false
        }
    }

    fn rewind(&mut self) {
        if self.reversed {
            self.seek_to_last()
        } else {
            self.seek_to_first()
        }
    }

    fn seek(&mut self, key: InnerKey<'_>) {
        if self.reversed {
            self.seek_for_prev(key.deref())
        } else {
            self.seek_for_next(key.deref())
        }
    }

    fn key(&self) -> InnerKey<'_> {
        let key = self
            .entry
            .as_ref()
            .map_or(&[][..], |entry| entry.key().as_ref());
        InnerKey::from_inner_buf(key)
    }

    fn value(&self) -> Value {
        self.current_value_node()
            .map(|node| node.value())
            .unwrap_or_else(Value::new)
    }

    fn valid(&self) -> bool {
        self.entry.is_some() && self.value_head.is_some()
    }

    #[cfg(debug_assertions)]
    fn tag(&self) -> String {
        "skl".to_string()
    }
}

/// A forward-only iterator over a permanently sealed Crossbeam memtable.
///
/// The borrowed `Entry` keeps the current map node and its `VersionChain`
/// alive. Sealing prevents head replacement and node removal, so `current`
/// remains valid until the iterator clears it before moving or dropping the
/// entry. Each older node is owned by the preceding node's `Arc` for the full
/// lifetime of the chain.
pub struct FlushIterator<'a> {
    list: &'a SkipListCore,
    entry: Option<Entry<'a, Bytes, VersionChain>>,
    current: *const ValueNode,
}

impl<'a> FlushIterator<'a> {
    fn new(list: &'a SkipListCore) -> Self {
        Self {
            list,
            entry: None,
            current: ptr::null(),
        }
    }

    fn load_entry(&mut self, entry: Option<Entry<'a, Bytes, VersionChain>>) {
        self.current = ptr::null();
        self.entry = entry;
        self.current = self
            .entry
            .as_ref()
            .map_or(ptr::null(), |entry| entry.value().head_ptr());
    }

    fn current_node(&self) -> Option<&ValueNode> {
        // SAFETY: `current` comes from the current entry's VersionChain. The
        // entry keeps that chain alive, and the permanent seal prevents head
        // replacement or removal. Older nodes remain owned by the chain.
        unsafe { self.current.as_ref() }
    }

    fn next_entry(&mut self) {
        self.current = ptr::null();
        let Some(entry) = self.entry.as_mut() else {
            return;
        };
        if !entry.move_next() {
            self.entry = None;
            return;
        }
        self.current = entry.value().head_ptr();
    }
}

// SAFETY: at pinned Crossbeam revision c4abd1b9, `Entry` contains only a map
// reference and an atomically reference-counted node pointer; navigation and
// release pin the calling thread. `current` points into a permanently sealed,
// immutable chain owned by that entry. The borrowed map outlives the iterator.
// Re-audit Entry/RefEntry, navigation, release, and drop order before changing
// the pinned Crossbeam revision.
unsafe impl Send for FlushIterator<'_> {}

impl Iterator for FlushIterator<'_> {
    fn next(&mut self) {
        self.next_entry();
    }

    fn next_version(&mut self) -> bool {
        let Some(older) = self.current_node().and_then(|node| node.older.as_deref()) else {
            return false;
        };
        self.current = older;
        true
    }

    fn rewind(&mut self) {
        self.load_entry(self.list.data.front());
    }

    fn seek(&mut self, key: InnerKey<'_>) {
        self.load_entry(self.list.data.lower_bound(Included(key.deref())));
    }

    fn key(&self) -> InnerKey<'_> {
        let key = self
            .entry
            .as_ref()
            .map_or(&[][..], |entry| entry.key().as_ref());
        InnerKey::from_inner_buf(key)
    }

    fn value(&self) -> Value {
        self.current_node()
            .map(ValueNode::value)
            .unwrap_or_else(Value::new)
    }

    fn valid(&self) -> bool {
        self.entry.is_some() && !self.current.is_null()
    }

    #[cfg(debug_assertions)]
    fn tag(&self) -> String {
        "skl-flush".to_string()
    }
}

#[cfg(test)]
mod tests {
    use std::{
        ops::Deref,
        sync::{
            Arc, Barrier,
            atomic::{AtomicBool, Ordering},
        },
    };

    use super::*;
    use crate::table::{BIT_BLOB_REF, BIT_DELETE, InnerKey};

    fn put(list: &SkipList, key: &[u8], version: u64, value: &[u8]) {
        let mut batch = WriteBatch::new();
        batch.put(InnerKey::from_inner_buf(key), 0, &[], version, value);
        list.put_batch(&mut batch, None, 0);
    }

    #[test]
    fn test_get_versions_and_ignore_older_write() {
        let list = SkipList::new();
        put(&list, b"key", 2, b"v2");
        put(&list, b"key", 4, b"v4");
        put(&list, b"key", 3, b"ignored");

        assert!(list.get(b"missing", u64::MAX).is_empty());
        assert!(list.get(b"key", 1).is_empty());
        assert_eq!(list.get(b"key", 2).get_value(), b"v2");
        assert_eq!(list.get(b"key", 3).get_value(), b"v2");
        assert_eq!(list.get(b"key", 4).get_value(), b"v4");
        assert_eq!(list.get_newer(b"key", 3).get_value(), b"v4");
        assert!(list.get_newer(b"key", 4).is_empty());
        assert_eq!(list.data_max_ts(), 4);
    }

    #[test]
    fn test_iteration_seek_and_versions() {
        let list = SkipList::new();
        put(&list, b"a", 1, b"a1");
        put(&list, b"b", 1, b"b1");
        put(&list, b"b", 2, b"b2");
        put(&list, b"d", 1, b"d1");

        let mut iter = list.new_iterator(false);
        iter.rewind();
        assert_eq!(iter.key().deref(), b"a");
        iter.seek(InnerKey::from_inner_buf(b"b"));
        assert_eq!(iter.key().deref(), b"b");
        assert_eq!(iter.value().get_value(), b"b2");
        assert!(iter.next_version());
        assert_eq!(iter.value().get_value(), b"b1");
        assert!(!iter.next_version());
        iter.seek(InnerKey::from_inner_buf(b"c"));
        assert_eq!(iter.key().deref(), b"d");
        iter.next();
        assert!(!iter.valid());

        let mut reverse = list.new_iterator(true);
        reverse.rewind();
        assert_eq!(reverse.key().deref(), b"d");
        reverse.seek(InnerKey::from_inner_buf(b"c"));
        assert_eq!(reverse.key().deref(), b"b");
        reverse.next();
        assert_eq!(reverse.key().deref(), b"a");
    }

    fn collect_entries(iterator: &mut dyn Iterator) -> Vec<(Vec<u8>, u64, bool, Vec<u8>)> {
        let mut entries = Vec::new();
        iterator.rewind();
        while iterator.valid() {
            let value = iterator.value();
            entries.push((
                iterator.key().to_vec(),
                value.version,
                value.is_deleted(),
                value.get_value().to_vec(),
            ));
            iterator.next_all_version();
        }
        entries
    }

    #[test]
    fn test_flush_iterator_matches_regular_iterator() {
        let empty = SkipList::new();
        empty.seal();
        let mut empty_regular = empty.new_iterator(false);
        let mut empty_flush = empty.new_flush_iterator();
        assert_eq!(
            collect_entries(&mut empty_regular),
            collect_entries(&mut empty_flush)
        );

        let list = SkipList::new();
        put(&list, b"a", 1, b"a1");
        put(&list, b"b", 1, b"b1");
        put(&list, b"b", 2, b"b2");
        put(&list, b"d", 1, b"d1");
        let mut tombstone = WriteBatch::new();
        tombstone.put(
            InnerKey::from_inner_buf(b"tombstone"),
            BIT_DELETE,
            &[],
            3,
            &[],
        );
        list.put_batch_preserve_tombstones(&mut tombstone, 0);
        list.seal();
        list.seal();

        let mut regular = list.new_iterator(false);
        let mut flush = list.new_flush_iterator();
        assert_eq!(collect_entries(&mut regular), collect_entries(&mut flush));

        flush.seek(InnerKey::from_inner_buf(b"b"));
        assert_eq!(flush.key().deref(), b"b");
        assert_eq!(flush.value().get_value(), b"b2");
        assert!(flush.next_version());
        let older_value = flush.value();
        assert_eq!(older_value.get_value(), b"b1");
        flush.next();
        assert_eq!(flush.key().deref(), b"d");
        assert_eq!(flush.value().get_value(), b"d1");
        assert_eq!(older_value.get_value(), b"b1");
        flush.rewind();
        assert_eq!(flush.key().deref(), b"a");
    }

    #[test]
    fn test_flush_iterator_does_not_clone_value_nodes() {
        let list = SkipList::new();
        put(&list, b"key", 1, b"value-1");
        put(&list, b"key", 2, b"value-2");
        list.seal();

        let head = list.data.front().unwrap().value().load_head();
        let older = head.older.as_ref().unwrap().clone();
        let head_refs = Arc::strong_count(&head);
        let older_refs = Arc::strong_count(&older);

        let mut regular = list.new_iterator(false);
        regular.rewind();
        assert_eq!(Arc::strong_count(&head), head_refs + 1);
        assert!(regular.next_version());
        assert_eq!(Arc::strong_count(&older), older_refs + 1);
        drop(regular);
        assert_eq!(Arc::strong_count(&head), head_refs);
        assert_eq!(Arc::strong_count(&older), older_refs);

        let mut flush = list.new_flush_iterator();
        flush.rewind();
        assert_eq!(Arc::strong_count(&head), head_refs);
        assert!(flush.next_version());
        assert_eq!(Arc::strong_count(&older), older_refs);
    }

    #[test]
    #[should_panic(expected = "cannot write a sealed Crossbeam memtable")]
    fn test_sealed_memtable_rejects_writes() {
        let list = SkipList::new();
        list.seal();
        put(&list, b"key", 1, b"value");
    }

    #[test]
    fn test_positioned_flush_iterator_is_send() {
        fn assert_send<T: Send>() {}

        assert_send::<FlushIterator<'static>>();

        let list = SkipList::new();
        put(&list, b"a", 1, b"a1");
        put(&list, b"a", 2, b"a2");
        put(&list, b"b", 1, b"b1");
        list.seal();
        let mut iterator = list.new_flush_iterator();
        iterator.rewind();
        assert_eq!(iterator.key().deref(), b"a");
        std::thread::scope(|scope| {
            scope
                .spawn(move || {
                    assert_eq!(iterator.value().get_value(), b"a2");
                    assert!(iterator.next_version());
                    assert_eq!(iterator.value().get_value(), b"a1");
                    iterator.next();
                    assert_eq!(iterator.key().deref(), b"b");
                    assert_eq!(iterator.value().get_value(), b"b1");
                })
                .join()
                .unwrap();
        });
    }

    #[test]
    fn test_iterator_owners_are_send() {
        fn assert_send<T: Send>() {}

        assert_send::<OwnedEntry>();
        assert_send::<ValueHeadGuard>();
        assert_send::<SkIterator>();
    }

    fn assert_iterator_moves_threads_while_current_entry_is_replaced(reversed: bool) {
        let list = SkipList::new();
        put(&list, b"a", 1, b"a1");
        put(&list, b"b", 1, b"b1");
        put(&list, b"c", 1, b"c1");

        let mut iterator = list.new_iterator(reversed);
        iterator.seek(InnerKey::from_inner_buf(b"b"));
        let positioned = Arc::new(Barrier::new(2));
        let replaced = Arc::new(Barrier::new(2));
        let reader = std::thread::spawn({
            let positioned = positioned.clone();
            let replaced = replaced.clone();
            move || {
                assert_eq!(iterator.key().deref(), b"b");
                assert_eq!(iterator.value().get_value(), b"b1");
                positioned.wait();
                replaced.wait();

                assert_eq!(iterator.key().deref(), b"b");
                assert_eq!(iterator.value().get_value(), b"b1");
                iterator.next();
                let expected = if reversed { b"a" } else { b"c" };
                assert_eq!(iterator.key().deref(), expected);
            }
        });

        positioned.wait();
        let mut delete = WriteBatch::new();
        delete.put(InnerKey::from_inner_buf(b"b"), BIT_DELETE, &[], 2, &[]);
        list.put_batch(&mut delete, None, 0);
        put(&list, b"b", 3, b"b3");
        replaced.wait();
        reader.join().unwrap();
    }

    #[test]
    fn test_iterator_moves_threads_while_current_entry_is_replaced() {
        assert_iterator_moves_threads_while_current_entry_is_replaced(false);
        assert_iterator_moves_threads_while_current_entry_is_replaced(true);
    }

    #[test]
    fn test_iterator_skips_removed_entries_in_both_directions() {
        for (reversed, current, removed, expected) in [
            (false, b"a", [b"a", b"b", b"c"], b"d"),
            (true, b"d", [b"d", b"c", b"b"], b"a"),
        ] {
            let list = SkipList::new();
            for key in [b"a", b"b", b"c", b"d"] {
                put(&list, key, 1, key);
            }
            let mut iterator = list.new_iterator(reversed);
            iterator.seek(InnerKey::from_inner_buf(current));

            let mut delete = WriteBatch::new();
            for key in removed {
                delete.put(InnerKey::from_inner_buf(key), BIT_DELETE, &[], 2, &[]);
            }
            list.put_batch(&mut delete, None, 0);

            assert_eq!(iterator.key().deref(), current);
            iterator.next();
            assert_eq!(iterator.key().deref(), expected);
        }
    }

    #[test]
    fn test_delete_keeps_active_reader_memory_accounted() {
        let list = SkipList::new();
        put(&list, b"next", 1, b"next-value");
        put(&list, b"key", 1, b"value");
        let value = list.get(b"key", 1);
        let mut iter = list.new_iterator(false);
        iter.seek(InnerKey::from_inner_buf(b"key"));
        let memory_usage_before_delete = list.memory_usage();

        let mut batch = WriteBatch::new();
        batch.put(InnerKey::from_inner_buf(b"key"), BIT_DELETE, &[], 2, &[]);
        list.put_batch(&mut batch, None, 0);

        assert!(list.get(b"key", u64::MAX).is_empty());
        assert_eq!(list.memory_usage(), memory_usage_before_delete);
        put(&list, b"key", 3, b"new-value");
        assert!(list.memory_usage() > memory_usage_before_delete);
        assert_eq!(list.get(b"key", u64::MAX).get_value(), b"new-value");
        assert!(list.get(b"key", 2).is_empty());
        assert_eq!(value.get_value(), b"value");
        assert_eq!(iter.key().deref(), b"key");
        assert_eq!(iter.value().get_value(), b"value");
        iter.next();
        assert_eq!(iter.key().deref(), b"next");
    }

    #[test]
    fn test_value_guard_can_cross_threads_after_delete() {
        let list = SkipList::new();
        put(&list, b"key", 1, b"value");
        let value = list.get(b"key", u64::MAX);

        let mut batch = WriteBatch::new();
        batch.put(InnerKey::from_inner_buf(b"key"), BIT_DELETE, &[], 2, &[]);
        list.put_batch(&mut batch, None, 0);

        std::thread::spawn(move || assert_eq!(value.get_value(), b"value"))
            .join()
            .unwrap();
    }

    #[test]
    fn test_delete_deep_version_chain_without_recursive_drop() {
        const VERSION_COUNT: u64 = 100_000;

        let list = SkipList::new();
        let mut batch = WriteBatch::new();
        for version in 1..=VERSION_COUNT {
            batch.put(InnerKey::from_inner_buf(b"key"), 0, &[], version, b"value");
        }
        list.put_batch(&mut batch, None, 0);
        let value = list.get(b"key", VERSION_COUNT / 2);

        let mut delete = WriteBatch::new();
        delete.put(
            InnerKey::from_inner_buf(b"key"),
            BIT_DELETE,
            &[],
            VERSION_COUNT + 1,
            &[],
        );
        list.put_batch(&mut delete, None, 0);

        assert_eq!(value.get_value(), b"value");
        assert!(list.memory_usage() > 0);
        drop(value);
    }

    #[test]
    fn test_has_over_bound_data() {
        let list = SkipList::new();
        put(&list, b"k10", 1, b"v");
        put(&list, b"k19", 1, b"v");

        let cases = [
            (b"k00".as_slice(), b"k99".as_slice(), false),
            (b"k10".as_slice(), b"k20".as_slice(), false),
            (b"k11".as_slice(), b"k20".as_slice(), true),
            (b"k10".as_slice(), b"k19".as_slice(), true),
        ];
        for (start, end, expected) in cases {
            assert_eq!(
                list.has_over_bound_data(
                    InnerKey::from_inner_buf(start),
                    InnerKey::from_inner_buf(end),
                ),
                expected
            );
        }
    }

    #[test]
    fn test_concurrent_reads_and_writes() {
        let list = Arc::new(SkipList::new());
        let reader_list = list.clone();
        let reader = std::thread::spawn(move || {
            for _ in 0..10 {
                for i in 0u64..1000 {
                    let key = i.to_be_bytes();
                    let value = reader_list.get(&key, u64::MAX);
                    if value.is_valid() {
                        assert_eq!(value.get_value(), key);
                    }
                }
            }
        });
        for i in 0u64..1000 {
            let key = i.to_be_bytes();
            put(&list, &key, i, &key);
        }
        reader.join().unwrap();
    }

    #[test]
    fn test_concurrent_reads_and_version_updates() {
        const READER_COUNT: usize = 4;
        const VERSION_COUNT: u64 = 10_000;

        let list = Arc::new(SkipList::new());
        put(&list, b"key", 1, &1u64.to_le_bytes());
        let start = Arc::new(Barrier::new(READER_COUNT + 1));
        let finished = Arc::new(AtomicBool::new(false));
        let mut readers = Vec::with_capacity(READER_COUNT);
        for _ in 0..READER_COUNT {
            let list = list.clone();
            let start = start.clone();
            let finished = finished.clone();
            readers.push(std::thread::spawn(move || {
                start.wait();
                loop {
                    let value = list.get(b"key", u64::MAX);
                    let version = value.value().version;
                    assert_eq!(value.get_value(), version.to_le_bytes());
                    if finished.load(Ordering::Acquire) {
                        break;
                    }
                }
            }));
        }

        start.wait();
        for version in 2..=VERSION_COUNT {
            put(&list, b"key", version, &version.to_le_bytes());
        }
        finished.store(true, Ordering::Release);
        for reader in readers {
            reader.join().unwrap();
        }
        assert_eq!(
            list.get(b"key", u64::MAX).get_value(),
            VERSION_COUNT.to_le_bytes()
        );
    }

    #[test]
    fn test_concurrent_reads_and_iteration_during_reinsertion() {
        const KEY_COUNT: usize = 16;
        const READER_COUNT: usize = 4;
        const ROUND_COUNT: u64 = 10_000;

        fn value_for(key: &[u8; 8], version: u64) -> [u8; 16] {
            let mut value = [0; 16];
            value[..8].copy_from_slice(key);
            value[8..].copy_from_slice(&version.to_le_bytes());
            value
        }

        fn assert_value_matches(key: &[u8], value: &Value) {
            let encoded = value.get_value();
            assert_eq!(&encoded[..8], key);
            assert_eq!(&encoded[8..], &value.version.to_le_bytes());
        }

        let list = Arc::new(SkipList::new());
        for key_index in 0..KEY_COUNT {
            let key = (key_index as u64).to_be_bytes();
            put(&list, &key, 1, &value_for(&key, 1));
        }

        let start = Arc::new(Barrier::new(READER_COUNT + 1));
        let finished = Arc::new(AtomicBool::new(false));
        let mut readers = Vec::with_capacity(READER_COUNT);
        for reader_index in 0..READER_COUNT {
            let list = list.clone();
            let start = start.clone();
            let finished = finished.clone();
            readers.push(std::thread::spawn(move || {
                start.wait();
                let mut iteration = reader_index;
                loop {
                    let key = ((iteration % KEY_COUNT) as u64).to_be_bytes();
                    if reader_index < READER_COUNT / 2 {
                        let value = list.get(&key, u64::MAX);
                        if value.is_valid() {
                            assert_value_matches(&key, value.value());
                        }
                    } else {
                        let mut iterator = list.new_iterator(reader_index % 2 == 0);
                        iterator.seek(InnerKey::from_inner_buf(&key));
                        for _ in 0..2 {
                            if !iterator.valid() {
                                break;
                            }
                            assert_value_matches(iterator.key().deref(), &iterator.value());
                            iterator.next();
                        }
                    }
                    iteration += 1;
                    if finished.load(Ordering::Acquire) {
                        break;
                    }
                }
            }));
        }

        start.wait();
        for round in 0..ROUND_COUNT {
            let key = (round % KEY_COUNT as u64).to_be_bytes();
            let version = 2 * round + 2;
            let mut batch = WriteBatch::new();
            batch.put(
                InnerKey::from_inner_buf(&key),
                BIT_DELETE,
                &[],
                version,
                &[],
            );
            batch.put(
                InnerKey::from_inner_buf(&key),
                0,
                &[],
                version + 1,
                &value_for(&key, version + 1),
            );
            list.put_batch(&mut batch, None, 0);
            if round % 64 == 0 {
                std::thread::yield_now();
            }
        }
        finished.store(true, Ordering::Release);
        for reader in readers {
            reader.join().unwrap();
        }
    }

    #[test]
    fn test_value_larger_than_arena_limit() {
        const OLD_ARENA_LIMIT: usize = 16 * 1024 * 1024;

        let list = SkipList::new();
        let value = vec![b'x'; OLD_ARENA_LIMIT + 1];
        let mut batch = WriteBatch::new();
        batch.put(InnerKey::from_inner_buf(b"large"), 0, &[], 1, &value);
        let entry = batch.get(0);
        let batch_value_ptr = batch.value(&entry).as_ptr();
        assert!(batch.oversized_value(&entry).is_some());
        assert_ne!(batch_value_ptr, value.as_ptr());
        assert!(batch.buf.capacity() < 1024);
        batch.iterate(|entry, buf, iterated_value| {
            assert_eq!(entry.key(buf), b"large");
            assert_eq!(iterated_value.as_ptr(), batch_value_ptr);
            assert_eq!(iterated_value.len(), value.len());
        });
        list.put_batch(&mut batch, None, 0);

        let stored = list.get(b"large", 1);
        assert_eq!(stored.get_value().len(), value.len());
        assert_eq!(stored.get_value()[OLD_ARENA_LIMIT], b'x');
        assert_eq!(stored.get_value().as_ptr(), batch_value_ptr);
        batch.reset();
        assert_eq!(batch.buffered_data_size(), 0);
        assert_eq!(stored.get_value()[OLD_ARENA_LIMIT], b'x');
    }

    #[test]
    fn test_preserves_blob_ref_meta() {
        let list = SkipList::new();
        let mut batch = WriteBatch::new();
        batch.put(
            InnerKey::from_inner_buf(b"key"),
            BIT_BLOB_REF,
            &[],
            1,
            b"blob-ref",
        );
        list.put_batch(&mut batch, None, 0);

        let value = list.get(b"key", u64::MAX);
        assert!(value.is_blob_ref());
        assert_eq!(value.get_value(), b"blob-ref");
    }

    #[test]
    fn test_size_and_entries_keep_submitted_data_accounting() {
        let list = SkipList::new();
        let mut batch = WriteBatch::new();
        batch.put(InnerKey::from_inner_buf(b"a"), 0, &[], 1, b"1");
        batch.put(InnerKey::from_inner_buf(b"b"), 0, &[], 1, b"2");
        let expected_size = batch.buf.len() + batch.len() * AVG_NODE_SIZE;

        list.put_batch(&mut batch, None, 0);

        assert_eq!(list.entries(), 2);
        assert_eq!(list.size(), expected_size as u64);
        assert!(list.memory_usage() > list.size());
    }

    #[test]
    fn test_memory_tracker_releases_version_chain_with_last_reader() {
        let memory_tracker = Arc::new(MemoryTracker::default());
        let mut batch = WriteBatch::new();
        batch.put(InnerKey::from_inner_buf(b"key"), 0, &[], 1, b"value");
        let entry = &batch.entries[0];
        let chain = VersionChain::new(
            &batch,
            entry,
            entry.key(&batch.buf).len(),
            memory_tracker.clone(),
        );
        let value = chain.get(u64::MAX).unwrap();

        let chain_memory_usage = memory_tracker.usage();
        drop(chain);

        assert!(memory_tracker.usage() < chain_memory_usage);
        assert!(memory_tracker.usage() > 0);
        drop(value);
        assert_eq!(memory_tracker.usage(), 0);
    }
}