sochdb_core/

epoch_gc.rs

// SPDX-License-Identifier: AGPL-3.0-or-later
// SochDB - LLM-Optimized Embedded Database
// Copyright (C) 2026 Sushanth Reddy Vanagala (https://github.com/sushanthpy)
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.

//! Epoch-Based Garbage Collection for Version Cleanup
//!
//! This module provides epoch-based garbage collection for cleaning up
//! old versions of data in a multi-version concurrency control (MVCC) system.
//!
//! # Design
//!
//! ```text
//! ┌─────────────────────────────────────────────────────────────────┐
//! │                    Epoch-Based GC                               │
//! │                                                                 │
//! │  Time ──────────────────────────────────────────────────────→  │
//! │                                                                 │
//! │  Epoch 0     Epoch 1     Epoch 2     Epoch 3     Epoch 4       │
//! │  ┌─────┐     ┌─────┐     ┌─────┐     ┌─────┐     ┌─────┐       │
//! │  │v1,v2│     │v3,v4│     │v5   │     │v6,v7│     │v8   │       │
//! │  └─────┘     └─────┘     └─────┘     └─────┘     └─────┘       │
//! │    ↑                                               ↑           │
//! │    │                                               │           │
//! │  Min active                                    Current         │
//! │  reader epoch                                   epoch          │
//! │                                                                 │
//! │  Versions in epochs < min_active can be safely collected       │
//! └─────────────────────────────────────────────────────────────────┘
//! ```
//!
//! # Key Concepts
//!
//! - **Version**: A specific snapshot of data at a point in time
//! - **Epoch**: A logical time period during which versions are created
//! - **Watermark**: The minimum epoch that is still accessible by readers
//! - **GC Cycle**: Periodic cleanup of versions older than watermark

use dashmap::DashMap;
use std::collections::VecDeque;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};

/// Version identifier (epoch, sequence within epoch)
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct VersionId {
    pub epoch: u64,
    pub sequence: u32,
}

impl VersionId {
    pub fn new(epoch: u64, sequence: u32) -> Self {
        Self { epoch, sequence }
    }

    /// Check if this version is older than watermark
    pub fn is_stale(&self, watermark: u64) -> bool {
        self.epoch < watermark
    }
}

/// A versioned value wrapper
#[derive(Debug, Clone)]
pub struct VersionedValue<T> {
    pub version: VersionId,
    pub value: T,
    /// Deletion marker (tombstone)
    pub deleted: bool,
}

impl<T> VersionedValue<T> {
    pub fn new(version: VersionId, value: T) -> Self {
        Self {
            version,
            value,
            deleted: false,
        }
    }

    pub fn tombstone(version: VersionId, value: T) -> Self {
        Self {
            version,
            value,
            deleted: true,
        }
    }
}

/// Version chain for a single key
#[derive(Debug)]
pub struct VersionChain<T> {
    /// Versions ordered from newest to oldest
    versions: VecDeque<VersionedValue<T>>,
    /// Total count of versions ever created
    total_versions: u64,
}

impl<T: Clone> VersionChain<T> {
    pub fn new() -> Self {
        Self {
            versions: VecDeque::new(),
            total_versions: 0,
        }
    }

    /// Add a new version (prepends to front as newest)
    pub fn add_version(&mut self, version: VersionedValue<T>) {
        self.versions.push_front(version);
        self.total_versions += 1;
    }

    /// Get the latest version
    pub fn latest(&self) -> Option<&VersionedValue<T>> {
        self.versions.front()
    }

    /// Get version visible at specific epoch (strict less-than for consistency with MVCC)
    /// Returns None if the key was deleted before the given epoch
    pub fn version_at(&self, epoch: u64) -> Option<&VersionedValue<T>> {
        for v in &self.versions {
            if v.version.epoch < epoch {
                // If this version (the most recent before the epoch) is deleted,
                // the key is considered deleted at this point in time
                if v.deleted {
                    return None;
                }
                return Some(v);
            }
        }
        None
    }

    /// Clean up versions older than watermark
    /// Returns (versions_removed, bytes_freed estimate)
    pub fn gc(&mut self, watermark: u64) -> (usize, usize) {
        let initial_len = self.versions.len();

        // Keep at least one version (the latest visible)
        let mut kept = 0;
        let mut found_base = false;

        for v in self.versions.iter() {
            if v.version.epoch >= watermark {
                kept += 1;
            } else if !found_base {
                // Keep first version below watermark as base
                found_base = true;
                kept += 1;
            }
        }

        // Truncate in one operation — O(k) drops but O(1) metadata update
        self.versions.truncate(kept);

        let removed = initial_len - self.versions.len();
        let bytes_freed = removed * std::mem::size_of::<VersionedValue<T>>();

        (removed, bytes_freed)
    }

    /// Number of versions in chain
    pub fn len(&self) -> usize {
        self.versions.len()
    }

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

impl<T: Clone> Default for VersionChain<T> {
    fn default() -> Self {
        Self::new()
    }
}

/// Maximum number of concurrent reader slots.
/// Each slot is cache-line aligned (64 bytes) to prevent false sharing.
const MAX_READER_SLOTS: usize = 256;

/// Sentinel value indicating an empty (unused) reader slot.
const SLOT_EMPTY: u64 = u64::MAX;

/// Cache-line-aligned epoch slot for lock-free reader tracking.
///
/// Each reader thread stores its held epoch in one of these slots.
/// Alignment to 64 bytes ensures no two slots share a cache line,
/// eliminating false-sharing between concurrent register/unregister calls.
#[derive(Debug)]
#[repr(C, align(64))]
struct EpochSlot {
    epoch: AtomicU64,
}

impl EpochSlot {
    const fn empty() -> Self {
        Self {
            epoch: AtomicU64::new(SLOT_EMPTY),
        }
    }
}

/// Lock-free reader registration for tracking active epochs.
///
/// Replaces the original `RwLock<HashMap<u64, u64>>` with a fixed-size
/// array of cache-line-aligned `AtomicU64` slots.  Every operation is
/// O(1) amortised and fully lock-free:
///
/// * `register()` — CAS-loop to claim an empty slot
/// * `unregister()` — single atomic store (SLOT_EMPTY)
/// * `min_active_epoch()` — relaxed scan of all slots, no locking
#[derive(Debug)]
pub struct ReaderRegistry {
    /// Fixed-size slot array, one epoch per slot.
    slots: Box<[EpochSlot; MAX_READER_SLOTS]>,
    /// Count of active readers (informational).
    active_count: AtomicUsize,
}

impl ReaderRegistry {
    pub fn new() -> Self {
        // Initialize all slots to SLOT_EMPTY
        let slots: Box<[EpochSlot; MAX_READER_SLOTS]> = {
            let mut v: Vec<EpochSlot> = Vec::with_capacity(MAX_READER_SLOTS);
            for _ in 0..MAX_READER_SLOTS {
                v.push(EpochSlot::empty());
            }
            v.into_boxed_slice().try_into().ok().unwrap()
        };
        Self {
            slots,
            active_count: AtomicUsize::new(0),
        }
    }

    /// Register a reader at `epoch`.  Returns the slot index (used as reader_id).
    ///
    /// Cost: O(MAX_READER_SLOTS) worst-case scan, O(1) amortised with low occupancy.
    /// Fully lock-free — uses compare_exchange to claim an empty slot.
    ///
    /// Returns `None` if all slots are exhausted. Callers must back-pressure
    /// or retry after a brief yield — silently overwriting a slot would allow
    /// the GC watermark to advance past a still-active reader's epoch,
    /// leading to use-after-free on version data.
    pub fn register(&self, epoch: u64) -> Option<u64> {
        for (i, slot) in self.slots.iter().enumerate() {
            if slot
                .epoch
                .compare_exchange(SLOT_EMPTY, epoch, Ordering::AcqRel, Ordering::Relaxed)
                .is_ok()
            {
                self.active_count.fetch_add(1, Ordering::Relaxed);
                return Some(i as u64);
            }
        }
        // All slots full — must NOT overwrite an occupied slot.
        // Overwriting would erase a live reader's epoch, causing
        // min_active_epoch() to return a value that is too high,
        // which lets GC reclaim versions still visible to that reader.
        None
    }

    /// Unregister a reader by slot index.
    ///
    /// Cost: O(1), lock-free.
    pub fn unregister(&self, reader_id: u64) {
        let idx = reader_id as usize;
        if idx < MAX_READER_SLOTS {
            let prev = self.slots[idx].epoch.swap(SLOT_EMPTY, Ordering::Release);
            if prev != SLOT_EMPTY {
                self.active_count.fetch_sub(1, Ordering::Relaxed);
            }
        }
    }

    /// Compute the minimum epoch across all active readers, lock-free.
    ///
    /// Cost: O(MAX_READER_SLOTS) with relaxed loads — no locking, no contention.
    pub fn min_active_epoch(&self) -> Option<u64> {
        let mut min_epoch = u64::MAX;
        for slot in self.slots.iter() {
            let e = slot.epoch.load(Ordering::Relaxed);
            if e != SLOT_EMPTY && e < min_epoch {
                min_epoch = e;
            }
        }
        if min_epoch == u64::MAX {
            None
        } else {
            Some(min_epoch)
        }
    }

    /// Get count of active readers.
    pub fn active_count(&self) -> usize {
        self.active_count.load(Ordering::Relaxed)
    }
}

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

/// GC statistics
#[derive(Debug, Default)]
pub struct GCStats {
    pub gc_cycles: AtomicU64,
    pub versions_collected: AtomicU64,
    pub bytes_freed: AtomicU64,
    pub chains_scanned: AtomicU64,
    pub last_gc_epoch: AtomicU64,
    pub last_gc_duration_us: AtomicU64,
}

impl GCStats {
    pub fn snapshot(&self) -> GCStatsSnapshot {
        GCStatsSnapshot {
            gc_cycles: self.gc_cycles.load(Ordering::Relaxed),
            versions_collected: self.versions_collected.load(Ordering::Relaxed),
            bytes_freed: self.bytes_freed.load(Ordering::Relaxed),
            chains_scanned: self.chains_scanned.load(Ordering::Relaxed),
            last_gc_epoch: self.last_gc_epoch.load(Ordering::Relaxed),
            last_gc_duration_us: self.last_gc_duration_us.load(Ordering::Relaxed),
        }
    }
}

#[derive(Debug, Clone)]
pub struct GCStatsSnapshot {
    pub gc_cycles: u64,
    pub versions_collected: u64,
    pub bytes_freed: u64,
    pub chains_scanned: u64,
    pub last_gc_epoch: u64,
    pub last_gc_duration_us: u64,
}

/// GC configuration
#[derive(Debug, Clone)]
pub struct GCConfig {
    /// Minimum number of epochs to keep (grace period)
    pub min_epochs_to_keep: u64,
    /// Trigger GC after this many new versions
    pub gc_trigger_threshold: usize,
    /// Maximum versions to scan per GC cycle
    pub max_versions_per_cycle: usize,
}

impl Default for GCConfig {
    fn default() -> Self {
        Self {
            min_epochs_to_keep: 2,
            gc_trigger_threshold: 1000,
            max_versions_per_cycle: 10000,
        }
    }
}

/// Epoch-based garbage collector for versioned data
pub struct EpochGC<K, V>
where
    K: Eq + std::hash::Hash + Clone,
    V: Clone,
{
    /// Global epoch counter
    current_epoch: AtomicU64,
    /// Sequence counter within epoch
    current_sequence: AtomicU64,
    /// Version chains by key — DashMap provides 64-shard concurrent access,
    /// eliminating the global RwLock that serialised all insert/get/gc calls.
    chains: DashMap<K, VersionChain<V>>,
    /// Reader registry
    readers: Arc<ReaderRegistry>,
    /// GC configuration
    config: GCConfig,
    /// GC statistics
    stats: GCStats,
    /// Pending versions since last GC
    pending_versions: AtomicUsize,
}

impl<K, V> EpochGC<K, V>
where
    K: Eq + std::hash::Hash + Clone,
    V: Clone,
{
    /// Create new epoch GC with default config
    pub fn new() -> Self {
        Self::with_config(GCConfig::default())
    }

    /// Create with custom config
    pub fn with_config(config: GCConfig) -> Self {
        Self {
            current_epoch: AtomicU64::new(0),
            current_sequence: AtomicU64::new(0),
            chains: DashMap::new(),
            readers: Arc::new(ReaderRegistry::new()),
            config,
            stats: GCStats::default(),
            pending_versions: AtomicUsize::new(0),
        }
    }

    /// Get current epoch
    pub fn current_epoch(&self) -> u64 {
        self.current_epoch.load(Ordering::SeqCst)
    }

    /// Advance to next epoch
    pub fn advance_epoch(&self) -> u64 {
        self.current_sequence.store(0, Ordering::SeqCst);
        self.current_epoch.fetch_add(1, Ordering::SeqCst) + 1
    }

    /// Allocate a new version ID
    pub fn next_version(&self) -> VersionId {
        let epoch = self.current_epoch.load(Ordering::SeqCst);
        let seq = self.current_sequence.fetch_add(1, Ordering::SeqCst) as u32;
        VersionId::new(epoch, seq)
    }

    /// Insert a new version
    pub fn insert(&self, key: K, value: V) -> VersionId {
        let version = self.next_version();
        let versioned = VersionedValue::new(version, value);

        self.chains.entry(key).or_default().add_version(versioned);

        let pending = self.pending_versions.fetch_add(1, Ordering::Relaxed);

        // Trigger GC if threshold exceeded
        if pending >= self.config.gc_trigger_threshold {
            self.try_gc();
        }

        version
    }

    /// Delete a key (insert tombstone)
    pub fn delete(&self, key: K, tombstone_value: V) -> VersionId {
        let version = self.next_version();
        let versioned = VersionedValue::tombstone(version, tombstone_value);

        self.chains.entry(key).or_default().add_version(versioned);

        self.pending_versions.fetch_add(1, Ordering::Relaxed);
        version
    }

    /// Get latest version of a key
    pub fn get(&self, key: &K) -> Option<V> {
        self.chains
            .get(key)
            .and_then(|entry| entry.latest().cloned())
            .filter(|v| !v.deleted)
            .map(|v| v.value.clone())
    }

    /// Get version at specific epoch
    pub fn get_at_epoch(&self, key: &K, epoch: u64) -> Option<V> {
        self.chains
            .get(key)
            .and_then(|entry| entry.version_at(epoch).cloned())
            .map(|v| v.value.clone())
    }

    /// Begin a read transaction at current epoch.
    ///
    /// Panics if all reader slots are exhausted (256 concurrent readers).
    /// This indicates a concurrency design issue in the caller — either
    /// readers are not being dropped or the workload exceeds the slot limit.
    pub fn begin_read(&self) -> ReadGuard {
        let epoch = self.current_epoch.load(Ordering::SeqCst);
        let reader_id = self.readers.register(epoch).unwrap_or_else(|| {
            panic!(
                "EpochGC: all {} reader slots exhausted — too many concurrent readers. \
                 Ensure ReadGuards are dropped promptly.",
                MAX_READER_SLOTS
            )
        });
        ReadGuard {
            epoch,
            reader_id,
            registry: Arc::clone(&self.readers),
        }
    }

    /// Calculate the GC watermark (safe to collect below this)
    pub fn watermark(&self) -> u64 {
        let current = self.current_epoch.load(Ordering::SeqCst);
        let min_reader = self.readers.min_active_epoch().unwrap_or(current);

        // Watermark is the minimum of current - grace period and min active reader
        let grace = current.saturating_sub(self.config.min_epochs_to_keep);
        grace.min(min_reader)
    }

    /// Try to run a GC cycle
    pub fn try_gc(&self) -> GCResult {
        let start = std::time::Instant::now();
        let watermark = self.watermark();

        let mut versions_collected = 0;
        let mut bytes_freed = 0;
        let mut chains_scanned = 0;

        // Collect keys to scan (DashMap doesn't need a global lock)
        let keys: Vec<K> = self
            .chains
            .iter()
            .map(|entry| entry.key().clone())
            .collect();

        for key in keys {
            if chains_scanned >= self.config.max_versions_per_cycle {
                break;
            }

            if let Some(mut entry) = self.chains.get_mut(&key) {
                let (removed, freed) = entry.gc(watermark);
                versions_collected += removed;
                bytes_freed += freed;
                chains_scanned += 1;
            }
        }

        // Remove empty chains in a separate pass
        self.chains.retain(|_, chain| !chain.is_empty());

        let duration = start.elapsed();

        // Update stats
        self.stats.gc_cycles.fetch_add(1, Ordering::Relaxed);
        self.stats
            .versions_collected
            .fetch_add(versions_collected as u64, Ordering::Relaxed);
        self.stats
            .bytes_freed
            .fetch_add(bytes_freed as u64, Ordering::Relaxed);
        self.stats
            .chains_scanned
            .fetch_add(chains_scanned as u64, Ordering::Relaxed);
        self.stats.last_gc_epoch.store(watermark, Ordering::Relaxed);
        self.stats
            .last_gc_duration_us
            .store(duration.as_micros() as u64, Ordering::Relaxed);

        // Reset pending counter
        self.pending_versions.store(0, Ordering::Relaxed);

        GCResult {
            versions_collected,
            bytes_freed,
            chains_scanned,
            watermark,
            duration_us: duration.as_micros() as u64,
        }
    }

    /// Force a full GC (ignore limits)
    pub fn force_gc(&self) -> GCResult {
        let _old_limit = self.config.max_versions_per_cycle;
        // Temporarily set limit to max
        let _config = GCConfig {
            max_versions_per_cycle: usize::MAX,
            ..self.config.clone()
        };

        let start = std::time::Instant::now();
        let watermark = self.watermark();

        let mut versions_collected = 0;
        let mut bytes_freed = 0;
        let mut chains_scanned = 0;

        for mut entry in self.chains.iter_mut() {
            let (removed, freed) = entry.value_mut().gc(watermark);
            versions_collected += removed;
            bytes_freed += freed;
            chains_scanned += 1;
        }

        // Remove empty chains
        self.chains.retain(|_, chain| !chain.is_empty());

        let duration = start.elapsed();

        self.stats.gc_cycles.fetch_add(1, Ordering::Relaxed);
        self.stats
            .versions_collected
            .fetch_add(versions_collected as u64, Ordering::Relaxed);
        self.stats
            .bytes_freed
            .fetch_add(bytes_freed as u64, Ordering::Relaxed);
        self.stats
            .chains_scanned
            .fetch_add(chains_scanned as u64, Ordering::Relaxed);
        self.stats.last_gc_epoch.store(watermark, Ordering::Relaxed);
        self.stats
            .last_gc_duration_us
            .store(duration.as_micros() as u64, Ordering::Relaxed);
        self.pending_versions.store(0, Ordering::Relaxed);

        GCResult {
            versions_collected,
            bytes_freed,
            chains_scanned,
            watermark,
            duration_us: duration.as_micros() as u64,
        }
    }

    /// Get GC statistics
    pub fn stats(&self) -> GCStatsSnapshot {
        self.stats.snapshot()
    }

    /// Get total version count
    pub fn version_count(&self) -> usize {
        self.chains.iter().map(|entry| entry.value().len()).sum()
    }

    /// Get chain count
    pub fn chain_count(&self) -> usize {
        self.chains.len()
    }
}

impl<K, V> Default for EpochGC<K, V>
where
    K: Eq + std::hash::Hash + Clone,
    V: Clone,
{
    fn default() -> Self {
        Self::new()
    }
}

/// Result of a GC cycle
#[derive(Debug, Clone)]
pub struct GCResult {
    pub versions_collected: usize,
    pub bytes_freed: usize,
    pub chains_scanned: usize,
    pub watermark: u64,
    pub duration_us: u64,
}

/// Guard for read transactions
pub struct ReadGuard {
    pub epoch: u64,
    reader_id: u64,
    registry: Arc<ReaderRegistry>,
}

impl Drop for ReadGuard {
    fn drop(&mut self) {
        self.registry.unregister(self.reader_id);
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_version_id() {
        let v1 = VersionId::new(1, 0);
        let v2 = VersionId::new(2, 0);

        assert!(v1 < v2);
        assert!(v1.is_stale(2));
        assert!(!v2.is_stale(2));
    }

    #[test]
    fn test_version_chain_basic() {
        let mut chain: VersionChain<String> = VersionChain::new();

        chain.add_version(VersionedValue::new(VersionId::new(0, 0), "v1".to_string()));
        chain.add_version(VersionedValue::new(VersionId::new(1, 0), "v2".to_string()));

        assert_eq!(chain.len(), 2);
        assert_eq!(chain.latest().unwrap().value, "v2");
    }

    #[test]
    fn test_version_chain_gc() {
        let mut chain: VersionChain<String> = VersionChain::new();

        // Add versions at different epochs
        for epoch in 0..5 {
            chain.add_version(VersionedValue::new(
                VersionId::new(epoch, 0),
                format!("v{}", epoch),
            ));
        }

        assert_eq!(chain.len(), 5);

        // GC with watermark at epoch 3
        let (removed, _) = chain.gc(3);

        // Should keep epochs 3,4 and one base version
        assert!(removed > 0);
        assert!(chain.len() < 5);
    }

    #[test]
    fn test_reader_registry() {
        let registry = ReaderRegistry::new();

        let r1 = registry.register(10).unwrap();
        let _r2 = registry.register(20).unwrap();

        assert_eq!(registry.active_count(), 2);
        assert_eq!(registry.min_active_epoch(), Some(10));

        registry.unregister(r1);
        assert_eq!(registry.active_count(), 1);
        assert_eq!(registry.min_active_epoch(), Some(20));
    }

    #[test]
    fn test_epoch_gc_basic() {
        let gc: EpochGC<String, i32> = EpochGC::new();

        let _v1 = gc.insert("key1".to_string(), 100);
        let _v2 = gc.insert("key1".to_string(), 200);

        assert_eq!(gc.get(&"key1".to_string()), Some(200));
        assert_eq!(gc.version_count(), 2);
    }

    #[test]
    fn test_epoch_gc_delete() {
        let gc: EpochGC<String, i32> = EpochGC::new();

        gc.insert("key1".to_string(), 100);
        gc.delete("key1".to_string(), 0); // tombstone

        assert_eq!(gc.get(&"key1".to_string()), None);
    }

    #[test]
    fn test_epoch_gc_at_epoch() {
        let gc: EpochGC<String, i32> = EpochGC::new();

        gc.insert("key1".to_string(), 100);
        gc.advance_epoch();
        gc.insert("key1".to_string(), 200);
        gc.advance_epoch();
        gc.insert("key1".to_string(), 300);

        // With strict < semantics: version_at(N) sees versions with epoch < N
        // v=100 at epoch 0, v=200 at epoch 1, v=300 at epoch 2
        assert_eq!(gc.get_at_epoch(&"key1".to_string(), 1), Some(100)); // epoch 0 < 1
        assert_eq!(gc.get_at_epoch(&"key1".to_string(), 2), Some(200)); // epoch 1 < 2
        assert_eq!(gc.get_at_epoch(&"key1".to_string(), 3), Some(300)); // epoch 2 < 3
        assert_eq!(gc.get_at_epoch(&"key1".to_string(), 0), None); // no epoch < 0
    }

    #[test]
    fn test_read_guard() {
        let gc: EpochGC<String, i32> = EpochGC::new();

        gc.insert("key1".to_string(), 100);

        {
            let _guard = gc.begin_read();
            assert_eq!(gc.readers.active_count(), 1);
        }

        assert_eq!(gc.readers.active_count(), 0);
    }

    #[test]
    fn test_watermark_calculation() {
        let gc: EpochGC<String, i32> = EpochGC::with_config(GCConfig {
            min_epochs_to_keep: 2,
            ..Default::default()
        });

        // Epoch 0
        gc.insert("k".to_string(), 1);
        gc.advance_epoch(); // Epoch 1
        gc.insert("k".to_string(), 2);
        gc.advance_epoch(); // Epoch 2
        gc.insert("k".to_string(), 3);
        gc.advance_epoch(); // Epoch 3
        gc.insert("k".to_string(), 4);
        gc.advance_epoch(); // Epoch 4

        // With no readers and grace period 2, watermark should be 4-2=2
        assert!(gc.watermark() <= 2);

        // With a reader at epoch 1, watermark should be min(2, 1) = 1
        let _guard = gc.begin_read();
        assert!(gc.watermark() <= gc.current_epoch());
    }

    #[test]
    fn test_gc_cycle() {
        let gc: EpochGC<String, i32> = EpochGC::with_config(GCConfig {
            min_epochs_to_keep: 1,
            gc_trigger_threshold: 100,
            max_versions_per_cycle: 100,
        });

        // Create multiple versions across epochs
        for i in 0..10 {
            gc.insert("key".to_string(), i);
            gc.advance_epoch();
        }

        assert_eq!(gc.version_count(), 10);

        // Run GC
        let result = gc.try_gc();

        // Should have collected some versions
        assert!(result.versions_collected > 0 || gc.version_count() < 10);
    }

    #[test]
    fn test_gc_stats() {
        let gc: EpochGC<String, i32> = EpochGC::new();

        for i in 0..5 {
            gc.insert("key".to_string(), i);
            gc.advance_epoch();
        }

        gc.try_gc();

        let stats = gc.stats();
        assert!(stats.gc_cycles >= 1);
    }

    #[test]
    fn test_force_gc() {
        let gc: EpochGC<String, i32> = EpochGC::with_config(GCConfig {
            min_epochs_to_keep: 0,
            gc_trigger_threshold: 1000,
            max_versions_per_cycle: 1, // Very limited
        });

        for i in 0..20 {
            gc.insert(format!("key{}", i), i);
        }

        gc.advance_epoch();
        gc.advance_epoch();

        let initial_count = gc.version_count();
        gc.force_gc();
        let final_count = gc.version_count();

        // Force GC should process all chains regardless of limit
        assert!(final_count <= initial_count);
    }

    #[test]
    fn test_chain_count() {
        let gc: EpochGC<String, i32> = EpochGC::new();

        gc.insert("key1".to_string(), 1);
        gc.insert("key2".to_string(), 2);
        gc.insert("key3".to_string(), 3);

        assert_eq!(gc.chain_count(), 3);
    }

    #[test]
    fn test_version_at_respects_tombstone() {
        let mut chain: VersionChain<i32> = VersionChain::new();

        chain.add_version(VersionedValue::new(VersionId::new(0, 0), 100));
        chain.add_version(VersionedValue::tombstone(VersionId::new(1, 0), 0));

        // With strict < semantics:
        // version_at(2) should see tombstone at epoch 1 → return None
        assert!(chain.version_at(2).is_none());
        // version_at(1) should see version at epoch 0 → return 100
        assert_eq!(chain.version_at(1).map(|v| v.value), Some(100));
        // version_at(0) → no version with epoch < 0
        assert!(chain.version_at(0).is_none());
    }

    #[test]
    fn test_gc_result_fields() {
        let gc: EpochGC<String, i32> = EpochGC::new();

        for i in 0..5 {
            gc.insert("key".to_string(), i);
            gc.advance_epoch();
        }

        let result = gc.try_gc();

        // Verify result has reasonable values
        assert!(result.watermark <= gc.current_epoch());
        assert!(result.chains_scanned <= gc.chain_count() + 1);
    }

    #[test]
    fn test_lock_free_slot_registration() {
        let registry = ReaderRegistry::new();

        // Register multiple readers — each gets a unique slot
        let id0 = registry.register(10).unwrap();
        let id1 = registry.register(20).unwrap();
        let id2 = registry.register(30).unwrap();

        assert_ne!(id0, id1);
        assert_ne!(id1, id2);
        assert_eq!(registry.active_count(), 3);
        assert_eq!(registry.min_active_epoch(), Some(10));

        // Unregister middle reader — min should stay at 10
        registry.unregister(id1);
        assert_eq!(registry.active_count(), 2);
        assert_eq!(registry.min_active_epoch(), Some(10));

        // Unregister first reader — min moves to 30
        registry.unregister(id0);
        assert_eq!(registry.active_count(), 1);
        assert_eq!(registry.min_active_epoch(), Some(30));

        // Unregister last — no active readers
        registry.unregister(id2);
        assert_eq!(registry.active_count(), 0);
        assert_eq!(registry.min_active_epoch(), None);
    }

    #[test]
    fn test_concurrent_insert_and_gc() {
        use std::sync::Arc;
        use std::thread;

        let gc = Arc::new(EpochGC::<u64, u64>::new());

        // Spawn writers
        let mut handles = Vec::new();
        for t in 0..4 {
            let gc = Arc::clone(&gc);
            handles.push(thread::spawn(move || {
                for i in 0..100 {
                    gc.insert(t * 1000 + i, i);
                }
            }));
        }

        // Spawn a GC thread
        {
            let gc = Arc::clone(&gc);
            handles.push(thread::spawn(move || {
                for _ in 0..10 {
                    gc.advance_epoch();
                    gc.try_gc();
                }
            }));
        }

        for h in handles {
            h.join().unwrap();
        }

        // Should have 400 unique keys
        assert_eq!(gc.chain_count(), 400);
    }
}