featherdb_mvcc/

transaction.rs

//! Transaction management

use crate::{Snapshot, VersionStore};
use featherdb_core::{Error, Lsn, PageId, Result, TransactionConfig, TransactionId};
use featherdb_storage::Wal;
use parking_lot::{Mutex, RwLock};
use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::{Duration, Instant};

/// Transaction mode
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TransactionMode {
    /// Read-only transaction
    ReadOnly,
    /// Read-write transaction
    ReadWrite,
}

/// Transaction status
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TransactionStatus {
    /// Transaction is active
    Active,
    /// Transaction is committing
    Committing,
    /// Transaction has committed
    Committed,
    /// Transaction has been aborted
    Aborted,
}

/// Information about a transaction
#[derive(Debug, Clone)]
pub struct TransactionInfo {
    /// Transaction ID
    pub id: u64,
    /// Age of the transaction (time since it started)
    pub age: Duration,
}

/// Status of garbage collection
#[derive(Debug, Clone)]
pub struct GcStatus {
    /// ID of the oldest active transaction (blocks GC)
    pub oldest_active_txn: Option<u64>,
    /// Whether GC is currently blocked
    pub blocked: bool,
    /// Reason GC is blocked (if blocked)
    pub blocked_reason: Option<String>,
}

/// Transaction metrics for monitoring
#[derive(Debug, Default)]
pub struct TransactionMetrics {
    /// Number of currently active transactions
    active_count: AtomicU64,
    /// Total transactions committed
    total_commits: AtomicU64,
    /// Total transactions rolled back
    total_rollbacks: AtomicU64,
    /// Number of long-running transactions (updated periodically)
    long_running_count: AtomicU64,
    /// Total duration of all completed transactions (microseconds)
    total_duration_us: AtomicU64,
    /// Total number of transactions (commits + rollbacks)
    total_transactions: AtomicU64,
}

impl TransactionMetrics {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn record_begin(&self) {
        self.active_count.fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_commit(&self, duration: Duration) {
        self.active_count.fetch_sub(1, Ordering::Relaxed);
        self.total_commits.fetch_add(1, Ordering::Relaxed);
        self.total_transactions.fetch_add(1, Ordering::Relaxed);
        self.total_duration_us
            .fetch_add(duration.as_micros() as u64, Ordering::Relaxed);
    }

    pub fn record_rollback(&self, duration: Duration) {
        self.active_count.fetch_sub(1, Ordering::Relaxed);
        self.total_rollbacks.fetch_add(1, Ordering::Relaxed);
        self.total_transactions.fetch_add(1, Ordering::Relaxed);
        self.total_duration_us
            .fetch_add(duration.as_micros() as u64, Ordering::Relaxed);
    }

    pub fn update_long_running_count(&self, count: u64) {
        self.long_running_count.store(count, Ordering::Relaxed);
    }

    /// Get a snapshot of metrics
    pub fn snapshot(&self) -> TransactionMetricsSnapshot {
        let active = self.active_count.load(Ordering::Relaxed);
        let commits = self.total_commits.load(Ordering::Relaxed);
        let rollbacks = self.total_rollbacks.load(Ordering::Relaxed);
        let long_running = self.long_running_count.load(Ordering::Relaxed);
        let total_duration = self.total_duration_us.load(Ordering::Relaxed);
        let total = self.total_transactions.load(Ordering::Relaxed);

        let avg_duration_us = if total > 0 { total_duration / total } else { 0 };

        TransactionMetricsSnapshot {
            active_count: active,
            total_commits: commits,
            total_rollbacks: rollbacks,
            long_running_count: long_running,
            avg_duration_us,
            total_transactions: total,
        }
    }
}

/// Snapshot of transaction metrics at a point in time
#[derive(Debug, Clone)]
pub struct TransactionMetricsSnapshot {
    /// Number of currently active transactions
    pub active_count: u64,
    /// Total transactions committed
    pub total_commits: u64,
    /// Total transactions rolled back
    pub total_rollbacks: u64,
    /// Number of long-running transactions
    pub long_running_count: u64,
    /// Average transaction duration (microseconds)
    pub avg_duration_us: u64,
    /// Total number of transactions
    pub total_transactions: u64,
}

impl TransactionMetricsSnapshot {
    /// Calculate commit rate (commits / total transactions)
    /// Returns 0.0 if no transactions have completed
    pub fn commit_rate(&self) -> f64 {
        if self.total_transactions == 0 {
            0.0
        } else {
            self.total_commits as f64 / self.total_transactions as f64
        }
    }
}

/// Internal transaction state
#[allow(dead_code)]
struct TransactionState {
    id: TransactionId,
    mode: TransactionMode,
    status: TransactionStatus,
    snapshot: Snapshot,
    start_time: Instant,
    last_lsn: Lsn,
    /// Write set: (table_name, serialized_pk) for conflict detection
    write_set: HashSet<(String, Vec<u8>)>,
}

/// Manages all transactions in the database
pub struct TransactionManager {
    /// Next transaction ID to assign
    next_txn_id: AtomicU64,
    /// Currently active transactions
    active_txns: RwLock<HashMap<TransactionId, TransactionState>>,
    /// Write lock for serializing writers
    write_lock: Mutex<()>,
    /// Oldest active transaction ID (for GC)
    oldest_active: AtomicU64,
    /// Version store for old versions
    version_store: RwLock<VersionStore>,
    /// Aborted transaction IDs - these transactions' writes should be invisible
    aborted_txns: RwLock<HashSet<TransactionId>>,
    /// Fast-path flag: true if any transaction has ever been aborted
    has_aborted: AtomicBool,
    /// Transaction metrics
    metrics: TransactionMetrics,
    /// Pages dirtied by each active transaction (for deferred flush on commit)
    dirty_pages: RwLock<HashMap<TransactionId, HashSet<PageId>>>,
    /// Write sets of recently committed transactions for conflict detection.
    /// Maps committed txn_id to its write set. Cleaned up when older than oldest active txn.
    #[allow(clippy::type_complexity)]
    committed_writes: RwLock<HashMap<TransactionId, HashSet<(String, Vec<u8>)>>>,
}

impl TransactionManager {
    /// Create a new transaction manager
    ///
    /// For new databases, use this constructor. For restoring from a persisted
    /// database, use `TransactionManager::with_start_txn_id()` instead.
    pub fn new() -> Self {
        TransactionManager {
            next_txn_id: AtomicU64::new(1),
            active_txns: RwLock::new(HashMap::new()),
            write_lock: Mutex::new(()),
            oldest_active: AtomicU64::new(u64::MAX),
            version_store: RwLock::new(VersionStore::new()),
            aborted_txns: RwLock::new(HashSet::new()),
            has_aborted: AtomicBool::new(false),
            metrics: TransactionMetrics::new(),
            dirty_pages: RwLock::new(HashMap::new()),
            committed_writes: RwLock::new(HashMap::new()),
        }
    }

    /// Create a new transaction manager with a specific starting transaction ID.
    ///
    /// Used when restoring from a persisted database to maintain MVCC visibility.
    /// The `start_txn_id` should be the value persisted from the previous session,
    /// ensuring transaction IDs remain monotonically increasing across restarts.
    ///
    /// For new databases, use `TransactionManager::new()` instead.
    pub fn with_start_txn_id(start_txn_id: u64) -> Self {
        TransactionManager {
            next_txn_id: AtomicU64::new(start_txn_id),
            active_txns: RwLock::new(HashMap::new()),
            write_lock: Mutex::new(()),
            oldest_active: AtomicU64::new(u64::MAX),
            version_store: RwLock::new(VersionStore::new()),
            aborted_txns: RwLock::new(HashSet::new()),
            has_aborted: AtomicBool::new(false),
            metrics: TransactionMetrics::new(),
            dirty_pages: RwLock::new(HashMap::new()),
            committed_writes: RwLock::new(HashMap::new()),
        }
    }

    /// Get the current next transaction ID (for persistence)
    ///
    /// This value should be persisted to ensure transaction IDs remain
    /// monotonically increasing across database restarts.
    pub fn next_txn_id(&self) -> u64 {
        self.next_txn_id.load(Ordering::SeqCst)
    }

    /// Begin a new transaction with default configuration
    pub fn begin(&self, mode: TransactionMode) -> Transaction<'_> {
        self.begin_with_config(mode, TransactionConfig::default())
    }

    /// Begin a new transaction with specific configuration
    pub fn begin_with_config(
        &self,
        mode: TransactionMode,
        config: TransactionConfig,
    ) -> Transaction<'_> {
        let txn_id = TransactionId(self.next_txn_id.fetch_add(1, Ordering::SeqCst));
        let created_at = Instant::now();

        // Create snapshot of current state
        let snapshot = {
            let active = self.active_txns.read();
            let in_progress: HashSet<_> = active.keys().copied().collect();
            let hwm = TransactionId(self.next_txn_id.load(Ordering::SeqCst));
            Snapshot::new(txn_id, hwm, in_progress)
        };

        let state = TransactionState {
            id: txn_id,
            mode,
            status: TransactionStatus::Active,
            snapshot: snapshot.clone(),
            start_time: created_at,
            last_lsn: Lsn::ZERO,
            write_set: HashSet::new(),
        };

        // Register transaction
        {
            let mut active = self.active_txns.write();
            active.insert(txn_id, state);

            // Update oldest active
            let oldest = active.keys().min().map(|t| t.0).unwrap_or(u64::MAX);
            self.oldest_active.store(oldest, Ordering::SeqCst);
        }

        // Record transaction begin in metrics
        self.metrics.record_begin();

        Transaction {
            id: txn_id,
            mode,
            snapshot,
            manager: self,
            created_at,
            config,
            warned: false,
        }
    }

    /// Begin a read-only transaction
    pub fn begin_read_only(&self) -> Transaction<'_> {
        self.begin(TransactionMode::ReadOnly)
    }

    /// Begin a read-write transaction
    pub fn begin_read_write(&self) -> Transaction<'_> {
        self.begin(TransactionMode::ReadWrite)
    }

    /// Commit a transaction (public API without WAL)
    pub fn commit_txn(&self, txn_id: TransactionId) -> Result<()> {
        self.commit(txn_id, None)
    }

    /// Abort a transaction (public API without WAL)
    pub fn abort_txn(&self, txn_id: TransactionId) -> Result<()> {
        self.abort(txn_id, None)
    }

    /// Record a write operation for conflict detection (first-committer-wins).
    ///
    /// Called by the DML executor for each row modified by INSERT, UPDATE, or DELETE.
    /// The (table_name, pk_bytes) pair is added to the transaction's write set,
    /// which is checked against committed write sets at commit time.
    /// Add an ancestor to a transaction's snapshot (for savepoint support).
    ///
    /// When a savepoint creates a sub-transaction, the sub-transaction needs to
    /// recognize the parent transaction as an ancestor so that write-write conflict
    /// detection doesn't flag writes from the same logical transaction.
    pub fn add_snapshot_ancestor(&self, txn_id: TransactionId, ancestor: TransactionId) {
        let mut active = self.active_txns.write();
        if let Some(state) = active.get_mut(&txn_id) {
            state.snapshot.add_ancestor(ancestor);
        }
    }

    /// Record a write and check for conflicts (first-committer-wins).
    ///
    /// Returns `Err(WriteConflict)` if another transaction has already written
    /// to the same (table, pk) — either committed after our snapshot or still active.
    pub fn record_write(
        &self,
        txn_id: TransactionId,
        table: String,
        pk_bytes: Vec<u8>,
    ) -> Result<()> {
        let entry = (table.clone(), pk_bytes);

        // Get our snapshot for visibility checks
        let snapshot = {
            let active = self.active_txns.read();
            match active.get(&txn_id) {
                Some(state) => state.snapshot.clone(),
                None => return Ok(()), // txn not found, nothing to do
            }
        };

        // Check committed_writes for conflicts (txns committed after our snapshot)
        {
            let committed = self.committed_writes.read();
            for (&committed_txn_id, committed_ws) in committed.iter() {
                if !snapshot.can_see(committed_txn_id) && committed_ws.contains(&entry) {
                    return Err(Error::WriteConflict { table });
                }
            }
        }

        // Check active transactions' write sets for conflicts (other in-progress txns).
        // Skip ancestor transactions (savepoint sub-txns in the same logical txn).
        {
            let active = self.active_txns.read();
            for (&other_txn_id, other_state) in active.iter() {
                if other_txn_id != txn_id
                    && !snapshot.is_ancestor(other_txn_id)
                    && other_state.write_set.contains(&entry)
                {
                    return Err(Error::WriteConflict { table });
                }
            }
        }

        // No conflict — record the write
        let mut active = self.active_txns.write();
        if let Some(state) = active.get_mut(&txn_id) {
            state.write_set.insert(entry);
        }

        Ok(())
    }

    /// Commit a transaction
    fn commit(&self, txn_id: TransactionId, wal: Option<&mut Wal>) -> Result<()> {
        let (duration, result) = {
            let mut active = self.active_txns.write();

            let state = active.get_mut(&txn_id).ok_or(Error::TransactionEnded)?;

            if state.status != TransactionStatus::Active {
                return Err(Error::TransactionEnded);
            }

            let start_time = state.start_time;

            // Write-write conflict detection (first-committer-wins).
            // Extract what we need before the conflict check to avoid borrow issues.
            let write_set = std::mem::take(&mut state.write_set);
            let snapshot = state.snapshot.clone();

            if !write_set.is_empty() {
                let committed = self.committed_writes.read();
                let mut conflict_table: Option<String> = None;

                'outer: for (&committed_txn_id, committed_ws) in committed.iter() {
                    // Only check transactions that are NOT visible to our snapshot
                    // (i.e., they committed after we started)
                    if !snapshot.can_see(committed_txn_id) {
                        for entry in &write_set {
                            if committed_ws.contains(entry) {
                                conflict_table = Some(entry.0.clone());
                                break 'outer;
                            }
                        }
                    }
                }
                drop(committed);

                if let Some(table) = conflict_table {
                    // Conflict detected — abort this transaction
                    if let Some(s) = active.get_mut(&txn_id) {
                        s.status = TransactionStatus::Aborted;
                    }
                    active.remove(&txn_id);
                    let oldest = active.keys().min().map(|t| t.0).unwrap_or(u64::MAX);
                    self.oldest_active.store(oldest, Ordering::SeqCst);
                    drop(active);
                    self.aborted_txns.write().insert(txn_id);
                    self.has_aborted.store(true, Ordering::Release);
                    self.metrics.record_rollback(start_time.elapsed());
                    return Err(Error::WriteConflict { table });
                }
            }

            state.status = TransactionStatus::Committing;

            // Log commit to WAL
            if let Some(wal) = wal {
                wal.log_commit(txn_id, state.last_lsn)?;
            }

            state.status = TransactionStatus::Committed;

            // Move write set to committed_writes for future conflict checks
            if !write_set.is_empty() {
                self.committed_writes.write().insert(txn_id, write_set);
            }

            // Remove from active transactions
            active.remove(&txn_id);

            // Update oldest active
            let oldest = active.keys().min().map(|t| t.0).unwrap_or(u64::MAX);
            self.oldest_active.store(oldest, Ordering::SeqCst);

            // GC committed_writes: only safe when no active transactions remain,
            // because an active txn may have started while a now-committed txn was
            // in-progress (making it invisible to the active txn's snapshot).
            let oldest_val = self.oldest_active.load(Ordering::SeqCst);
            if oldest_val == u64::MAX {
                // No active transactions — all committed writes can be cleaned up
                self.committed_writes.write().clear();
            }

            (start_time.elapsed(), Ok(()))
        };

        // Record commit in metrics
        self.metrics.record_commit(duration);

        result
    }

    /// Abort a transaction
    fn abort(&self, txn_id: TransactionId, wal: Option<&mut Wal>) -> Result<()> {
        let (duration, result) = {
            let mut active = self.active_txns.write();

            let state = active.get_mut(&txn_id).ok_or(Error::TransactionEnded)?;

            if state.status != TransactionStatus::Active {
                return Err(Error::TransactionEnded);
            }

            let start_time = state.start_time;

            // Log abort to WAL
            if let Some(wal) = wal {
                wal.log_abort(txn_id, state.last_lsn)?;
            }

            state.status = TransactionStatus::Aborted;

            // Remove from active transactions
            active.remove(&txn_id);

            // Add to aborted transactions set so visibility checks can exclude its writes
            self.aborted_txns.write().insert(txn_id);
            self.has_aborted.store(true, Ordering::Release);

            // Update oldest active
            let oldest = active.keys().min().map(|t| t.0).unwrap_or(u64::MAX);
            self.oldest_active.store(oldest, Ordering::SeqCst);

            (start_time.elapsed(), Ok(()))
        };

        // Record rollback in metrics
        self.metrics.record_rollback(duration);

        result
    }

    /// Acquire write lock for exclusive writing
    pub fn acquire_write_lock(&self) -> impl Drop + '_ {
        self.write_lock.lock()
    }

    /// Update last LSN for a transaction (used for WAL integration)
    #[allow(dead_code)]
    fn update_last_lsn(&self, txn_id: TransactionId, lsn: Lsn) {
        let mut active = self.active_txns.write();
        if let Some(state) = active.get_mut(&txn_id) {
            state.last_lsn = lsn;
        }
    }

    /// Get the oldest active transaction ID
    pub fn oldest_active_txn(&self) -> Option<TransactionId> {
        let oldest = self.oldest_active.load(Ordering::SeqCst);
        if oldest == u64::MAX {
            None
        } else {
            Some(TransactionId(oldest))
        }
    }

    /// Get the version store
    pub fn version_store(&self) -> &RwLock<VersionStore> {
        &self.version_store
    }

    /// Get number of active transactions
    pub fn active_count(&self) -> usize {
        self.active_txns.read().len()
    }

    /// Quick check whether any aborted transactions exist.
    /// When false, all `is_aborted` calls are guaranteed to return false.
    #[inline]
    pub fn has_aborted_transactions(&self) -> bool {
        self.has_aborted.load(Ordering::Acquire)
    }

    /// Check if a transaction was aborted
    #[inline]
    pub fn is_aborted(&self, txn_id: TransactionId) -> bool {
        // Fast path: if no transactions have ever been aborted, skip the lock
        if !self.has_aborted.load(Ordering::Acquire) {
            return false;
        }
        self.aborted_txns.read().contains(&txn_id)
    }

    /// Check if a row is visible to a snapshot, accounting for aborted transactions.
    ///
    /// A row is visible if:
    /// 1. The creating transaction was NOT aborted
    /// 2. The snapshot can see the creating transaction (per standard MVCC rules)
    /// 3. Either the row is not deleted, the deleting transaction was aborted,
    ///    or the snapshot cannot see the deletion
    #[inline]
    pub fn is_row_visible(
        &self,
        snapshot: &Snapshot,
        created_by: TransactionId,
        deleted_by: Option<TransactionId>,
    ) -> bool {
        // Check if the creating transaction was aborted
        // If it was aborted, the row should not be visible to anyone
        if self.is_aborted(created_by) {
            return false;
        }

        // If there's a deletion, check if the deleting transaction was aborted
        // If aborted, treat the row as not deleted
        let effective_deleted_by = match deleted_by {
            Some(del_by) if self.is_aborted(del_by) => None, // Deletion was aborted, ignore it
            other => other,
        };

        // Now check standard MVCC visibility with the effective deletion status
        snapshot.is_visible(created_by, effective_deleted_by)
    }

    /// Get the number of aborted transactions being tracked
    pub fn aborted_count(&self) -> usize {
        self.aborted_txns.read().len()
    }

    /// Clean up old aborted transaction IDs that are no longer needed.
    ///
    /// Aborted transaction IDs can be removed once all their written data
    /// has been garbage collected. This is typically safe to do when
    /// the aborted transaction ID is older than any active transaction's
    /// snapshot, since all such data would have been cleaned up by GC.
    pub fn cleanup_aborted_txns(&self, older_than: TransactionId) {
        let mut aborted = self.aborted_txns.write();
        aborted.retain(|&txn_id| txn_id >= older_than);
        if aborted.is_empty() {
            self.has_aborted.store(false, Ordering::Release);
        }
    }

    // ==================== Dirty Page Tracking ====================

    /// Register a page as dirtied by a transaction
    ///
    /// This is called when a transaction modifies a page. The page ID
    /// is tracked so that on commit, only this transaction's pages are
    /// flushed, and on abort, they can be discarded.
    pub fn register_dirty_page(&self, txn_id: TransactionId, page_id: PageId) {
        let mut dirty = self.dirty_pages.write();
        dirty.entry(txn_id).or_default().insert(page_id);
    }

    /// Get all pages dirtied by a transaction
    pub fn get_dirty_pages(&self, txn_id: TransactionId) -> HashSet<PageId> {
        self.dirty_pages
            .read()
            .get(&txn_id)
            .cloned()
            .unwrap_or_default()
    }

    /// Clear dirty page tracking for a transaction (called on commit/abort)
    pub fn clear_dirty_pages(&self, txn_id: TransactionId) {
        self.dirty_pages.write().remove(&txn_id);
    }

    /// Get the number of dirty pages for a transaction
    pub fn dirty_page_count(&self, txn_id: TransactionId) -> usize {
        self.dirty_pages
            .read()
            .get(&txn_id)
            .map(|s| s.len())
            .unwrap_or(0)
    }

    /// Run garbage collection on old versions.
    ///
    /// Removes old versions that no active transaction can see.
    /// Returns statistics about what was collected.
    pub fn run_gc(&self) -> crate::version::GcStats {
        let oldest = self.oldest_active_txn();
        let mut version_store = self.version_store.write();
        version_store.gc(oldest)
    }

    /// Get the number of old versions in the version store
    pub fn version_count(&self) -> usize {
        self.version_store.read().len()
    }

    /// Get information about long-running transactions
    ///
    /// Returns a list of transactions that have been running longer than the
    /// specified threshold, sorted by age (oldest first).
    pub fn long_running_transactions(&self, threshold: Duration) -> Vec<TransactionInfo> {
        let active = self.active_txns.read();
        let now = Instant::now();

        let mut results: Vec<TransactionInfo> = active
            .iter()
            .filter_map(|(id, state)| {
                let elapsed = now.duration_since(state.start_time);
                if elapsed >= threshold {
                    Some(TransactionInfo {
                        id: id.0,
                        age: elapsed,
                    })
                } else {
                    None
                }
            })
            .collect();

        // Sort by age, oldest first
        results.sort_by(|a, b| b.age.cmp(&a.age));
        results
    }

    /// Get current GC status
    ///
    /// Returns information about whether garbage collection is blocked
    /// by active transactions.
    pub fn gc_status(&self) -> GcStatus {
        let active = self.active_txns.read();
        let oldest = active.keys().min().copied();

        GcStatus {
            oldest_active_txn: oldest.map(|id| id.0),
            blocked: oldest.is_some(),
            blocked_reason: oldest
                .map(|id| format!("Waiting for transaction {} to complete", id.0)),
        }
    }

    /// Get transaction metrics snapshot
    ///
    /// Returns a snapshot of transaction metrics including active count,
    /// commits, rollbacks, and average transaction duration.
    pub fn metrics(&self) -> TransactionMetricsSnapshot {
        // Update long-running count before taking snapshot
        let long_running = self.long_running_transactions(Duration::from_secs(30));
        self.metrics
            .update_long_running_count(long_running.len() as u64);

        self.metrics.snapshot()
    }
}

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

/// A database transaction
pub struct Transaction<'a> {
    /// Transaction ID
    id: TransactionId,
    /// Transaction mode
    mode: TransactionMode,
    /// Snapshot for visibility
    snapshot: Snapshot,
    /// Reference to transaction manager
    manager: &'a TransactionManager,
    /// Time when transaction was created
    created_at: Instant,
    /// Transaction configuration (timeout, warning threshold)
    config: TransactionConfig,
    /// Track if we've already issued a warning
    warned: bool,
}

impl<'a> Transaction<'a> {
    /// Get the transaction ID
    pub fn id(&self) -> TransactionId {
        self.id
    }

    /// Get the transaction mode
    pub fn mode(&self) -> TransactionMode {
        self.mode
    }

    /// Get the snapshot
    pub fn snapshot(&self) -> &Snapshot {
        &self.snapshot
    }

    /// Check if this is a read-only transaction
    pub fn is_read_only(&self) -> bool {
        self.mode == TransactionMode::ReadOnly
    }

    /// Commit the transaction
    pub fn commit(self) -> Result<()> {
        self.check_timeout()?;
        self.manager.commit(self.id, None)
    }

    /// Commit with WAL
    pub fn commit_with_wal(self, wal: &mut Wal) -> Result<()> {
        self.check_timeout()?;
        self.manager.commit(self.id, Some(wal))
    }

    /// Abort the transaction
    pub fn abort(self) -> Result<()> {
        self.manager.abort(self.id, None)
    }

    /// Abort with WAL
    pub fn abort_with_wal(self, wal: &mut Wal) -> Result<()> {
        self.manager.abort(self.id, Some(wal))
    }

    /// Check if a version is visible
    pub fn can_see(&self, created_by: TransactionId) -> bool {
        self.snapshot.can_see(created_by)
    }

    /// Check write permission
    pub fn check_write(&self) -> Result<()> {
        self.check_timeout()?;
        if self.mode == TransactionMode::ReadOnly {
            return Err(Error::ReadOnly);
        }
        Ok(())
    }

    /// Check read permission (primarily for timeout enforcement)
    pub fn check_read(&self) -> Result<()> {
        self.check_timeout()
    }

    /// Check if this transaction has exceeded its timeout
    pub fn is_expired(&self) -> bool {
        if let Some(timeout_ms) = self.config.timeout_ms {
            self.created_at.elapsed().as_millis() as u64 > timeout_ms
        } else {
            false
        }
    }

    /// Get remaining time before timeout (None if no timeout configured or already expired)
    pub fn remaining_time(&self) -> Option<Duration> {
        let timeout_ms = self.config.timeout_ms?;
        let elapsed = self.created_at.elapsed();
        let timeout = Duration::from_millis(timeout_ms);
        if elapsed >= timeout {
            None
        } else {
            Some(timeout - elapsed)
        }
    }

    /// Get elapsed time since transaction started
    pub fn elapsed(&self) -> Duration {
        self.created_at.elapsed()
    }

    /// Check if warning threshold exceeded (call this periodically)
    /// Returns true only once when the threshold is first exceeded
    pub fn should_warn(&mut self) -> bool {
        if self.warned {
            return false;
        }
        if let Some(warn_ms) = self.config.warn_after_ms {
            if self.created_at.elapsed().as_millis() as u64 > warn_ms {
                self.warned = true;
                return true;
            }
        }
        false
    }

    /// Check if transaction is expired and return error if so
    fn check_timeout(&self) -> Result<()> {
        if self.is_expired() {
            return Err(Error::transaction_timeout(
                self.id.0,
                self.created_at.elapsed().as_millis() as u64,
                self.config.timeout_ms.unwrap_or(0),
            ));
        }
        Ok(())
    }
}

impl<'a> Drop for Transaction<'a> {
    fn drop(&mut self) {
        // Auto-abort if not committed
        let _ = self.manager.abort(self.id, None);
    }
}

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

    #[test]
    fn test_transaction_manager() {
        let tm = TransactionManager::new();

        let txn1 = tm.begin_read_write();
        assert_eq!(txn1.id(), TransactionId(1));

        let txn2 = tm.begin_read_only();
        assert_eq!(txn2.id(), TransactionId(2));

        assert_eq!(tm.active_count(), 2);

        txn1.commit().unwrap();
        assert_eq!(tm.active_count(), 1);

        txn2.abort().unwrap();
        assert_eq!(tm.active_count(), 0);
    }

    #[test]
    fn test_snapshot_isolation() {
        let tm = TransactionManager::new();

        let txn1 = tm.begin_read_write();
        let txn2 = tm.begin_read_only();

        // txn2 cannot see txn1's changes (txn1 started before txn2 but hasn't committed)
        assert!(!txn2.can_see(txn1.id()));

        // Each transaction can see its own changes
        assert!(txn1.can_see(txn1.id()));
        assert!(txn2.can_see(txn2.id()));
    }

    #[test]
    fn test_read_only_cannot_write() {
        let tm = TransactionManager::new();
        let txn = tm.begin_read_only();

        assert!(txn.check_write().is_err());
    }

    #[test]
    fn test_auto_abort_on_drop() {
        let tm = TransactionManager::new();

        {
            let _txn = tm.begin_read_write();
            assert_eq!(tm.active_count(), 1);
            // txn dropped here without commit
        }

        assert_eq!(tm.active_count(), 0);
    }

    #[test]
    fn test_run_gc() {
        use featherdb_core::Value;

        let tm = TransactionManager::new();

        // Add some old versions to the version store
        {
            let mut vs = tm.version_store().write();

            // Insert version that was deleted by an old transaction
            let old_version = crate::version::OldVersion {
                data: vec![Value::Integer(42)],
                created_by: TransactionId(1),
                deleted_by: Some(TransactionId(2)),
                prev_version: None,
            };
            vs.insert(old_version);

            // Insert version deleted by a future transaction
            let new_version = crate::version::OldVersion {
                data: vec![Value::Integer(100)],
                created_by: TransactionId(3),
                deleted_by: Some(TransactionId(100)),
                prev_version: None,
            };
            vs.insert(new_version);
        }

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

        // No active transactions, so GC should clean up all deleted versions
        let stats = tm.run_gc();

        assert_eq!(stats.versions_removed, 2);
        assert_eq!(tm.version_count(), 0);
    }

    #[test]
    fn test_gc_preserves_versions_for_active_transactions() {
        let tm = TransactionManager::new();

        // Add an old version deleted at transaction 5
        {
            let mut vs = tm.version_store().write();
            let old_version = crate::version::OldVersion {
                data: vec![featherdb_core::Value::Integer(42)],
                created_by: TransactionId(1),
                deleted_by: Some(TransactionId(5)),
                prev_version: None,
            };
            vs.insert(old_version);
        }

        // Start a transaction (will be txn id 1)
        let txn = tm.begin_read_write();

        // Advance transaction IDs by starting and committing several transactions
        for _ in 0..10 {
            let t = tm.begin_read_write();
            t.commit().unwrap();
        }

        // Now run GC - should NOT remove version because txn is still active
        // (txn was started before we simulate the deletion)
        // Actually, we need to think about this more carefully:
        // The oldest active is txn.id() = 1
        // The version was deleted_by = 5
        // So 5 >= 1, meaning NOT all active can see the deletion
        // Therefore GC should NOT collect this version

        let stats = tm.run_gc();

        // Version should NOT be removed because oldest active (txn id 1)
        // cannot see the deletion (deleted_by 5 >= oldest 1)
        assert_eq!(stats.versions_removed, 0);
        assert_eq!(tm.version_count(), 1);

        // Commit the transaction
        txn.commit().unwrap();

        // Now GC should clean it up (no active transactions)
        let stats = tm.run_gc();
        assert_eq!(stats.versions_removed, 1);
        assert_eq!(tm.version_count(), 0);
    }

    #[test]
    fn test_transaction_default_has_timeout() {
        let tm = TransactionManager::new();
        let txn = tm.begin_read_write();

        // Default config has 30 second timeout (safe default)
        assert!(!txn.is_expired());
        let remaining = txn.remaining_time();
        assert!(remaining.is_some());
        // Should be close to 30 seconds
        assert!(remaining.unwrap().as_secs() >= 29);
    }

    #[test]
    fn test_transaction_no_limits() {
        let tm = TransactionManager::new();
        let config = TransactionConfig::no_limits();
        let txn = tm.begin_with_config(TransactionMode::ReadWrite, config);

        // No timeout configured with no_limits()
        assert!(!txn.is_expired());
        assert_eq!(txn.remaining_time(), None);
    }

    #[test]
    fn test_transaction_with_timeout() {
        use std::thread;
        use std::time::Duration;

        let tm = TransactionManager::new();
        let config = TransactionConfig::new().with_timeout(100);
        let txn = tm.begin_with_config(TransactionMode::ReadWrite, config);

        // Should not be expired immediately
        assert!(!txn.is_expired());
        assert!(txn.remaining_time().is_some());

        // Wait for timeout to expire
        thread::sleep(Duration::from_millis(150));

        // Should be expired now
        assert!(txn.is_expired());
        assert_eq!(txn.remaining_time(), None);
    }

    #[test]
    fn test_transaction_elapsed_time() {
        use std::thread;
        use std::time::Duration;

        let tm = TransactionManager::new();
        let txn = tm.begin_read_write();

        // Elapsed time should be very small initially
        assert!(txn.elapsed().as_millis() < 10);

        thread::sleep(Duration::from_millis(50));

        // Should have elapsed at least 50ms
        assert!(txn.elapsed().as_millis() >= 50);
    }

    #[test]
    fn test_transaction_warning_threshold() {
        use std::thread;
        use std::time::Duration;

        let tm = TransactionManager::new();
        let config = TransactionConfig::new().with_warning(50);
        let mut txn = tm.begin_with_config(TransactionMode::ReadWrite, config);

        // Should not warn immediately
        assert!(!txn.should_warn());

        thread::sleep(Duration::from_millis(70));

        // Should warn after threshold
        assert!(txn.should_warn());

        // Should only warn once
        assert!(!txn.should_warn());
    }

    #[test]
    fn test_transaction_timeout_and_warning() {
        use std::thread;
        use std::time::Duration;

        let tm = TransactionManager::new();
        let config = TransactionConfig::with_timeout_and_warning(200, 50);
        let mut txn = tm.begin_with_config(TransactionMode::ReadWrite, config);

        // Initially neither expired nor warning
        assert!(!txn.is_expired());
        assert!(!txn.should_warn());

        thread::sleep(Duration::from_millis(70));

        // Should warn but not expired yet
        assert!(txn.should_warn());
        assert!(!txn.is_expired());
        assert!(txn.remaining_time().is_some());

        thread::sleep(Duration::from_millis(150));

        // Should be expired now
        assert!(txn.is_expired());
        assert_eq!(txn.remaining_time(), None);
    }

    #[test]
    fn test_transaction_remaining_time_decreases() {
        use std::thread;
        use std::time::Duration;

        let tm = TransactionManager::new();
        let config = TransactionConfig::new().with_timeout(500);
        let txn = tm.begin_with_config(TransactionMode::ReadWrite, config);

        let remaining1 = txn.remaining_time().unwrap();
        thread::sleep(Duration::from_millis(100));
        let remaining2 = txn.remaining_time().unwrap();

        // Remaining time should decrease
        assert!(remaining2 < remaining1);
        assert!(remaining2.as_millis() < remaining1.as_millis());
    }

    #[test]
    fn test_default_transaction_config() {
        let tm = TransactionManager::new();
        let config = TransactionConfig::default();
        let txn = tm.begin_with_config(TransactionMode::ReadWrite, config);

        // Default config has 30 second timeout (safe default)
        assert!(!txn.is_expired());
        let remaining = txn.remaining_time();
        assert!(remaining.is_some());
        assert!(remaining.unwrap().as_secs() >= 29);
    }

    #[test]
    fn test_timeout_prevents_commit() {
        use std::thread;
        use std::time::Duration;

        let tm = TransactionManager::new();
        let config = TransactionConfig::new().with_timeout(50);
        let txn = tm.begin_with_config(TransactionMode::ReadWrite, config);

        // Sleep past the timeout
        thread::sleep(Duration::from_millis(70));

        // Commit should fail with TransactionTimeout error
        let result = txn.commit();
        assert!(result.is_err());
        match result {
            Err(Error::TransactionTimeout {
                transaction_id,
                elapsed_ms,
                timeout_ms,
            }) => {
                assert_eq!(transaction_id, 1);
                assert!(elapsed_ms >= 70);
                assert_eq!(timeout_ms, 50);
            }
            _ => panic!("Expected TransactionTimeout error"),
        }
    }

    #[test]
    fn test_timeout_prevents_write_operations() {
        use std::thread;
        use std::time::Duration;

        let tm = TransactionManager::new();
        let config = TransactionConfig::new().with_timeout(50);
        let txn = tm.begin_with_config(TransactionMode::ReadWrite, config);

        // Sleep past the timeout
        thread::sleep(Duration::from_millis(70));

        // check_write should fail with TransactionTimeout error
        let result = txn.check_write();
        assert!(result.is_err());
        match result {
            Err(Error::TransactionTimeout {
                transaction_id,
                elapsed_ms,
                timeout_ms,
            }) => {
                assert_eq!(transaction_id, 1);
                assert!(elapsed_ms >= 70);
                assert_eq!(timeout_ms, 50);
            }
            _ => panic!("Expected TransactionTimeout error"),
        }
    }

    #[test]
    fn test_timeout_prevents_read_operations() {
        use std::thread;
        use std::time::Duration;

        let tm = TransactionManager::new();
        let config = TransactionConfig::new().with_timeout(50);
        let txn = tm.begin_with_config(TransactionMode::ReadOnly, config);

        // Sleep past the timeout
        thread::sleep(Duration::from_millis(70));

        // check_read should fail with TransactionTimeout error
        let result = txn.check_read();
        assert!(result.is_err());
        match result {
            Err(Error::TransactionTimeout {
                transaction_id,
                elapsed_ms,
                timeout_ms,
            }) => {
                assert_eq!(transaction_id, 1);
                assert!(elapsed_ms >= 70);
                assert_eq!(timeout_ms, 50);
            }
            _ => panic!("Expected TransactionTimeout error"),
        }
    }

    #[test]
    fn test_no_timeout_allows_operations() {
        use std::thread;
        use std::time::Duration;

        let tm = TransactionManager::new();
        let txn = tm.begin_read_write();

        // Sleep for a while
        thread::sleep(Duration::from_millis(50));

        // Operations should succeed without timeout configured
        assert!(txn.check_write().is_ok());
        assert!(txn.check_read().is_ok());
        assert!(txn.commit().is_ok());
    }

    #[test]
    fn test_operations_succeed_within_timeout() {
        use std::thread;
        use std::time::Duration;

        let tm = TransactionManager::new();
        let config = TransactionConfig::new().with_timeout(200);
        let txn = tm.begin_with_config(TransactionMode::ReadWrite, config);

        // Sleep but stay within timeout
        thread::sleep(Duration::from_millis(50));

        // Operations should succeed
        assert!(txn.check_write().is_ok());
        assert!(txn.check_read().is_ok());
        assert!(txn.commit().is_ok());
    }

    #[test]
    fn test_timeout_error_message() {
        use std::thread;
        use std::time::Duration;

        let tm = TransactionManager::new();
        let config = TransactionConfig::new().with_timeout(30);
        let txn = tm.begin_with_config(TransactionMode::ReadWrite, config);

        thread::sleep(Duration::from_millis(50));

        let result = txn.commit();
        assert!(result.is_err());

        let error_msg = result.unwrap_err().to_string();
        assert!(error_msg.contains("Transaction"));
        assert!(error_msg.contains("exceeded timeout"));
        assert!(error_msg.contains("30ms"));
    }

    #[test]
    fn test_long_running_transactions_empty() {
        use std::time::Duration;

        let tm = TransactionManager::new();

        // No transactions running
        let long_running = tm.long_running_transactions(Duration::from_secs(1));
        assert!(long_running.is_empty());
    }

    #[test]
    fn test_long_running_transactions_below_threshold() {
        use std::time::Duration;

        let tm = TransactionManager::new();
        let _txn1 = tm.begin_read_write();
        let _txn2 = tm.begin_read_only();

        // Transactions just started, should not appear as long-running
        let long_running = tm.long_running_transactions(Duration::from_secs(1));
        assert!(long_running.is_empty());
    }

    #[test]
    fn test_long_running_transactions_above_threshold() {
        use std::thread;
        use std::time::Duration;

        let tm = TransactionManager::new();
        let _txn1 = tm.begin_read_write();

        // Sleep past threshold
        thread::sleep(Duration::from_millis(100));

        let _txn2 = tm.begin_read_only();

        // Check with 50ms threshold
        let long_running = tm.long_running_transactions(Duration::from_millis(50));

        // Both transactions should appear (txn1 > 100ms, txn2 > 0ms but we check against 50ms)
        // Actually only txn1 should appear since txn2 was just created
        assert_eq!(long_running.len(), 1);
        assert_eq!(long_running[0].id, 1);
        assert!(long_running[0].age >= Duration::from_millis(100));
    }

    #[test]
    fn test_long_running_transactions_sorted_by_age() {
        use std::thread;
        use std::time::Duration;

        let tm = TransactionManager::new();
        let _txn1 = tm.begin_read_write();

        thread::sleep(Duration::from_millis(50));
        let _txn2 = tm.begin_read_only();

        thread::sleep(Duration::from_millis(50));
        let _txn3 = tm.begin_read_write();

        // Sleep a bit more to ensure all transactions are above threshold
        thread::sleep(Duration::from_millis(30));

        // All transactions should be above 20ms threshold
        let long_running = tm.long_running_transactions(Duration::from_millis(20));

        // Should be sorted by age, oldest first
        assert_eq!(long_running.len(), 3);
        assert_eq!(long_running[0].id, 1); // Oldest
        assert_eq!(long_running[1].id, 2); // Middle
        assert_eq!(long_running[2].id, 3); // Youngest

        // Verify ages are in descending order
        assert!(long_running[0].age >= long_running[1].age);
        assert!(long_running[1].age >= long_running[2].age);
    }

    #[test]
    fn test_long_running_transactions_after_commit() {
        use std::thread;
        use std::time::Duration;

        let tm = TransactionManager::new();
        let txn1 = tm.begin_read_write();

        thread::sleep(Duration::from_millis(50));

        // Transaction should appear as long-running
        let long_running = tm.long_running_transactions(Duration::from_millis(20));
        assert_eq!(long_running.len(), 1);
        assert_eq!(long_running[0].id, 1);

        // Commit the transaction
        txn1.commit().unwrap();

        // Should no longer appear
        let long_running = tm.long_running_transactions(Duration::from_millis(20));
        assert!(long_running.is_empty());
    }

    #[test]
    fn test_gc_status_no_active_transactions() {
        let tm = TransactionManager::new();

        let status = tm.gc_status();
        assert_eq!(status.oldest_active_txn, None);
        assert!(!status.blocked);
        assert_eq!(status.blocked_reason, None);
    }

    #[test]
    fn test_gc_status_with_active_transaction() {
        let tm = TransactionManager::new();
        let _txn = tm.begin_read_write();

        let status = tm.gc_status();
        assert_eq!(status.oldest_active_txn, Some(1));
        assert!(status.blocked);
        assert!(status.blocked_reason.is_some());
        assert!(status.blocked_reason.unwrap().contains("transaction 1"));
    }

    #[test]
    fn test_gc_status_multiple_transactions() {
        let tm = TransactionManager::new();
        let _txn1 = tm.begin_read_write();
        let _txn2 = tm.begin_read_only();
        let _txn3 = tm.begin_read_write();

        let status = tm.gc_status();
        // Oldest should be transaction 1
        assert_eq!(status.oldest_active_txn, Some(1));
        assert!(status.blocked);
        assert!(status.blocked_reason.unwrap().contains("transaction 1"));
    }

    #[test]
    fn test_gc_status_after_oldest_commits() {
        let tm = TransactionManager::new();
        let txn1 = tm.begin_read_write();
        let _txn2 = tm.begin_read_only();
        let _txn3 = tm.begin_read_write();

        // Initially blocked by transaction 1
        let status = tm.gc_status();
        assert_eq!(status.oldest_active_txn, Some(1));
        assert!(status.blocked);

        // Commit transaction 1
        txn1.commit().unwrap();

        // Now blocked by transaction 2
        let status = tm.gc_status();
        assert_eq!(status.oldest_active_txn, Some(2));
        assert!(status.blocked);
        assert!(status.blocked_reason.unwrap().contains("transaction 2"));
    }

    #[test]
    fn test_gc_status_after_all_commit() {
        let tm = TransactionManager::new();
        let txn1 = tm.begin_read_write();
        let txn2 = tm.begin_read_only();

        // Initially blocked
        let status = tm.gc_status();
        assert!(status.blocked);

        // Commit all
        txn1.commit().unwrap();
        txn2.commit().unwrap();

        // No longer blocked
        let status = tm.gc_status();
        assert_eq!(status.oldest_active_txn, None);
        assert!(!status.blocked);
        assert_eq!(status.blocked_reason, None);
    }

    #[test]
    fn test_long_running_and_gc_status_integration() {
        use std::thread;
        use std::time::Duration;

        let tm = TransactionManager::new();

        // Start a long-running transaction
        let _old_txn = tm.begin_read_write();

        thread::sleep(Duration::from_millis(100));

        // Start newer transactions
        let _new_txn1 = tm.begin_read_only();
        let _new_txn2 = tm.begin_read_write();

        // Check long-running transactions
        let long_running = tm.long_running_transactions(Duration::from_millis(50));
        assert_eq!(long_running.len(), 1);
        assert_eq!(long_running[0].id, 1);

        // Check GC status
        let gc_status = tm.gc_status();
        assert_eq!(gc_status.oldest_active_txn, Some(1));
        assert!(gc_status.blocked);
        assert!(gc_status.blocked_reason.unwrap().contains("transaction 1"));

        // The oldest transaction (1) is long-running and blocking GC
        assert_eq!(long_running[0].id, gc_status.oldest_active_txn.unwrap());
    }

    #[test]
    fn test_transaction_info_structure() {
        use std::thread;
        use std::time::Duration;

        let tm = TransactionManager::new();
        let _txn = tm.begin_read_write();

        thread::sleep(Duration::from_millis(50));

        let long_running = tm.long_running_transactions(Duration::from_millis(20));
        assert_eq!(long_running.len(), 1);

        let info = &long_running[0];
        assert_eq!(info.id, 1);
        assert!(info.age >= Duration::from_millis(50));
        assert!(info.age < Duration::from_secs(1));
    }

    #[test]
    fn test_gc_status_structure() {
        let tm = TransactionManager::new();
        let _txn = tm.begin_read_write();

        let status = tm.gc_status();
        assert_eq!(status.oldest_active_txn, Some(1));
        assert!(status.blocked);

        let reason = status.blocked_reason.unwrap();
        assert!(reason.contains("Waiting for"));
        assert!(reason.contains("transaction 1"));
        assert!(reason.contains("complete"));
    }

    #[test]
    fn test_transaction_metrics_initial_state() {
        let tm = TransactionManager::new();
        let metrics = tm.metrics();

        assert_eq!(metrics.active_count, 0);
        assert_eq!(metrics.total_commits, 0);
        assert_eq!(metrics.total_rollbacks, 0);
        assert_eq!(metrics.long_running_count, 0);
        assert_eq!(metrics.avg_duration_us, 0);
        assert_eq!(metrics.total_transactions, 0);
        assert_eq!(metrics.commit_rate(), 0.0);
    }

    #[test]
    fn test_transaction_metrics_begin() {
        let tm = TransactionManager::new();
        let _txn1 = tm.begin_read_write();
        let _txn2 = tm.begin_read_only();

        let metrics = tm.metrics();
        assert_eq!(metrics.active_count, 2);
        assert_eq!(metrics.total_commits, 0);
        assert_eq!(metrics.total_rollbacks, 0);
    }

    #[test]
    fn test_transaction_metrics_commit() {
        use std::thread;

        let tm = TransactionManager::new();
        let txn = tm.begin_read_write();

        // Sleep a bit to ensure measurable duration
        thread::sleep(Duration::from_millis(10));

        txn.commit().unwrap();

        let metrics = tm.metrics();
        assert_eq!(metrics.active_count, 0);
        assert_eq!(metrics.total_commits, 1);
        assert_eq!(metrics.total_rollbacks, 0);
        assert_eq!(metrics.total_transactions, 1);
        assert!(metrics.avg_duration_us > 0);
        assert_eq!(metrics.commit_rate(), 1.0);
    }

    #[test]
    fn test_transaction_metrics_rollback() {
        use std::thread;

        let tm = TransactionManager::new();
        let txn = tm.begin_read_write();

        thread::sleep(Duration::from_millis(10));

        txn.abort().unwrap();

        let metrics = tm.metrics();
        assert_eq!(metrics.active_count, 0);
        assert_eq!(metrics.total_commits, 0);
        assert_eq!(metrics.total_rollbacks, 1);
        assert_eq!(metrics.total_transactions, 1);
        assert!(metrics.avg_duration_us > 0);
        assert_eq!(metrics.commit_rate(), 0.0);
    }

    #[test]
    fn test_transaction_metrics_mixed() {
        use std::thread;

        let tm = TransactionManager::new();

        // Commit 3 transactions
        for _ in 0..3 {
            let txn = tm.begin_read_write();
            thread::sleep(Duration::from_millis(5));
            txn.commit().unwrap();
        }

        // Rollback 1 transaction
        let txn = tm.begin_read_write();
        thread::sleep(Duration::from_millis(5));
        txn.abort().unwrap();

        let metrics = tm.metrics();
        assert_eq!(metrics.active_count, 0);
        assert_eq!(metrics.total_commits, 3);
        assert_eq!(metrics.total_rollbacks, 1);
        assert_eq!(metrics.total_transactions, 4);
        assert!(metrics.avg_duration_us > 0);
        assert_eq!(metrics.commit_rate(), 0.75); // 3/4
    }

    #[test]
    fn test_transaction_metrics_long_running() {
        use std::thread;

        let tm = TransactionManager::new();
        let _txn1 = tm.begin_read_write();

        // Sleep past the 30s threshold used in metrics() method
        // But we'll use a shorter sleep and just check the structure
        thread::sleep(Duration::from_millis(10));

        let _txn2 = tm.begin_read_only();

        let metrics = tm.metrics();
        assert_eq!(metrics.active_count, 2);
        // long_running_count depends on the 30s threshold in metrics()
        // so it will be 0 in this quick test
        assert_eq!(metrics.long_running_count, 0);
    }

    #[test]
    fn test_transaction_metrics_snapshot_commit_rate() {
        let snapshot = TransactionMetricsSnapshot {
            active_count: 2,
            total_commits: 7,
            total_rollbacks: 3,
            long_running_count: 1,
            avg_duration_us: 1000,
            total_transactions: 10,
        };

        assert_eq!(snapshot.commit_rate(), 0.7);
    }

    #[test]
    fn test_transaction_metrics_snapshot_commit_rate_zero_transactions() {
        let snapshot = TransactionMetricsSnapshot {
            active_count: 0,
            total_commits: 0,
            total_rollbacks: 0,
            long_running_count: 0,
            avg_duration_us: 0,
            total_transactions: 0,
        };

        assert_eq!(snapshot.commit_rate(), 0.0);
    }

    #[test]
    fn test_transaction_metrics_average_duration() {
        use std::thread;

        let tm = TransactionManager::new();

        // Create several transactions with different durations
        let txn1 = tm.begin_read_write();
        thread::sleep(Duration::from_millis(10));
        txn1.commit().unwrap();

        let txn2 = tm.begin_read_write();
        thread::sleep(Duration::from_millis(20));
        txn2.commit().unwrap();

        let metrics = tm.metrics();
        assert_eq!(metrics.total_transactions, 2);
        // Average should be at least 10ms (in microseconds)
        // No upper bound — CI runners have unpredictable scheduling latency
        assert!(metrics.avg_duration_us >= 10_000);
    }

    #[test]
    fn test_transaction_metrics_auto_abort() {
        let tm = TransactionManager::new();

        {
            let _txn = tm.begin_read_write();
            // txn dropped here without commit - should auto-abort
        }

        let metrics = tm.metrics();
        assert_eq!(metrics.active_count, 0);
        assert_eq!(metrics.total_commits, 0);
        assert_eq!(metrics.total_rollbacks, 1);
        assert_eq!(metrics.total_transactions, 1);
    }
}