oxirs-tdb 0.3.1

//! Group commit optimization for WAL
//!
//! This module implements group commit, a technique that batches multiple
//! transaction commits together to reduce the number of fsync() operations.
//!
//! Benefits:
//! - Reduces disk I/O overhead by batching fsync calls
//! - Improves throughput for write-heavy workloads
//! - Maintains durability guarantees (all commits are written before ack)

use super::wal::{Lsn, TxnId, WriteAheadLog};
use crate::error::Result;
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};

/// Configuration for group commit
#[derive(Debug, Clone)]
pub struct GroupCommitConfig {
    /// Maximum number of commits to batch together
    pub max_batch_size: usize,
    /// Maximum time to wait before flushing (even if batch not full)
    pub max_wait_time: Duration,
    /// Whether group commit is enabled
    pub enabled: bool,
}

impl Default for GroupCommitConfig {
    fn default() -> Self {
        Self {
            max_batch_size: 100,
            max_wait_time: Duration::from_millis(10),
            enabled: true,
        }
    }
}

/// Pending commit request
#[derive(Debug)]
struct PendingCommit {
    /// Transaction ID
    txn_id: TxnId,
    /// LSN of commit record
    commit_lsn: Lsn,
    /// Time when commit was requested
    requested_at: Instant,
}

/// Group commit coordinator
///
/// Batches multiple transaction commits together to reduce fsync overhead.
/// When a transaction wants to commit:
/// 1. It adds itself to the pending queue
/// 2. It waits on the condition variable
/// 3. When batch is full or timeout expires, a flush occurs
/// 4. All transactions in the batch are notified
pub struct GroupCommitCoordinator {
    /// Configuration
    config: GroupCommitConfig,
    /// Write-ahead log
    wal: Arc<WriteAheadLog>,
    /// Pending commits
    pending: Arc<Mutex<PendingCommitQueue>>,
    /// Condition variable for commit notifications
    commit_cv: Arc<Condvar>,
    /// Statistics
    stats: Arc<Mutex<GroupCommitStats>>,
}

/// Queue of pending commits
struct PendingCommitQueue {
    /// List of pending commits
    commits: Vec<PendingCommit>,
    /// Last flush time
    last_flush: Instant,
    /// Last flushed LSN
    last_flushed_lsn: Lsn,
}

impl PendingCommitQueue {
    fn new() -> Self {
        Self {
            commits: Vec::new(),
            last_flush: Instant::now(),
            last_flushed_lsn: Lsn::ZERO,
        }
    }

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

    fn len(&self) -> usize {
        self.commits.len()
    }

    fn should_flush(&self, config: &GroupCommitConfig) -> bool {
        if self.commits.is_empty() {
            return false;
        }

        // Flush if batch is full
        if self.commits.len() >= config.max_batch_size {
            return true;
        }

        // Flush if oldest commit has waited too long
        if let Some(oldest) = self.commits.first() {
            if oldest.requested_at.elapsed() >= config.max_wait_time {
                return true;
            }
        }

        false
    }

    fn drain_batch(&mut self) -> Vec<PendingCommit> {
        std::mem::take(&mut self.commits)
    }
}

/// Group commit statistics
#[derive(Debug, Default)]
pub struct GroupCommitStats {
    /// Total number of commits processed
    pub total_commits: u64,
    /// Total number of flush operations
    pub total_flushes: u64,
    /// Average batch size
    pub avg_batch_size: f64,
    /// Total wait time across all commits (microseconds)
    pub total_wait_time_us: u64,
    /// Maximum wait time seen (microseconds)
    pub max_wait_time_us: u64,
}

impl GroupCommitCoordinator {
    /// Create a new group commit coordinator
    pub fn new(wal: Arc<WriteAheadLog>, config: GroupCommitConfig) -> Self {
        Self {
            config,
            wal,
            pending: Arc::new(Mutex::new(PendingCommitQueue::new())),
            commit_cv: Arc::new(Condvar::new()),
            stats: Arc::new(Mutex::new(GroupCommitStats::default())),
        }
    }

    /// Request a commit (blocking until commit is durable)
    ///
    /// This method adds the transaction to the pending queue and waits
    /// until the WAL is flushed. Multiple transactions may be flushed
    /// together in a single batch.
    pub fn commit(&self, txn_id: TxnId, commit_lsn: Lsn) -> Result<()> {
        let requested_at = Instant::now();

        if !self.config.enabled {
            // Group commit disabled - flush immediately
            self.wal.flush()?;

            // Still update stats
            let wait_time_us = requested_at.elapsed().as_micros() as u64;
            let mut stats = self.stats.lock().expect("lock poisoned");
            stats.total_commits += 1;
            stats.total_flushes += 1; // Each commit flushes immediately when disabled
            stats.total_wait_time_us += wait_time_us;
            stats.max_wait_time_us = stats.max_wait_time_us.max(wait_time_us);

            return Ok(());
        }

        // Add to pending queue
        {
            let mut pending = self.pending.lock().expect("lock poisoned");
            pending.commits.push(PendingCommit {
                txn_id,
                commit_lsn,
                requested_at,
            });
        }

        // Try to flush if conditions are met
        self.try_flush()?;

        // Wait for flush (with timeout to prevent deadlock)
        let timeout = self.config.max_wait_time * 2; // Safety margin
        self.wait_for_flush(commit_lsn, timeout)?;

        // Record wait time
        let wait_time_us = requested_at.elapsed().as_micros() as u64;
        let mut stats = self.stats.lock().expect("lock poisoned");
        stats.total_commits += 1;
        stats.total_wait_time_us += wait_time_us;
        stats.max_wait_time_us = stats.max_wait_time_us.max(wait_time_us);

        Ok(())
    }

    /// Try to flush pending commits if conditions are met
    fn try_flush(&self) -> Result<()> {
        let mut pending = self.pending.lock().expect("lock poisoned");

        if !pending.should_flush(&self.config) {
            return Ok(());
        }

        // Take the batch
        let batch = pending.drain_batch();
        let batch_size = batch.len();

        // Record flush time
        pending.last_flush = Instant::now();

        // Find highest LSN in batch
        let max_lsn = batch
            .iter()
            .map(|c| c.commit_lsn)
            .max()
            .unwrap_or(Lsn::ZERO);

        // Release lock before flushing (don't hold lock during I/O)
        drop(pending);

        // Flush WAL to disk
        self.wal.flush()?;

        // Update flushed LSN and notify waiters
        {
            let mut pending = self.pending.lock().expect("lock poisoned");
            pending.last_flushed_lsn = max_lsn;
        }

        // Notify all waiting transactions
        self.commit_cv.notify_all();

        // Update statistics
        {
            let mut stats = self.stats.lock().expect("lock poisoned");
            stats.total_flushes += 1;
            let total_commits = stats.total_commits as f64;
            stats.avg_batch_size = (stats.avg_batch_size * (total_commits - batch_size as f64)
                + batch_size as f64)
                / total_commits.max(1.0);
        }

        Ok(())
    }

    /// Wait for a specific LSN to be flushed
    fn wait_for_flush(&self, target_lsn: Lsn, timeout: Duration) -> Result<()> {
        let deadline = Instant::now() + timeout;

        let mut pending = self.pending.lock().expect("lock poisoned");

        loop {
            // Check if already flushed
            if pending.last_flushed_lsn >= target_lsn {
                return Ok(());
            }

            // Wait with timeout
            let now = Instant::now();
            if now >= deadline {
                // Timeout - force flush
                drop(pending);
                self.force_flush()?;
                return Ok(());
            }

            let remaining = deadline.duration_since(now);
            let (guard, timeout_result) = self
                .commit_cv
                .wait_timeout(pending, remaining)
                .expect("lock poisoned");
            pending = guard;

            if timeout_result.timed_out() {
                // Timeout - force flush
                drop(pending);
                self.force_flush()?;
                return Ok(());
            }
        }
    }

    /// Force an immediate flush of pending commits
    pub fn force_flush(&self) -> Result<()> {
        let mut pending = self.pending.lock().expect("lock poisoned");

        if pending.is_empty() {
            return Ok(());
        }

        let batch = pending.drain_batch();
        let batch_size = batch.len();

        let max_lsn = batch
            .iter()
            .map(|c| c.commit_lsn)
            .max()
            .unwrap_or(Lsn::ZERO);

        pending.last_flush = Instant::now();

        drop(pending);

        // Flush WAL
        self.wal.flush()?;

        // Update flushed LSN
        {
            let mut pending = self.pending.lock().expect("lock poisoned");
            pending.last_flushed_lsn = max_lsn;
        }

        // Notify waiters
        self.commit_cv.notify_all();

        // Update stats
        {
            let mut stats = self.stats.lock().expect("lock poisoned");
            stats.total_flushes += 1;
            let total_commits = stats.total_commits as f64;
            stats.avg_batch_size = (stats.avg_batch_size * (total_commits - batch_size as f64)
                + batch_size as f64)
                / total_commits.max(1.0);
        }

        Ok(())
    }

    /// Get current statistics
    pub fn stats(&self) -> GroupCommitStats {
        let stats = self.stats.lock().expect("lock poisoned");
        GroupCommitStats {
            total_commits: stats.total_commits,
            total_flushes: stats.total_flushes,
            avg_batch_size: stats.avg_batch_size,
            total_wait_time_us: stats.total_wait_time_us,
            max_wait_time_us: stats.max_wait_time_us,
        }
    }

    /// Get average commits per flush
    pub fn avg_commits_per_flush(&self) -> f64 {
        let stats = self.stats.lock().expect("lock poisoned");
        if stats.total_flushes == 0 {
            0.0
        } else {
            stats.total_commits as f64 / stats.total_flushes as f64
        }
    }

    /// Get number of pending commits
    pub fn pending_count(&self) -> usize {
        self.pending.lock().expect("lock poisoned").len()
    }
}

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

    #[test]
    fn test_group_commit_config() {
        let config = GroupCommitConfig::default();
        assert_eq!(config.max_batch_size, 100);
        assert_eq!(config.max_wait_time, Duration::from_millis(10));
        assert!(config.enabled);
    }

    #[test]
    fn test_single_commit() {
        let temp_dir = env::temp_dir().join("oxirs_group_commit_single");
        std::fs::create_dir_all(&temp_dir).unwrap();

        let wal = Arc::new(WriteAheadLog::new(&temp_dir).unwrap());
        let coordinator = GroupCommitCoordinator::new(wal.clone(), GroupCommitConfig::default());

        let lsn = wal
            .append(super::super::wal::LogRecord::Commit {
                txn_id: TxnId::new(1),
            })
            .unwrap();

        coordinator.commit(TxnId::new(1), lsn).unwrap();

        // Explicitly force flush to ensure stats are updated
        coordinator.force_flush().unwrap();

        let stats = coordinator.stats();
        assert_eq!(stats.total_commits, 1);
        assert!(stats.total_flushes >= 1);

        std::fs::remove_dir_all(&temp_dir).ok();
    }

    #[test]
    fn test_batch_commit() {
        let temp_dir = env::temp_dir().join("oxirs_group_commit_batch");
        std::fs::create_dir_all(&temp_dir).unwrap();

        let wal = Arc::new(WriteAheadLog::new(&temp_dir).unwrap());
        let config = GroupCommitConfig {
            max_batch_size: 5,
            max_wait_time: Duration::from_millis(50), // Reduced from 100ms
            enabled: true,
        };
        let coordinator = Arc::new(GroupCommitCoordinator::new(wal.clone(), config));

        // Spawn multiple threads to commit concurrently (reduced for faster tests)
        let mut handles = vec![];
        for i in 0..5 {
            let coordinator = Arc::clone(&coordinator);
            let wal = Arc::clone(&wal);
            let handle = thread::spawn(move || {
                let lsn = wal
                    .append(super::super::wal::LogRecord::Commit {
                        txn_id: TxnId::new(i),
                    })
                    .unwrap();
                coordinator.commit(TxnId::new(i), lsn).unwrap();
            });
            handles.push(handle);
        }

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

        let stats = coordinator.stats();
        assert_eq!(stats.total_commits, 5);
        // Should have fewer flushes than commits due to batching
        assert!(stats.total_flushes <= 5);
        assert!(stats.avg_batch_size >= 1.0);

        std::fs::remove_dir_all(&temp_dir).ok();
    }

    #[test]
    fn test_force_flush() {
        let temp_dir = env::temp_dir().join("oxirs_group_commit_force");
        std::fs::create_dir_all(&temp_dir).unwrap();

        let wal = Arc::new(WriteAheadLog::new(&temp_dir).unwrap());
        let config = GroupCommitConfig {
            max_batch_size: 100,
            max_wait_time: Duration::from_millis(50), // Short timeout for tests
            enabled: false,                           // Disable to manually control flushing
        };
        let coordinator = Arc::new(GroupCommitCoordinator::new(wal.clone(), config));

        // Do sequential commits instead of concurrent to avoid potential deadlocks
        for i in 0..3 {
            let lsn = wal
                .append(super::super::wal::LogRecord::Commit {
                    txn_id: TxnId::new(i),
                })
                .unwrap();
            // With group commit disabled, each commit flushes immediately
            coordinator.commit(TxnId::new(i), lsn).unwrap();
        }

        let stats = coordinator.stats();
        // With group commit disabled, each commit flushes immediately
        assert_eq!(stats.total_commits, 3);
        assert!(stats.total_flushes >= 1);

        std::fs::remove_dir_all(&temp_dir).ok();
    }

    #[test]
    fn test_disabled_group_commit() {
        let temp_dir = env::temp_dir().join("oxirs_group_commit_disabled");
        std::fs::create_dir_all(&temp_dir).unwrap();

        let wal = Arc::new(WriteAheadLog::new(&temp_dir).unwrap());
        let config = GroupCommitConfig {
            enabled: false,
            ..Default::default()
        };
        let coordinator = GroupCommitCoordinator::new(wal.clone(), config);

        for i in 0..5 {
            let lsn = wal
                .append(super::super::wal::LogRecord::Commit {
                    txn_id: TxnId::new(i),
                })
                .unwrap();
            coordinator.commit(TxnId::new(i), lsn).unwrap();
        }

        let stats = coordinator.stats();
        assert_eq!(stats.total_commits, 5);
        // With group commit disabled, each commit should flush immediately
        assert_eq!(stats.avg_batch_size, 0.0);

        std::fs::remove_dir_all(&temp_dir).ok();
    }

    #[test]
    #[ignore] // Timing-dependent test, not suitable for CI
    fn test_timeout_flush() {
        let temp_dir = env::temp_dir().join("oxirs_group_commit_timeout");
        std::fs::create_dir_all(&temp_dir).unwrap();

        let wal = Arc::new(WriteAheadLog::new(&temp_dir).unwrap());
        let config = GroupCommitConfig {
            max_batch_size: 100,
            max_wait_time: Duration::from_millis(50), // Short timeout
            enabled: true,
        };
        let coordinator = Arc::new(GroupCommitCoordinator::new(wal.clone(), config));

        // Single commit should timeout and flush
        let lsn = wal
            .append(super::super::wal::LogRecord::Commit {
                txn_id: TxnId::new(1),
            })
            .unwrap();

        coordinator.commit(TxnId::new(1), lsn).unwrap();

        let stats = coordinator.stats();
        assert_eq!(stats.total_commits, 1);
        assert!(stats.max_wait_time_us > 0);

        std::fs::remove_dir_all(&temp_dir).ok();
    }

    #[test]
    fn test_concurrent_commits() {
        let temp_dir = env::temp_dir().join("oxirs_group_commit_concurrent");
        std::fs::create_dir_all(&temp_dir).unwrap();

        let wal = Arc::new(WriteAheadLog::new(&temp_dir).unwrap());
        let config = GroupCommitConfig {
            max_batch_size: 10,
            max_wait_time: Duration::from_millis(50), // Reduced from 100ms
            enabled: true,
        };
        let coordinator = Arc::new(GroupCommitCoordinator::new(wal.clone(), config));

        // Concurrent commits (reduced to 5 for faster tests)
        let mut handles = vec![];
        for i in 0..5 {
            let coordinator = Arc::clone(&coordinator);
            let wal = Arc::clone(&wal);
            let handle = thread::spawn(move || {
                let lsn = wal
                    .append(super::super::wal::LogRecord::Commit {
                        txn_id: TxnId::new(i),
                    })
                    .unwrap();
                coordinator.commit(TxnId::new(i), lsn).unwrap();
            });
            handles.push(handle);
        }

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

        let stats = coordinator.stats();
        assert_eq!(stats.total_commits, 5);
        // With batch size of 10 and 5 commits, batching may or may not occur
        assert!(stats.total_flushes <= 5); // Should not exceed total commits

        std::fs::remove_dir_all(&temp_dir).ok();
    }
}