sochdb_storage/

mvcc_snapshot.rs

// Copyright 2025 Sushanth (https://github.com/sushanthpy)
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! MVCC Snapshot Plumbing (Task 3)
//!
//! Multi-Version Concurrency Control for snapshot isolation:
//! - Readers don't block writers
//! - Writers don't block readers
//! - Consistent point-in-time snapshots
//!
//! ## Version Visibility Rules
//!
//! A version is visible to transaction T if:
//! 1. xmin (creating txn) committed before T started
//! 2. xmax (deleting txn) is either:
//!    - Not set (version still active), OR
//!    - Aborted, OR
//!    - Committed after T started
//!
//! ```text
//! Version: [xmin=10, xmax=20, data]
//!
//! Transaction T (start_ts=15):
//!   - xmin=10 < 15 ✓ (created before T)
//!   - xmax=20 > 15 ✓ (deleted after T started, so still visible)
//!   → VISIBLE
//!
//! Transaction T' (start_ts=25):
//!   - xmin=10 < 25 ✓
//!   - xmax=20 < 25 ✗ (already deleted)
//!   → NOT VISIBLE
//! ```
//!
//! ## Snapshot Types
//!
//! - **Read Snapshot**: Fixed point-in-time view
//! - **Serializable Snapshot**: Tracks read/write sets for conflict detection
//!
//! ## Lock-Free MVCC (Task 2 Enhancement)
//!
//! The `LockFreeMvccManager` uses epoch-based reclamation for:
//! - Wait-free reads during visibility checks
//! - Lock-free garbage collection
//! - Reduced contention under high concurrency

use crossbeam_epoch::{self as epoch, Atomic, Owned};
use parking_lot::RwLock;
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

/// Global transaction ID type
pub type TxnId = u64;

/// Timestamp type (logical, not wall-clock)
pub type Timestamp = u64;

/// Transaction status
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TxnStatus {
    /// Transaction in progress
    Active,
    /// Transaction committed with commit timestamp
    Committed(Timestamp),
    /// Transaction aborted
    Aborted,
}

/// Version metadata for a single row version
#[derive(Debug, Clone)]
pub struct VersionInfo {
    /// Transaction that created this version
    pub xmin: TxnId,
    /// Transaction that deleted this version (or 0 if active)
    pub xmax: TxnId,
    /// Creation timestamp
    pub created_ts: Timestamp,
    /// Deletion timestamp (or MAX if active)
    pub deleted_ts: Timestamp,
}

impl VersionInfo {
    /// Create a new active version
    pub fn new(xmin: TxnId, created_ts: Timestamp) -> Self {
        Self {
            xmin,
            xmax: 0,
            created_ts,
            deleted_ts: Timestamp::MAX,
        }
    }

    /// Mark version as deleted
    pub fn delete(&mut self, xmax: TxnId, deleted_ts: Timestamp) {
        self.xmax = xmax;
        self.deleted_ts = deleted_ts;
    }

    /// Check if version is visible to a snapshot
    #[allow(deprecated)]
    pub fn is_visible(&self, snapshot: &Snapshot, txn_manager: &TransactionManager) -> bool {
        // Check xmin visibility
        if self.xmin == snapshot.txn_id {
            // Created by current transaction - visible
            if self.xmax == snapshot.txn_id {
                // Also deleted by current transaction - not visible
                return false;
            }
            return true;
        }

        // xmin must be committed before snapshot
        match txn_manager.get_status(self.xmin) {
            Some(TxnStatus::Committed(commit_ts)) => {
                if commit_ts >= snapshot.start_ts {
                    return false; // Created after snapshot
                }
            }
            Some(TxnStatus::Active) => {
                // Created by an in-progress transaction - check if in snapshot's active set
                if snapshot.active_txns.contains(&self.xmin) {
                    return false; // Not yet committed when snapshot was taken
                }
                return false; // Still not committed
            }
            Some(TxnStatus::Aborted) | None => {
                return false; // Created by aborted transaction
            }
        }

        // Check xmax visibility
        if self.xmax == 0 {
            return true; // Not deleted
        }

        if self.xmax == snapshot.txn_id {
            return false; // Deleted by current transaction
        }

        // xmax must NOT be committed before snapshot
        match txn_manager.get_status(self.xmax) {
            Some(TxnStatus::Committed(commit_ts)) => {
                if commit_ts < snapshot.start_ts {
                    return false; // Deleted before snapshot
                }
                true // Deleted after snapshot - still visible
            }
            Some(TxnStatus::Active) | Some(TxnStatus::Aborted) | None => {
                true // Deletion not committed - still visible
            }
        }
    }
}

/// Read snapshot for consistent point-in-time queries
#[derive(Debug, Clone)]
pub struct Snapshot {
    /// Transaction ID that owns this snapshot
    pub txn_id: TxnId,
    /// Snapshot timestamp (all versions created before this are potentially visible)
    pub start_ts: Timestamp,
    /// Set of transaction IDs that were active when snapshot was taken
    pub active_txns: HashSet<TxnId>,
    /// Minimum active transaction ID (for garbage collection)
    pub xmin: TxnId,
}

impl Snapshot {
    /// Create a new snapshot
    pub fn new(txn_id: TxnId, start_ts: Timestamp, active_txns: HashSet<TxnId>) -> Self {
        let xmin = active_txns.iter().copied().min().unwrap_or(txn_id);
        Self {
            txn_id,
            start_ts,
            active_txns,
            xmin,
        }
    }

    /// Check if a transaction's changes are visible
    pub fn is_txn_visible(&self, txn_id: TxnId, commit_ts: Option<Timestamp>) -> bool {
        if txn_id == self.txn_id {
            return true; // Own changes are visible
        }

        if self.active_txns.contains(&txn_id) {
            return false; // Was in-progress when snapshot was taken
        }

        match commit_ts {
            Some(ts) => ts < self.start_ts,
            None => false, // Not committed
        }
    }
}

/// Transaction manager for MVCC (snapshot-only, no WAL durability)
/// 
/// # Deprecation Notice
///
/// **This implementation is deprecated for production use.** It provides snapshot
/// isolation but does NOT include WAL integration for crash recovery.
///
/// ## Migration Guide
///
/// For production workloads requiring durability, use [`MvccTransactionManager`]
/// from `sochdb_storage::wal_integration` which includes:
/// - Write-ahead logging for crash recovery
/// - Serializable Snapshot Isolation (SSI)
/// - Group commit for high throughput
/// - Event-driven async architecture
///
/// ## When to Use This Implementation
///
/// - Unit testing without durability overhead
/// - Ephemeral in-memory operations
/// - Prototyping snapshot isolation logic
///
/// ## See Also
///
/// - [`crate::wal_integration::MvccTransactionManager`] - Production transaction manager
/// - [`crate::transaction::TransactionCoordinator`] - Unified transaction trait
/// 
/// [`MvccTransactionManager`]: crate::MvccTransactionManager
#[deprecated(
    since = "0.1.0",
    note = "Use MvccTransactionManager from wal_integration for production workloads with durability"
)]
pub struct TransactionManager {
    /// Next transaction ID
    next_txn_id: AtomicU64,
    /// Global timestamp counter
    timestamp: AtomicU64,
    /// Active transactions: txn_id -> (start_ts, status)
    active_txns: RwLock<HashMap<TxnId, (Timestamp, TxnStatus)>>,
    /// Committed transaction log: txn_id -> commit_ts
    commit_log: RwLock<BTreeMap<TxnId, Timestamp>>,
    /// Minimum active transaction ID (for GC)
    min_active_txn: AtomicU64,
}

#[allow(deprecated)]
impl TransactionManager {
    /// Create a new transaction manager
    pub fn new() -> Self {
        Self {
            next_txn_id: AtomicU64::new(1),
            timestamp: AtomicU64::new(1),
            active_txns: RwLock::new(HashMap::new()),
            commit_log: RwLock::new(BTreeMap::new()),
            min_active_txn: AtomicU64::new(u64::MAX),
        }
    }

    /// Begin a new transaction
    pub fn begin(&self) -> (TxnId, Timestamp) {
        let txn_id = self.next_txn_id.fetch_add(1, Ordering::SeqCst);
        let start_ts = self.timestamp.fetch_add(1, Ordering::SeqCst);

        {
            let mut active = self.active_txns.write();
            active.insert(txn_id, (start_ts, TxnStatus::Active));
        }

        // Update min active
        self.update_min_active();

        (txn_id, start_ts)
    }

    /// Acquire a read snapshot
    pub fn acquire_snapshot(&self, txn_id: TxnId) -> Snapshot {
        let active = self.active_txns.read();

        let start_ts = active
            .get(&txn_id)
            .map(|(ts, _)| *ts)
            .unwrap_or_else(|| self.timestamp.load(Ordering::SeqCst));

        let active_set: HashSet<TxnId> = active
            .iter()
            .filter(|(id, (_, status))| **id != txn_id && *status == TxnStatus::Active)
            .map(|(id, _)| *id)
            .collect();

        Snapshot::new(txn_id, start_ts, active_set)
    }

    /// Commit a transaction
    pub fn commit(&self, txn_id: TxnId) -> Option<Timestamp> {
        let commit_ts = self.timestamp.fetch_add(1, Ordering::SeqCst);

        {
            let mut active = self.active_txns.write();
            if let Some((_, status)) = active.get_mut(&txn_id) {
                if *status != TxnStatus::Active {
                    return None; // Already committed or aborted
                }
                *status = TxnStatus::Committed(commit_ts);
            } else {
                return None; // Unknown transaction
            }
        }

        {
            let mut log = self.commit_log.write();
            log.insert(txn_id, commit_ts);
        }

        // Update min active
        self.update_min_active();

        Some(commit_ts)
    }

    /// Abort a transaction
    pub fn abort(&self, txn_id: TxnId) -> bool {
        let mut active = self.active_txns.write();

        if let Some((_, status)) = active.get_mut(&txn_id) {
            if *status != TxnStatus::Active {
                return false;
            }
            *status = TxnStatus::Aborted;
            self.update_min_active();
            true
        } else {
            false
        }
    }

    /// Get transaction status
    pub fn get_status(&self, txn_id: TxnId) -> Option<TxnStatus> {
        let active = self.active_txns.read();
        active.get(&txn_id).map(|(_, status)| *status)
    }

    /// Get commit timestamp for a transaction
    pub fn get_commit_ts(&self, txn_id: TxnId) -> Option<Timestamp> {
        let log = self.commit_log.read();
        log.get(&txn_id).copied()
    }

    /// Get minimum active transaction ID (for garbage collection)
    pub fn min_active_txn_id(&self) -> TxnId {
        self.min_active_txn.load(Ordering::SeqCst)
    }

    /// Get current timestamp
    pub fn current_timestamp(&self) -> Timestamp {
        self.timestamp.load(Ordering::SeqCst)
    }

    /// Update minimum active transaction ID
    fn update_min_active(&self) {
        let active = self.active_txns.read();
        let min = active
            .iter()
            .filter(|(_, (_, status))| *status == TxnStatus::Active)
            .map(|(&id, _)| id)
            .min()
            .unwrap_or(u64::MAX);
        self.min_active_txn.store(min, Ordering::SeqCst);
    }

    /// Garbage collect old transaction records
    ///
    /// Removes committed transactions older than the watermark.
    pub fn gc(&self, watermark: Timestamp) -> usize {
        let mut log = self.commit_log.write();
        let mut active = self.active_txns.write();

        let old_len = log.len();

        // Remove old committed entries
        log.retain(|_, commit_ts| *commit_ts >= watermark);

        // Remove old active entries that are committed
        active.retain(|txn_id, (_, status)| match status {
            TxnStatus::Committed(ts) => *ts >= watermark,
            TxnStatus::Aborted => {
                // Keep aborted if there might be references
                log.get(txn_id).map(|t| *t >= watermark).unwrap_or(true)
            }
            TxnStatus::Active => true,
        });

        old_len - log.len()
    }
}

#[allow(deprecated)]
impl Default for TransactionManager {
    fn default() -> Self {
        Self::new()
    }
}

/// MVCC version chain for a single key
#[derive(Debug)]
pub struct VersionChain<V> {
    /// Versions ordered by creation timestamp (newest first)
    versions: Vec<(VersionInfo, V)>,
}

impl<V: Clone> VersionChain<V> {
    /// Create empty version chain
    pub fn new() -> Self {
        Self {
            versions: Vec::new(),
        }
    }

    /// Add a new version
    pub fn add_version(&mut self, info: VersionInfo, value: V) {
        // Insert at front (newest first)
        self.versions.insert(0, (info, value));
    }

    /// Get visible version for a snapshot
    #[allow(deprecated)]
    pub fn get_visible(&self, snapshot: &Snapshot, txn_manager: &TransactionManager) -> Option<&V> {
        for (info, value) in &self.versions {
            if info.is_visible(snapshot, txn_manager) {
                return Some(value);
            }
        }
        None
    }

    /// Mark latest version as deleted
    pub fn delete(&mut self, xmax: TxnId, deleted_ts: Timestamp) -> bool {
        if let Some((info, _)) = self.versions.first_mut()
            && info.xmax == 0
        {
            info.delete(xmax, deleted_ts);
            return true;
        }
        false
    }

    /// Garbage collect old versions
    pub fn gc(&mut self, min_visible_ts: Timestamp) -> usize {
        let old_len = self.versions.len();

        // Keep at least one version, and all versions visible to any active snapshot
        if self.versions.len() <= 1 {
            return 0;
        }

        self.versions
            .retain(|(info, _)| info.deleted_ts >= min_visible_ts);

        // Always keep at least one version
        if self.versions.is_empty() {
            return old_len; // All removed (shouldn't happen normally)
        }

        old_len - self.versions.len()
    }

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

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

/// MVCC-aware key-value store
#[allow(deprecated)]
pub struct MvccStore<V> {
    /// Version chains by key
    data: RwLock<HashMap<Vec<u8>, VersionChain<V>>>,
    /// Transaction manager
    txn_manager: Arc<TransactionManager>,
}

#[allow(deprecated)]
impl<V: Clone + Send + Sync> MvccStore<V> {
    /// Create a new MVCC store
    pub fn new(txn_manager: Arc<TransactionManager>) -> Self {
        Self {
            data: RwLock::new(HashMap::new()),
            txn_manager,
        }
    }

    /// Put a value (creates new version)
    pub fn put(&self, key: &[u8], value: V, txn_id: TxnId) -> Timestamp {
        let created_ts = self.txn_manager.current_timestamp();
        let info = VersionInfo::new(txn_id, created_ts);

        let mut data = self.data.write();
        let chain = data.entry(key.to_vec()).or_default();
        chain.add_version(info, value);

        created_ts
    }

    /// Get visible value for a snapshot
    pub fn get(&self, key: &[u8], snapshot: &Snapshot) -> Option<V> {
        let data = self.data.read();
        data.get(key)
            .and_then(|chain| chain.get_visible(snapshot, &self.txn_manager))
            .cloned()
    }

    /// Delete a key (marks version as deleted)
    pub fn delete(&self, key: &[u8], txn_id: TxnId) -> bool {
        let deleted_ts = self.txn_manager.current_timestamp();
        let mut data = self.data.write();

        if let Some(chain) = data.get_mut(key) {
            chain.delete(txn_id, deleted_ts)
        } else {
            false
        }
    }

    /// Garbage collect old versions
    pub fn gc(&self) -> usize {
        let min_visible = self.txn_manager.min_active_txn_id();
        let min_ts = self
            .txn_manager
            .get_commit_ts(min_visible)
            .unwrap_or(self.txn_manager.current_timestamp());

        let mut data = self.data.write();
        let mut total_gc = 0;

        for chain in data.values_mut() {
            total_gc += chain.gc(min_ts);
        }

        total_gc
    }
}

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

    #[test]
    fn test_basic_visibility() {
        let manager = TransactionManager::new();

        // Transaction 1 creates version
        let (txn1, _) = manager.begin();
        let _snapshot1 = manager.acquire_snapshot(txn1);
        manager.commit(txn1);

        // Transaction 2 reads
        let (txn2, _) = manager.begin();
        let snapshot2 = manager.acquire_snapshot(txn2);

        // Transaction 1's changes should be visible to transaction 2
        assert!(snapshot2.is_txn_visible(txn1, manager.get_commit_ts(txn1)));

        manager.commit(txn2);
    }

    #[test]
    fn test_snapshot_isolation() {
        let manager = Arc::new(TransactionManager::new());
        let store = MvccStore::new(manager.clone());

        // Transaction 1 writes value
        let (txn1, _) = manager.begin();
        store.put(b"key1", "value1".to_string(), txn1);
        manager.commit(txn1);

        // Transaction 2 takes snapshot
        let (txn2, _) = manager.begin();
        let snapshot2 = manager.acquire_snapshot(txn2);

        // Transaction 3 updates value (after snapshot2)
        let (txn3, _) = manager.begin();
        store.put(b"key1", "value2".to_string(), txn3);
        manager.commit(txn3);

        // Snapshot2 should still see "value1"
        let value = store.get(b"key1", &snapshot2);
        assert_eq!(value, Some("value1".to_string()));

        manager.commit(txn2);
    }

    #[test]
    fn test_version_chain() {
        let manager = Arc::new(TransactionManager::new());

        let mut chain: VersionChain<String> = VersionChain::new();

        // Add first version
        let (txn1, _) = manager.begin();
        let info1 = VersionInfo::new(txn1, manager.current_timestamp());
        chain.add_version(info1, "v1".to_string());
        manager.commit(txn1);

        // Add second version
        let (txn2, _) = manager.begin();
        let info2 = VersionInfo::new(txn2, manager.current_timestamp());
        chain.add_version(info2, "v2".to_string());
        manager.commit(txn2);

        assert_eq!(chain.version_count(), 2);

        // Latest snapshot should see v2
        let (txn3, _) = manager.begin();
        let snapshot = manager.acquire_snapshot(txn3);
        assert_eq!(
            chain.get_visible(&snapshot, &manager),
            Some(&"v2".to_string())
        );
    }

    #[test]
    #[ignore] // Slow test - run locally with: cargo test -- --ignored
    fn test_abort_not_visible() {
        let manager = Arc::new(TransactionManager::new());
        let store = MvccStore::new(manager.clone());

        // Transaction 1 writes and aborts
        let (txn1, _) = manager.begin();
        store.put(b"key1", "value1".to_string(), txn1);
        manager.abort(txn1);

        // Transaction 2 should not see the aborted write
        let (txn2, _) = manager.begin();
        let snapshot2 = manager.acquire_snapshot(txn2);

        let value = store.get(b"key1", &snapshot2);
        assert_eq!(value, None);
    }
}

// =============================================================================
// Lock-Free MVCC (Task 2 Enhancement)
// =============================================================================

/// Lock-free transaction entry using epoch-based reclamation
struct EpochTxnEntry {
    txn_id: TxnId,
    start_ts: Timestamp,
    status: AtomicU64, // 0 = Active, 1+ = Committed(ts), u64::MAX = Aborted
}

impl EpochTxnEntry {
    fn new(txn_id: TxnId, start_ts: Timestamp) -> Self {
        Self {
            txn_id,
            start_ts,
            status: AtomicU64::new(0), // Active
        }
    }

    fn get_status(&self) -> TxnStatus {
        let val = self.status.load(Ordering::Acquire);
        if val == 0 {
            TxnStatus::Active
        } else if val == u64::MAX {
            TxnStatus::Aborted
        } else {
            TxnStatus::Committed(val)
        }
    }

    fn try_commit(&self, commit_ts: Timestamp) -> bool {
        self.status
            .compare_exchange(0, commit_ts, Ordering::AcqRel, Ordering::Acquire)
            .is_ok()
    }

    fn try_abort(&self) -> bool {
        self.status
            .compare_exchange(0, u64::MAX, Ordering::AcqRel, Ordering::Acquire)
            .is_ok()
    }
}

/// Epoch node for lock-free linked list
struct EpochNode {
    entry: EpochTxnEntry,
    next: Atomic<EpochNode>,
}

/// Lock-free MVCC manager using epoch-based reclamation
///
/// ## Performance Characteristics
/// - Wait-free visibility checks (single atomic load)
/// - Lock-free transaction begin/commit/abort
/// - Epoch-based garbage collection (no synchronization for reads)
///
/// ## Memory Safety
/// Uses crossbeam-epoch for safe concurrent memory reclamation:
/// - Readers pin the current epoch
/// - Writers defer destruction to future epochs
/// - GC runs incrementally without blocking readers
pub struct LockFreeMvccManager {
    /// Next transaction ID
    next_txn_id: AtomicU64,
    /// Global timestamp counter
    timestamp: AtomicU64,
    /// Head of active transaction list
    active_head: Atomic<EpochNode>,
    /// Committed transactions for visibility (sorted by commit_ts)
    committed: crossbeam_skiplist::SkipMap<TxnId, Timestamp>,
    /// Minimum visible timestamp (for GC)
    min_visible_ts: AtomicU64,
    /// Number of active transactions
    active_count: AtomicU64,
}

impl LockFreeMvccManager {
    /// Create a new lock-free MVCC manager
    pub fn new() -> Self {
        Self {
            next_txn_id: AtomicU64::new(1),
            timestamp: AtomicU64::new(1),
            active_head: Atomic::null(),
            committed: crossbeam_skiplist::SkipMap::new(),
            min_visible_ts: AtomicU64::new(0),
            active_count: AtomicU64::new(0),
        }
    }

    /// Begin a new transaction (lock-free)
    pub fn begin(&self) -> (TxnId, Timestamp) {
        let txn_id = self.next_txn_id.fetch_add(1, Ordering::SeqCst);
        let start_ts = self.timestamp.fetch_add(1, Ordering::SeqCst);

        let guard = epoch::pin();

        // Create entry and insert at head (lock-free CAS loop)
        let mut new_node = Owned::new(EpochNode {
            entry: EpochTxnEntry::new(txn_id, start_ts),
            next: Atomic::null(),
        });

        loop {
            let head = self.active_head.load(Ordering::Acquire, &guard);
            new_node.next.store(head, Ordering::Release);

            match self.active_head.compare_exchange(
                head,
                new_node,
                Ordering::AcqRel,
                Ordering::Acquire,
                &guard,
            ) {
                Ok(_) => {
                    self.active_count.fetch_add(1, Ordering::Relaxed);
                    break;
                }
                Err(e) => {
                    // CAS failed, get ownership back and retry
                    new_node = e.new;
                }
            }
        }

        (txn_id, start_ts)
    }

    /// Commit a transaction (lock-free CAS)
    pub fn commit(&self, txn_id: TxnId) -> Option<Timestamp> {
        let commit_ts = self.timestamp.fetch_add(1, Ordering::SeqCst);

        // Find the transaction entry
        let guard = epoch::pin();
        let mut current = self.active_head.load(Ordering::Acquire, &guard);

        while let Some(node) = unsafe { current.as_ref() } {
            if node.entry.txn_id == txn_id {
                if node.entry.try_commit(commit_ts) {
                    // Record in committed set
                    self.committed.insert(txn_id, commit_ts);
                    self.active_count.fetch_sub(1, Ordering::Relaxed);
                    return Some(commit_ts);
                } else {
                    return None; // Already committed or aborted
                }
            }
            current = node.next.load(Ordering::Acquire, &guard);
        }

        None // Transaction not found
    }

    /// Abort a transaction (lock-free)
    pub fn abort(&self, txn_id: TxnId) -> bool {
        let guard = epoch::pin();
        let mut current = self.active_head.load(Ordering::Acquire, &guard);

        while let Some(node) = unsafe { current.as_ref() } {
            if node.entry.txn_id == txn_id {
                let success = node.entry.try_abort();
                if success {
                    self.active_count.fetch_sub(1, Ordering::Relaxed);
                }
                return success;
            }
            current = node.next.load(Ordering::Acquire, &guard);
        }

        false
    }

    /// Get transaction status (wait-free read)
    pub fn get_status(&self, txn_id: TxnId) -> Option<TxnStatus> {
        // First check committed set (fast path)
        if let Some(entry) = self.committed.get(&txn_id) {
            return Some(TxnStatus::Committed(*entry.value()));
        }

        // Search active list
        let guard = epoch::pin();
        let mut current = self.active_head.load(Ordering::Acquire, &guard);

        while let Some(node) = unsafe { current.as_ref() } {
            if node.entry.txn_id == txn_id {
                return Some(node.entry.get_status());
            }
            current = node.next.load(Ordering::Acquire, &guard);
        }

        None
    }

    /// Acquire a snapshot (wait-free for visibility checks)
    pub fn acquire_snapshot(&self, txn_id: TxnId) -> Snapshot {
        let guard = epoch::pin();

        // Get start timestamp
        let start_ts = {
            let mut ts = self.timestamp.load(Ordering::SeqCst);
            let mut current = self.active_head.load(Ordering::Acquire, &guard);

            while let Some(node) = unsafe { current.as_ref() } {
                if node.entry.txn_id == txn_id {
                    ts = node.entry.start_ts;
                    break;
                }
                current = node.next.load(Ordering::Acquire, &guard);
            }
            ts
        };

        // Collect active transactions
        let mut active_set = HashSet::new();
        let mut current = self.active_head.load(Ordering::Acquire, &guard);

        while let Some(node) = unsafe { current.as_ref() } {
            if node.entry.txn_id != txn_id && matches!(node.entry.get_status(), TxnStatus::Active) {
                active_set.insert(node.entry.txn_id);
            }
            current = node.next.load(Ordering::Acquire, &guard);
        }

        Snapshot::new(txn_id, start_ts, active_set)
    }

    /// Get current timestamp
    pub fn current_timestamp(&self) -> Timestamp {
        self.timestamp.load(Ordering::SeqCst)
    }

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

    /// Epoch-based garbage collection
    ///
    /// Removes transaction entries that are:
    /// 1. Committed before the watermark
    /// 2. No longer needed for any active snapshot
    pub fn gc(&self, watermark: Timestamp) -> usize {
        self.min_visible_ts.store(watermark, Ordering::Release);

        // Remove old committed entries
        let mut removed = 0;
        let entries_to_remove: Vec<_> = self
            .committed
            .iter()
            .filter(|entry| *entry.value() < watermark)
            .map(|entry| *entry.key())
            .collect();

        for txn_id in entries_to_remove {
            if self.committed.remove(&txn_id).is_some() {
                removed += 1;
            }
        }

        // Clean up inactive nodes from the linked list
        // This is done by epoch-based deferred destruction
        let guard = epoch::pin();
        let _prev: Option<&EpochNode> = None;
        let mut current = self.active_head.load(Ordering::Acquire, &guard);

        while let Some(node) = unsafe { current.as_ref() } {
            let status = node.entry.get_status();
            match status {
                TxnStatus::Committed(ts) if ts < watermark => {
                    // Node can be unlinked - but for simplicity, just mark and skip
                    // Full list cleanup would require double-CAS or hazard pointers
                }
                TxnStatus::Aborted => {
                    // Similar treatment for aborted
                }
                _ => {}
            }
            current = node.next.load(Ordering::Acquire, &guard);
        }

        // Advance the epoch to allow memory reclamation
        drop(guard);
        epoch::pin().flush();

        removed
    }
}

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

#[cfg(test)]
mod lock_free_tests {
    use super::*;
    use std::thread;

    #[test]
    fn test_lock_free_basic() {
        let manager = LockFreeMvccManager::new();

        let (txn1, ts1) = manager.begin();
        assert!(ts1 > 0);
        assert_eq!(manager.get_status(txn1), Some(TxnStatus::Active));

        let commit_ts = manager.commit(txn1).unwrap();
        assert!(commit_ts > ts1);
        assert_eq!(
            manager.get_status(txn1),
            Some(TxnStatus::Committed(commit_ts))
        );
    }

    #[test]
    fn test_lock_free_abort() {
        let manager = LockFreeMvccManager::new();

        let (txn1, _) = manager.begin();
        assert!(manager.abort(txn1));
        assert_eq!(manager.get_status(txn1), Some(TxnStatus::Aborted));

        // Cannot commit after abort
        assert!(manager.commit(txn1).is_none());
    }

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

        let manager = Arc::new(LockFreeMvccManager::new());
        let num_threads = 8;
        let txns_per_thread = 100;

        let handles: Vec<_> = (0..num_threads)
            .map(|_| {
                let m = Arc::clone(&manager);
                thread::spawn(move || {
                    for _ in 0..txns_per_thread {
                        let (txn_id, _) = m.begin();
                        m.commit(txn_id);
                    }
                })
            })
            .collect();

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

        // All transactions should be committed
        let total = num_threads * txns_per_thread;
        assert!(manager.committed.len() >= total as usize);
    }

    #[test]
    fn test_lock_free_snapshot() {
        let manager = LockFreeMvccManager::new();

        let (txn1, _) = manager.begin();
        manager.commit(txn1);

        let (txn2, _) = manager.begin();
        let snapshot = manager.acquire_snapshot(txn2);

        // txn1 should be visible, txn2 should not be in active set
        assert!(!snapshot.active_txns.contains(&txn1));
        assert!(!snapshot.active_txns.contains(&txn2));
    }
}