reddb_server/storage/wal/

transaction.rs

//! Transaction Manager
//!
//! Provides ACID transaction support for the RedDB storage engine.
//!
//! # Transaction Lifecycle
//!
//! 1. Begin: Allocate transaction ID, write Begin record to WAL
//! 2. Read/Write: Track page reads and buffer page writes
//! 3. Commit: Write Commit record to WAL, sync WAL
//! 4. Rollback: Write Rollback record to WAL, discard buffered writes
//!
//! # Isolation Level
//!
//! Currently implements Read Committed isolation:
//! - Reads see committed data at the start of the statement
//! - No dirty reads
//! - Possible non-repeatable reads
//!
//! # References
//!
//! - Turso `core/transaction.rs` - Transaction implementation
//! - SQLite transaction documentation

use std::collections::HashMap;
use std::io;
use std::path::{Path, PathBuf};
use std::sync::{Arc, RwLock, RwLockReadGuard, RwLockWriteGuard};

use parking_lot::{Mutex, MutexGuard};

use super::append_coordinator::WalAppendCoordinator;
use super::record::WalRecord;
use super::writer::WalWriter;
use crate::storage::engine::{Page, Pager, PAGE_SIZE};
use crate::storage::transaction::snapshot::SnapshotManager;

/// Transaction state
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TxState {
    /// Transaction is active and can perform operations
    Active,
    /// Transaction has been committed
    Committed,
    /// Transaction has been rolled back
    Aborted,
}

/// Transaction error types
#[derive(Debug)]
pub enum TxError {
    /// I/O error
    Io(io::Error),
    /// Pager error
    Pager(String),
    /// Internal lock was poisoned by a panic
    LockPoisoned(&'static str),
    /// Transaction is not active
    NotActive,
    /// Transaction already committed
    AlreadyCommitted,
    /// Transaction already aborted
    AlreadyAborted,
    /// Write conflict
    WriteConflict(u32),
    /// Invalid page data
    InvalidPage(String),
}

impl std::fmt::Display for TxError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Io(e) => write!(f, "I/O error: {}", e),
            Self::Pager(msg) => write!(f, "Pager error: {}", msg),
            Self::LockPoisoned(name) => write!(f, "Lock poisoned: {}", name),
            Self::NotActive => write!(f, "Transaction is not active"),
            Self::AlreadyCommitted => write!(f, "Transaction already committed"),
            Self::AlreadyAborted => write!(f, "Transaction already aborted"),
            Self::WriteConflict(page_id) => write!(f, "Write conflict on page {}", page_id),
            Self::InvalidPage(msg) => write!(f, "Invalid page: {}", msg),
        }
    }
}

impl std::error::Error for TxError {}

impl From<io::Error> for TxError {
    fn from(e: io::Error) -> Self {
        Self::Io(e)
    }
}

/// A buffered page write
#[derive(Clone)]
struct BufferedWrite {
    page_id: u32,
    data: [u8; PAGE_SIZE],
}

/// A single transaction
///
/// Transactions buffer writes and commit them atomically to the WAL.
pub struct Transaction {
    /// Transaction ID
    id: u64,
    /// Transaction state
    state: TxState,
    /// Buffered page writes (page_id -> page data)
    write_set: HashMap<u32, BufferedWrite>,
    /// Pages read in this transaction (for conflict detection)
    read_set: Vec<u32>,
    /// Reference to the transaction manager
    manager: Arc<TransactionManager>,
}

impl Transaction {
    /// Get transaction ID
    pub fn id(&self) -> u64 {
        self.id
    }

    /// Get transaction state
    pub fn state(&self) -> TxState {
        self.state
    }

    /// Check if transaction is active
    pub fn is_active(&self) -> bool {
        self.state == TxState::Active
    }

    /// Read a page through this transaction
    ///
    /// If the page has been written in this transaction, returns the buffered version.
    /// Otherwise, reads from the pager.
    pub fn read_page(&mut self, page_id: u32) -> Result<Page, TxError> {
        if self.state != TxState::Active {
            return Err(TxError::NotActive);
        }

        // Check write set first
        if let Some(buffered) = self.write_set.get(&page_id) {
            return Ok(Page::from_bytes(buffered.data));
        }

        // Track the read
        self.read_set.push(page_id);

        // Read from pager
        self.manager
            .pager
            .read_page(page_id)
            .map_err(|e| TxError::Pager(e.to_string()))
    }

    /// Write a page through this transaction
    ///
    /// The write is buffered and will be committed to the WAL on commit.
    pub fn write_page(&mut self, page_id: u32, page: Page) -> Result<(), TxError> {
        if self.state != TxState::Active {
            return Err(TxError::NotActive);
        }

        // Buffer the write
        let mut data = [0u8; PAGE_SIZE];
        data.copy_from_slice(page.as_bytes());

        self.write_set
            .insert(page_id, BufferedWrite { page_id, data });

        Ok(())
    }

    /// Commit the transaction
    ///
    /// Writes all buffered pages to the WAL, then writes a Commit record.
    ///
    /// **Read-only fast path:** when `write_set` is empty, the
    /// transaction wrote nothing, so there is nothing to make
    /// durable. We skip the WAL append, the `wal.sync()` (which costs
    /// ~100 µs of fsync), and the pager apply loop entirely. The
    /// transaction still transitions to `Committed` and unregisters
    /// from the manager so subsequent state checks work correctly.
    /// This mirrors postgres' optimisation in `RecordTransactionCommit`
    /// (`xact.c`) which skips `XLogFlush` when nothing was written.
    pub fn commit(mut self) -> Result<(), TxError> {
        if self.state != TxState::Active {
            return match self.state {
                TxState::Committed => Err(TxError::AlreadyCommitted),
                TxState::Aborted => Err(TxError::AlreadyAborted),
                _ => Err(TxError::NotActive),
            };
        }

        // ── Read-only fast path ─────────────────────────────────────
        // No writes → no WAL record → no fsync. Saves ~100 µs per
        // read-only commit and removes contention on the WAL writer
        // mutex for read-heavy workloads.
        if self.write_set.is_empty() {
            self.state = TxState::Committed;
            self.manager.unregister_transaction(self.id);
            self.manager.snapshot_manager.commit(self.id);
            return Ok(());
        }

        // ── Encode phase (no lock) ──────────────────────────────────
        // Encode every WAL record into one contiguous byte blob
        // OUTSIDE any lock. This is the bulk of the per-commit work
        // and pays no contention cost — the old path encoded each
        // record while holding `Mutex<WalWriter>`, which under 16-way
        // concurrency produced the park-convoy that bottlenecked
        // `concurrent` and `insert_sequential`.
        let mut blob = Vec::with_capacity(64 + self.write_set.len() * (PAGE_SIZE + 32));
        for (page_id, buffered) in &self.write_set {
            let record = WalRecord::PageWrite {
                tx_id: self.id,
                page_id: *page_id,
                data: buffered.data.to_vec(),
            };
            // Encode straight into the per-call blob scratch — no fresh Vec
            // per record. `blob` is owned by this commit frame and never
            // shared across appenders, so this is safe on the lock-free path.
            record.encode_into(&mut blob);
        }
        WalRecord::Commit { tx_id: self.id }.encode_into(&mut blob);

        // ── Reserve + enqueue (lock-free) ───────────────────────────
        // One atomic fetch_add reserves our LSN range; one
        // SegQueue::push hands the bytes to the leader. Both
        // operations are wait-free for the writer.
        let commit_lsn = self.manager.coordinator.reserve_and_enqueue(blob);

        // ── Wait for durability ─────────────────────────────────────
        // If `durable_lsn >= commit_lsn` already, the leader (some
        // earlier thread) covered us — return immediately. Otherwise
        // we either become the leader and drive the drain, or park
        // on the coordinator's parking_lot::Condvar until a leader
        // publishes a `durable_lsn` past our target.
        self.manager
            .coordinator
            .commit_at_least(commit_lsn, &self.manager.wal)
            .map_err(TxError::Io)?;

        // Apply writes to pager cache (for immediate visibility)
        for (page_id, buffered) in &self.write_set {
            let page = Page::from_bytes(buffered.data);
            self.manager
                .pager
                .write_page(*page_id, page)
                .map_err(|e| TxError::Pager(e.to_string()))?;
        }

        self.state = TxState::Committed;

        // Unregister from manager
        self.manager.unregister_transaction(self.id);
        self.manager.snapshot_manager.commit(self.id);

        Ok(())
    }

    /// Rollback the transaction
    ///
    /// Discards all buffered writes and writes a Rollback record to the WAL.
    pub fn rollback(mut self) -> Result<(), TxError> {
        if self.state != TxState::Active {
            return match self.state {
                TxState::Committed => Err(TxError::AlreadyCommitted),
                TxState::Aborted => Err(TxError::AlreadyAborted),
                _ => Err(TxError::NotActive),
            };
        }

        // Route rollback through the coordinator so its bytes land
        // in LSN order with any concurrent commits. Going around the
        // coordinator (direct `wal.lock().append`) would race with
        // the leader's `append_bytes` and corrupt the file.
        let blob = WalRecord::Rollback { tx_id: self.id }.encode();
        let target = self.manager.coordinator.reserve_and_enqueue(blob);
        self.manager
            .coordinator
            .commit_at_least(target, &self.manager.wal)
            .map_err(TxError::Io)?;

        // Clear write set
        self.write_set.clear();
        self.state = TxState::Aborted;

        // Unregister from manager
        self.manager.unregister_transaction(self.id);
        self.manager.snapshot_manager.rollback(self.id);

        Ok(())
    }
}

impl Drop for Transaction {
    fn drop(&mut self) {
        // If transaction is still active when dropped, it means it was neither
        // committed nor rolled back. This is a bug, but we'll clean up anyway.
        if self.state == TxState::Active {
            // Best-effort rollback through the coordinator. We can't
            // bypass the coordinator with a direct `wal.lock()` here:
            // any in-flight reservations would still be holding LSN
            // slots ahead of us and the file would gain a hole. The
            // coordinator handles ordering correctly even from Drop.
            let blob = WalRecord::Rollback { tx_id: self.id }.encode();
            let target = self.manager.coordinator.reserve_and_enqueue(blob);
            let _ = self
                .manager
                .coordinator
                .commit_at_least(target, &self.manager.wal);
            self.manager.unregister_transaction(self.id);
            self.manager.snapshot_manager.rollback(self.id);
        }
    }
}

/// Transaction Manager
///
/// Coordinates transactions and manages the WAL.
pub struct TransactionManager {
    /// Pager for reading/writing pages
    pager: Arc<Pager>,
    /// WAL writer protected by a `parking_lot::Mutex`. The mutex is
    /// taken only by the leader-flush path inside the coordinator;
    /// concurrent writers no longer contend on it during the
    /// encode-and-append step. See [`WalAppendCoordinator`].
    wal: Mutex<WalWriter>,
    /// WAL file path
    wal_path: PathBuf,
    /// Active transaction IDs
    active_transactions: RwLock<Vec<u64>>,
    /// Shared XID authority used by row MVCC and page-level WAL transactions.
    snapshot_manager: Arc<SnapshotManager>,
    /// Lock-free append coordinator. Replaces the per-commit
    /// `wal.lock()` that used to serialise 16 concurrent writers
    /// (Roadmap #2 / issue #157). Writers reserve an LSN range via
    /// atomic fetch_add, push their encoded bytes onto a SegQueue,
    /// then call `commit_at_least` to wait for durability. The
    /// first thread into `commit_at_least` becomes the leader and
    /// drives the WAL drain + fsync; everyone else parks on the
    /// coordinator's condvar.
    coordinator: WalAppendCoordinator,
}

impl TransactionManager {
    /// Create a new transaction manager
    ///
    /// # Arguments
    ///
    /// * `pager` - The pager to use for page I/O
    /// * `wal_path` - Path to the WAL file
    pub fn new(pager: Arc<Pager>, wal_path: impl AsRef<Path>) -> io::Result<Self> {
        Self::new_with_snapshot_manager(pager, wal_path, Arc::new(SnapshotManager::new()))
    }

    /// Create a transaction manager using the provided snapshot/XID manager.
    pub fn new_with_snapshot_manager(
        pager: Arc<Pager>,
        wal_path: impl AsRef<Path>,
        snapshot_manager: Arc<SnapshotManager>,
    ) -> io::Result<Self> {
        let wal_path = wal_path.as_ref().to_path_buf();
        let wal = WalWriter::open(&wal_path)?;
        let initial_current = wal.current_lsn();
        let initial_durable = wal.durable_lsn();

        Ok(Self {
            pager,
            wal: Mutex::new(wal),
            wal_path,
            active_transactions: RwLock::new(Vec::new()),
            snapshot_manager,
            coordinator: WalAppendCoordinator::new(initial_current, initial_durable),
        })
    }

    fn wal_writer(&self) -> Result<MutexGuard<'_, WalWriter>, TxError> {
        // parking_lot::Mutex does not poison on panic, so this is
        // infallible. We keep the `Result` return type to avoid
        // touching every call site, but the error variant is now
        // unreachable in normal operation. Tests that previously
        // poisoned the std::sync::Mutex deliberately have been
        // adjusted to assert the new non-poisoning behaviour.
        Ok(self.wal.lock())
    }

    fn active_transactions_write(&self) -> RwLockWriteGuard<'_, Vec<u64>> {
        self.active_transactions
            .write()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    fn active_transactions_read(&self) -> RwLockReadGuard<'_, Vec<u64>> {
        self.active_transactions
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    /// Begin a new transaction
    pub fn begin(self: &Arc<Self>) -> Result<Transaction, TxError> {
        let tx_id = self.snapshot_manager.begin();

        // Route the Begin record through the coordinator (same
        // ordering guarantees as commit/rollback). Begin is not
        // strictly required to be durable before subsequent appends
        // — recovery treats a Begin without a matching Commit as a
        // rolled-back txn — so we don't wait on `commit_at_least`.
        // The bytes are queued and the next leader picks them up
        // alongside our own Commit record when we eventually commit.
        let blob = WalRecord::Begin { tx_id }.encode();
        let _begin_lsn = self.coordinator.reserve_and_enqueue(blob);

        // Register transaction
        {
            let mut active = self.active_transactions_write();
            active.push(tx_id);
        }

        Ok(Transaction {
            id: tx_id,
            state: TxState::Active,
            write_set: HashMap::new(),
            read_set: Vec::new(),
            manager: Arc::clone(self),
        })
    }

    /// Unregister a transaction (called on commit/rollback)
    fn unregister_transaction(&self, tx_id: u64) {
        let mut active = self.active_transactions_write();
        active.retain(|&id| id != tx_id);
    }

    /// Get list of active transaction IDs
    pub fn active_transactions(&self) -> Vec<u64> {
        self.active_transactions_read().clone()
    }

    /// Get WAL file path
    pub fn wal_path(&self) -> &Path {
        &self.wal_path
    }

    /// Get reference to pager
    pub fn pager(&self) -> &Arc<Pager> {
        &self.pager
    }

    /// Sync WAL to disk. Drains every byte that has been reserved
    /// via the coordinator — i.e. waits until the coordinator's
    /// `next_lsn` is durable.
    pub fn sync_wal(&self) -> io::Result<()> {
        let target = self.coordinator.next_lsn();
        self.coordinator.commit_at_least(target, &self.wal)
    }

    /// Check if there are active transactions
    pub fn has_active_transactions(&self) -> bool {
        !self.active_transactions_read().is_empty()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::storage::engine::PageType;
    use std::fs;
    use std::time::{SystemTime, UNIX_EPOCH};

    fn temp_dir() -> PathBuf {
        let timestamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("value is present")
            .as_nanos();
        std::env::temp_dir().join(format!("reddb_tx_test_{}", timestamp))
    }

    fn cleanup(dir: &Path) {
        let _ = fs::remove_dir_all(dir);
    }

    fn temp_wal_path(dir: &Path, name: &str) -> PathBuf {
        reddb_file::layout::wal_component_temp_path(dir, "transaction", name, std::process::id())
    }

    #[test]
    fn test_transaction_commit() {
        let dir = temp_dir();
        let _ = fs::create_dir_all(&dir);
        let db_path = dir.join("test.db");
        let wal_path = temp_wal_path(&dir, "commit");

        // Create pager
        let pager = Arc::new(Pager::open_default(&db_path).expect("open_default() should succeed"));

        // Allocate a page
        let page = pager
            .allocate_page(PageType::BTreeLeaf)
            .expect("allocate_page() should succeed");
        let page_id = page.page_id();

        // Create transaction manager
        let tm = Arc::new(
            TransactionManager::new(Arc::clone(&pager), &wal_path).expect("new() should succeed"),
        );

        // Begin transaction
        let mut tx = tm.begin().expect("begin() should succeed");
        assert!(tx.is_active());

        // Write through transaction
        let mut page = Page::new(PageType::BTreeLeaf, page_id);
        page.as_bytes_mut()[100] = 0xAB;
        tx.write_page(page_id, page)
            .expect("write_page() should succeed");

        // Read through transaction (should see buffered write)
        let read_page = tx.read_page(page_id).expect("read_page() should succeed");
        assert_eq!(read_page.as_bytes()[100], 0xAB);

        // Commit
        tx.commit().expect("commit() should succeed");

        // Verify write is visible through pager
        let final_page = pager
            .read_page(page_id)
            .expect("read_page() should succeed");
        assert_eq!(final_page.as_bytes()[100], 0xAB);

        cleanup(&dir);
    }

    #[test]
    fn test_transaction_rollback() {
        let dir = temp_dir();
        let _ = fs::create_dir_all(&dir);
        let db_path = dir.join("test.db");
        let wal_path = temp_wal_path(&dir, "rollback");

        // Create pager
        let pager = Arc::new(Pager::open_default(&db_path).expect("open_default() should succeed"));

        // Allocate a page and write initial value
        let mut page = pager
            .allocate_page(PageType::BTreeLeaf)
            .expect("allocate_page() should succeed");
        let page_id = page.page_id();
        page.as_bytes_mut()[100] = 0x11;
        pager
            .write_page(page_id, page)
            .expect("write_page() should succeed");

        // Create transaction manager
        let tm = Arc::new(
            TransactionManager::new(Arc::clone(&pager), &wal_path).expect("new() should succeed"),
        );

        // Begin transaction
        let mut tx = tm.begin().expect("begin() should succeed");

        // Write through transaction
        let mut page = Page::new(PageType::BTreeLeaf, page_id);
        page.as_bytes_mut()[100] = 0xAB;
        tx.write_page(page_id, page)
            .expect("write_page() should succeed");

        // Read through transaction (should see buffered write)
        let read_page = tx.read_page(page_id).expect("read_page() should succeed");
        assert_eq!(read_page.as_bytes()[100], 0xAB);

        // Rollback
        tx.rollback().expect("rollback() should succeed");

        // Original value should be preserved
        let final_page = pager
            .read_page(page_id)
            .expect("read_page() should succeed");
        assert_eq!(final_page.as_bytes()[100], 0x11);

        cleanup(&dir);
    }

    #[test]
    fn test_multiple_transactions() {
        let dir = temp_dir();
        let _ = fs::create_dir_all(&dir);
        let db_path = dir.join("test.db");
        let wal_path = temp_wal_path(&dir, "multiple");

        // Create pager
        let pager = Arc::new(Pager::open_default(&db_path).expect("open_default() should succeed"));

        // Allocate two pages
        let page1 = pager
            .allocate_page(PageType::BTreeLeaf)
            .expect("allocate_page() should succeed");
        let page2 = pager
            .allocate_page(PageType::BTreeLeaf)
            .expect("allocate_page() should succeed");
        let page1_id = page1.page_id();
        let page2_id = page2.page_id();

        // Create transaction manager
        let tm = Arc::new(
            TransactionManager::new(Arc::clone(&pager), &wal_path).expect("new() should succeed"),
        );

        // Transaction 1: Write to page 1
        let mut tx1 = tm.begin().expect("begin() should succeed");
        let mut page1 = Page::new(PageType::BTreeLeaf, page1_id);
        page1.as_bytes_mut()[100] = 0x11;
        tx1.write_page(page1_id, page1)
            .expect("write_page() should succeed");
        tx1.commit().expect("commit() should succeed");

        // Transaction 2: Write to page 2
        let mut tx2 = tm.begin().expect("begin() should succeed");
        let mut page2 = Page::new(PageType::BTreeLeaf, page2_id);
        page2.as_bytes_mut()[100] = 0x22;
        tx2.write_page(page2_id, page2)
            .expect("write_page() should succeed");
        tx2.commit().expect("commit() should succeed");

        // Verify both writes
        let final_page1 = pager
            .read_page(page1_id)
            .expect("read_page() should succeed");
        let final_page2 = pager
            .read_page(page2_id)
            .expect("read_page() should succeed");
        assert_eq!(final_page1.as_bytes()[100], 0x11);
        assert_eq!(final_page2.as_bytes()[100], 0x22);

        cleanup(&dir);
    }

    #[test]
    fn test_transaction_isolation() {
        let dir = temp_dir();
        let _ = fs::create_dir_all(&dir);
        let db_path = dir.join("test.db");
        let wal_path = temp_wal_path(&dir, "isolation");

        // Create pager
        let pager = Arc::new(Pager::open_default(&db_path).expect("open_default() should succeed"));

        // Allocate a page with initial value
        let mut page = pager
            .allocate_page(PageType::BTreeLeaf)
            .expect("allocate_page() should succeed");
        let page_id = page.page_id();
        page.as_bytes_mut()[100] = 0x00;
        pager
            .write_page(page_id, page)
            .expect("write_page() should succeed");

        // Create transaction manager
        let tm = Arc::new(
            TransactionManager::new(Arc::clone(&pager), &wal_path).expect("new() should succeed"),
        );

        // Transaction 1: Begin and write (but don't commit yet)
        let mut tx1 = tm.begin().expect("begin() should succeed");
        let mut page1 = Page::new(PageType::BTreeLeaf, page_id);
        page1.as_bytes_mut()[100] = 0x11;
        tx1.write_page(page_id, page1)
            .expect("write_page() should succeed");

        // Transaction 1 should see its own write
        let tx1_read = tx1.read_page(page_id).expect("read_page() should succeed");
        assert_eq!(tx1_read.as_bytes()[100], 0x11);

        // Another read from pager should not see uncommitted write
        let pager_read = pager
            .read_page(page_id)
            .expect("read_page() should succeed");
        assert_eq!(pager_read.as_bytes()[100], 0x00);

        // Commit tx1
        tx1.commit().expect("commit() should succeed");

        // Now pager should see the write
        let final_read = pager
            .read_page(page_id)
            .expect("read_page() should succeed");
        assert_eq!(final_read.as_bytes()[100], 0x11);

        cleanup(&dir);
    }

    #[test]
    fn test_active_transaction_tracking() {
        let dir = temp_dir();
        let _ = fs::create_dir_all(&dir);
        let db_path = dir.join("test.db");
        let wal_path = temp_wal_path(&dir, "tracking");

        let pager = Arc::new(Pager::open_default(&db_path).expect("open_default() should succeed"));
        let tm = Arc::new(
            TransactionManager::new(Arc::clone(&pager), &wal_path).expect("new() should succeed"),
        );

        assert!(!tm.has_active_transactions());

        let tx1 = tm.begin().expect("begin() should succeed");
        let tx1_id = tx1.id();
        assert!(tm.has_active_transactions());
        assert!(tm.active_transactions().contains(&tx1_id));

        let tx2 = tm.begin().expect("begin() should succeed");
        let tx2_id = tx2.id();
        assert_eq!(tm.active_transactions().len(), 2);

        tx1.commit().expect("commit() should succeed");
        assert!(!tm.active_transactions().contains(&tx1_id));
        assert!(tm.active_transactions().contains(&tx2_id));

        tx2.rollback().expect("rollback() should succeed");
        assert!(!tm.has_active_transactions());

        cleanup(&dir);
    }

    #[test]
    fn page_wal_transactions_allocate_from_snapshot_manager() {
        let dir = temp_dir();
        let _ = fs::create_dir_all(&dir);
        let db_path = dir.join("test.db");
        let wal_path = temp_wal_path(&dir, "snapshot-xid");

        let pager = Arc::new(Pager::open_default(&db_path).expect("open_default() should succeed"));
        let snapshot_manager =
            Arc::new(crate::storage::transaction::snapshot::SnapshotManager::new());
        let row_xid = snapshot_manager.begin();
        snapshot_manager.commit(row_xid);

        let tm = Arc::new(
            TransactionManager::new_with_snapshot_manager(
                Arc::clone(&pager),
                &wal_path,
                Arc::clone(&snapshot_manager),
            )
            .expect("new() should succeed"),
        );

        let tx = tm.begin().expect("begin() should succeed");
        assert_eq!(tx.id(), row_xid + 1);
        tx.commit().expect("commit() should succeed");
        assert_eq!(snapshot_manager.peek_next_xid(), row_xid + 2);

        cleanup(&dir);
    }

    #[test]
    fn test_transaction_double_commit() {
        let dir = temp_dir();
        let _ = fs::create_dir_all(&dir);
        let db_path = dir.join("test.db");
        let wal_path = temp_wal_path(&dir, "double-commit");

        let pager = Arc::new(Pager::open_default(&db_path).expect("open_default() should succeed"));
        let tm = Arc::new(
            TransactionManager::new(Arc::clone(&pager), &wal_path).expect("new() should succeed"),
        );

        // The transaction is consumed on commit, so double commit is impossible at compile time
        // This test just verifies commit works
        let tx = tm.begin().expect("begin() should succeed");
        tx.commit().expect("commit() should succeed");

        cleanup(&dir);
    }

    #[test]
    fn test_begin_succeeds_after_panic_in_lock_holder() {
        // After Roadmap #2 the WAL mutex is `parking_lot::Mutex`,
        // which does NOT poison on panic. A previous version of this
        // test asserted that a panicking thread holding the lock
        // produced a `TxError::LockPoisoned` on the next `begin()` —
        // that was a property of `std::sync::Mutex`. The new
        // contract is the opposite: writers continue uninterrupted.
        let dir = temp_dir();
        let _ = fs::create_dir_all(&dir);
        let db_path = dir.join("test.db");
        let wal_path = temp_wal_path(&dir, "panic-lock-holder");

        let pager = Arc::new(Pager::open_default(&db_path).expect("open_default() should succeed"));
        let tm = Arc::new(
            TransactionManager::new(Arc::clone(&pager), &wal_path).expect("new() should succeed"),
        );

        let poison_target = Arc::clone(&tm);
        let _ = std::thread::spawn(move || {
            let _guard = poison_target.wal.lock();
            panic!("would-poison the wal mutex on std::sync");
        })
        .join();

        // parking_lot recovers transparently. begin() must succeed.
        match tm.begin() {
            Ok(_) => {}
            Err(err) => panic!("begin must succeed despite prior panic: {err:?}"),
        }

        cleanup(&dir);
    }

    // ---------------------------------------------------------------
    // Perf 1.2: read-only commit fast path
    // ---------------------------------------------------------------

    #[test]
    fn read_only_commit_does_not_advance_durable_lsn() {
        let dir = temp_dir();
        let _ = fs::create_dir_all(&dir);
        let db_path = dir.join("ro_durable.db");
        let wal_path = temp_wal_path(&dir, "ro-durable");

        let pager = Arc::new(Pager::open_default(&db_path).expect("open_default() should succeed"));
        let tm = Arc::new(
            TransactionManager::new(Arc::clone(&pager), &wal_path).expect("new() should succeed"),
        );

        // Snapshot the WAL durable_lsn BEFORE the txn.
        let before = {
            let wal = tm.wal_writer().expect("wal_writer() should succeed");
            wal.durable_lsn()
        };

        let tx = tm.begin().expect("begin() should succeed");
        // Empty write_set on purpose — read-only.
        tx.commit().expect("commit() should succeed");

        // After RO commit, the WAL durable_lsn must NOT have advanced.
        // No Begin record, no Commit record, no fsync.
        let after = {
            let wal = tm.wal_writer().expect("wal_writer() should succeed");
            wal.durable_lsn()
        };
        assert_eq!(
            before, after,
            "read-only commit must not advance durable_lsn (was {} → {})",
            before, after
        );

        cleanup(&dir);
    }

    #[test]
    fn read_only_commit_does_not_grow_wal_file() {
        let dir = temp_dir();
        let _ = fs::create_dir_all(&dir);
        let db_path = dir.join("ro_size.db");
        let wal_path = temp_wal_path(&dir, "ro-size");

        let pager = Arc::new(Pager::open_default(&db_path).expect("open_default() should succeed"));
        let tm = Arc::new(
            TransactionManager::new(Arc::clone(&pager), &wal_path).expect("new() should succeed"),
        );

        // Snapshot file size after WAL header.
        let size_before = std::fs::metadata(&wal_path)
            .expect("metadata() should succeed")
            .len();
        assert_eq!(
            size_before,
            reddb_file::WAL_FILE_HEADER_BYTES as u64,
            "fresh WAL must be exactly the 8-byte header"
        );

        // 100 read-only commits in a loop.
        for _ in 0..100 {
            let tx = tm.begin().expect("begin() should succeed");
            tx.commit().expect("commit() should succeed");
        }

        let size_after = std::fs::metadata(&wal_path)
            .expect("metadata() should succeed")
            .len();
        assert_eq!(
            size_after, size_before,
            "100 read-only commits should not have written any WAL bytes"
        );
        cleanup(&dir);
    }

    #[test]
    fn read_only_commit_marks_transaction_committed() {
        let dir = temp_dir();
        let _ = fs::create_dir_all(&dir);
        let db_path = dir.join("ro_state.db");
        let wal_path = temp_wal_path(&dir, "ro-state");

        let pager = Arc::new(Pager::open_default(&db_path).expect("open_default() should succeed"));
        let tm = Arc::new(
            TransactionManager::new(Arc::clone(&pager), &wal_path).expect("new() should succeed"),
        );

        let tx = tm.begin().expect("begin() should succeed");
        let id = tx.id();
        tx.commit().expect("commit() should succeed");

        // Manager must have unregistered this txn — the active list
        // no longer contains its id.
        assert!(
            !tm.active_transactions().contains(&id),
            "RO-committed txn {id} must no longer be active in the manager"
        );

        cleanup(&dir);
    }

    #[test]
    fn writing_commit_still_syncs_after_ro_fast_path() {
        // Sanity: the fast path must NOT short-circuit a transaction
        // that did write something. Verify the writing commit path
        // still flushes WAL and the value lands in the pager.
        let dir = temp_dir();
        let _ = fs::create_dir_all(&dir);
        let db_path = dir.join("rw_after_ro.db");
        let wal_path = temp_wal_path(&dir, "rw-after-ro");

        let pager = Arc::new(Pager::open_default(&db_path).expect("open_default() should succeed"));
        let allocated = pager
            .allocate_page(PageType::BTreeLeaf)
            .expect("allocate_page() should succeed");
        let page_id = allocated.page_id();
        let tm = Arc::new(
            TransactionManager::new(Arc::clone(&pager), &wal_path).expect("new() should succeed"),
        );

        // First a RO commit (must take the fast path).
        let ro = tm.begin().expect("begin() should succeed");
        ro.commit().expect("commit() should succeed");

        // Then a real writing commit.
        let mut rw = tm.begin().expect("begin() should succeed");
        let mut page = Page::new(PageType::BTreeLeaf, page_id);
        page.as_bytes_mut()[42] = 0x77;
        rw.write_page(page_id, page)
            .expect("write_page() should succeed");
        rw.commit().expect("commit() should succeed");

        // The WAL file must now contain bytes (PageWrite + Commit
        // records, and the BufWriter has been flushed by sync()).
        let size = std::fs::metadata(&wal_path)
            .expect("metadata() should succeed")
            .len();
        assert!(size > 8, "writing commit should grow the WAL");

        // The pager cache must reflect the write.
        let read_back = pager
            .read_page(page_id)
            .expect("read_page() should succeed");
        assert_eq!(read_back.as_bytes()[42], 0x77);

        cleanup(&dir);
    }
}