crush-core 0.3.0

//! Timeout protection for plugin operations
//!
//! Implements thread-based timeout enforcement with cooperative cancellation.
//! Uses crossbeam channels for reliable timeout detection and `Arc<AtomicBool>`
//! for cooperative cancellation within plugins.

use crate::cancel::CancellationToken;
use crate::error::{Result, TimeoutError};
use crossbeam::channel;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

/// RAII guard that sets cancellation flag on drop
///
/// When this guard is dropped (either normally or due to panic), it sets
/// the cancellation flag to signal the plugin to stop processing.
pub struct TimeoutGuard {
    cancel_flag: Arc<AtomicBool>,
}

impl Drop for TimeoutGuard {
    fn drop(&mut self) {
        // Signal cancellation when guard is dropped (timeout or panic)
        self.cancel_flag.store(true, Ordering::Release);
    }
}

/// Run an operation with timeout protection
///
/// Spawns the operation in a dedicated thread and enforces the specified timeout.
/// If the operation doesn't complete within the timeout, the cancellation flag
/// is set and an error is returned.
///
/// # Arguments
///
/// * `timeout` - Maximum duration to wait for operation completion (0 = no timeout)
/// * `operation` - The operation to run (receives cancellation flag)
///
/// # Returns
///
/// The operation's result if it completes within timeout, otherwise a timeout error
///
/// # Errors
///
/// Returns an error if:
/// - Operation times out
/// - Plugin thread panics during execution
/// - Operation returns an error
///
/// # Examples
///
/// ```no_run
/// use crush_core::plugin::timeout::run_with_timeout;
/// use std::sync::Arc;
/// use std::sync::atomic::AtomicBool;
/// use std::time::Duration;
///
/// let timeout = Duration::from_secs(5);
/// let result = run_with_timeout(timeout, |cancel_flag| {
///     // Operation code here
///     Ok(vec![1, 2, 3])
/// });
/// ```
pub fn run_with_timeout<F, T>(timeout: Duration, operation: F) -> Result<T>
where
    F: FnOnce(Arc<AtomicBool>) -> Result<T> + Send + 'static,
    T: Send + 'static,
{
    // Timeout of 0 means no timeout - use Duration::MAX for effectively infinite wait
    let effective_timeout = if timeout == Duration::from_secs(0) {
        Duration::MAX
    } else {
        timeout
    };

    let cancel_flag = Arc::new(AtomicBool::new(false));
    let cancel_flag_thread = Arc::clone(&cancel_flag);
    let cancel_flag_guard = Arc::clone(&cancel_flag);

    let (tx, rx) = channel::bounded(1);

    // Spawn operation in dedicated thread
    std::thread::spawn(move || {
        let _guard = TimeoutGuard {
            cancel_flag: cancel_flag_guard,
        };

        // Run operation and send result
        let result = operation(cancel_flag_thread);
        let _ = tx.send(result); // Ignore send errors (receiver might have timed out)
    });

    // Wait for completion or timeout
    match rx.recv_timeout(effective_timeout) {
        Ok(result) => result,
        Err(channel::RecvTimeoutError::Timeout) => {
            eprintln!("Warning: Plugin operation timed out after {timeout:?}");
            Err(TimeoutError::Timeout(timeout).into())
        }
        Err(channel::RecvTimeoutError::Disconnected) => {
            eprintln!("Warning: Plugin thread panicked during execution");
            Err(TimeoutError::PluginPanic.into())
        }
    }
}

/// Run an operation with timeout protection and external cancellation support
///
/// This version supports both timeout-based cancellation and external cancellation
/// via a `CancellationToken` (e.g., for Ctrl+C handling).
///
/// # Arguments
///
/// * `timeout` - Maximum duration to wait for operation completion (0 = no timeout)
/// * `cancel_token` - Optional external cancellation token
/// * `operation` - The operation to run (receives cancellation flag)
///
/// # Returns
///
/// The operation's result if it completes, otherwise a timeout or cancellation error
///
/// # Errors
///
/// Returns an error if:
/// - Operation times out
/// - External cancellation is triggered
/// - Plugin thread panics during execution
/// - Operation returns an error
pub fn run_with_timeout_and_cancel<F, T>(
    timeout: Duration,
    cancel_token: Option<Arc<dyn CancellationToken>>,
    operation: F,
) -> Result<T>
where
    F: FnOnce(Arc<AtomicBool>) -> Result<T> + Send + 'static,
    T: Send + 'static,
{
    // Check if already cancelled before starting
    if let Some(ref token) = cancel_token {
        if token.is_cancelled() {
            return Err(crate::error::CrushError::Cancelled);
        }
    }

    // Timeout of 0 means no timeout - use Duration::MAX for effectively infinite wait
    let effective_timeout = if timeout == Duration::from_secs(0) {
        Duration::MAX
    } else {
        timeout
    };

    let cancel_flag = Arc::new(AtomicBool::new(false));
    let cancel_flag_thread = Arc::clone(&cancel_flag);
    let cancel_flag_guard = Arc::clone(&cancel_flag);
    let cancel_flag_monitor = Arc::clone(&cancel_flag);

    // Spawn a monitor thread for external cancellation token
    let monitor_handle = if let Some(token) = cancel_token {
        let handle = std::thread::spawn(move || {
            // Poll the external token very frequently for responsive cancellation
            while !cancel_flag_monitor.load(Ordering::Acquire) {
                if token.is_cancelled() {
                    // External cancellation requested - signal the plugin
                    cancel_flag_monitor.store(true, Ordering::Release);
                    break;
                }
                // Use a very short sleep for fast response
                std::thread::sleep(Duration::from_micros(100));
            }
        });
        Some(handle)
    } else {
        None
    };

    let (tx, rx) = channel::bounded(1);

    // Spawn operation in dedicated thread
    std::thread::spawn(move || {
        let _guard = TimeoutGuard {
            cancel_flag: cancel_flag_guard,
        };

        // Run operation and send result
        let result = operation(cancel_flag_thread);
        let _ = tx.send(result); // Ignore send errors (receiver might have timed out)
    });

    // Wait for completion or timeout
    let result = match rx.recv_timeout(effective_timeout) {
        Ok(result) => {
            // Convert PluginError::Cancelled to CrushError::Cancelled
            match result {
                Err(crate::error::CrushError::Plugin(crate::error::PluginError::Cancelled)) => {
                    Err(crate::error::CrushError::Cancelled)
                }
                other => other,
            }
        }
        Err(channel::RecvTimeoutError::Timeout) => {
            // Signal cancellation to the operation
            cancel_flag.store(true, Ordering::Release);
            eprintln!("Warning: Plugin operation timed out after {timeout:?}");
            Err(TimeoutError::Timeout(timeout).into())
        }
        Err(channel::RecvTimeoutError::Disconnected) => {
            eprintln!("Warning: Plugin thread panicked during execution");
            Err(TimeoutError::PluginPanic.into())
        }
    };

    // Stop the monitor thread if it's still running
    cancel_flag.store(true, Ordering::Release);
    if let Some(handle) = monitor_handle {
        let _ = handle.join(); // Wait for monitor to finish
    }

    result
}

/// Scoped variant of [`run_with_timeout`] — operation may borrow non-`'static` data.
///
/// Uses `std::thread::scope` so the closure runs on a thread whose lifetime is bounded
/// by `'scope`. This lets hot-path callers pass `&[u8]` directly instead of allocating
/// an owned clone just to satisfy the `'static` bound of [`run_with_timeout`].
///
/// Semantics (timeout, panic, 0 = infinite) are identical to [`run_with_timeout`].
///
/// Note: kept `pub(crate)` to preserve the strict zero-diff public API contract
/// (SC-007). See specs/011-perf-optimizations/contracts/public-api.md.
///
/// # Errors
///
/// Same as [`run_with_timeout`].
pub(crate) fn run_with_timeout_scoped<'scope, 'env, F, T>(
    scope: &'scope std::thread::Scope<'scope, 'env>,
    timeout: Duration,
    operation: F,
) -> Result<T>
where
    F: FnOnce(Arc<AtomicBool>) -> Result<T> + Send + 'scope,
    T: Send + 'scope,
{
    let effective_timeout = if timeout == Duration::from_secs(0) {
        Duration::MAX
    } else {
        timeout
    };

    let cancel_flag = Arc::new(AtomicBool::new(false));
    let cancel_flag_thread = Arc::clone(&cancel_flag);
    let cancel_flag_guard = Arc::clone(&cancel_flag);

    let (tx, rx) = channel::bounded(1);

    scope.spawn(move || {
        let _guard = TimeoutGuard {
            cancel_flag: cancel_flag_guard,
        };
        let result = operation(cancel_flag_thread);
        let _ = tx.send(result);
    });

    match rx.recv_timeout(effective_timeout) {
        Ok(result) => result,
        Err(channel::RecvTimeoutError::Timeout) => {
            cancel_flag.store(true, Ordering::Release);
            eprintln!("Warning: Plugin operation timed out after {timeout:?}");
            Err(TimeoutError::Timeout(timeout).into())
        }
        Err(channel::RecvTimeoutError::Disconnected) => {
            eprintln!("Warning: Plugin thread panicked during execution");
            Err(TimeoutError::PluginPanic.into())
        }
    }
}

/// Scoped variant of [`run_with_timeout_and_cancel`] — operation may borrow non-`'static` data.
///
/// See [`run_with_timeout_scoped`] for the motivation. Adds the same external-cancellation
/// monitor thread as [`run_with_timeout_and_cancel`].
///
/// # Errors
///
/// Same as [`run_with_timeout_and_cancel`].
pub(crate) fn run_with_timeout_and_cancel_scoped<'scope, 'env, F, T>(
    scope: &'scope std::thread::Scope<'scope, 'env>,
    timeout: Duration,
    cancel_token: Option<Arc<dyn CancellationToken>>,
    operation: F,
) -> Result<T>
where
    F: FnOnce(Arc<AtomicBool>) -> Result<T> + Send + 'scope,
    T: Send + 'scope,
{
    if let Some(ref token) = cancel_token {
        if token.is_cancelled() {
            return Err(crate::error::CrushError::Cancelled);
        }
    }

    let effective_timeout = if timeout == Duration::from_secs(0) {
        Duration::MAX
    } else {
        timeout
    };

    let cancel_flag = Arc::new(AtomicBool::new(false));
    let cancel_flag_thread = Arc::clone(&cancel_flag);
    let cancel_flag_guard = Arc::clone(&cancel_flag);
    let cancel_flag_monitor = Arc::clone(&cancel_flag);

    let monitor_handle = cancel_token.map(|token| {
        scope.spawn(move || {
            while !cancel_flag_monitor.load(Ordering::Acquire) {
                if token.is_cancelled() {
                    cancel_flag_monitor.store(true, Ordering::Release);
                    break;
                }
                std::thread::sleep(Duration::from_micros(100));
            }
        })
    });

    let (tx, rx) = channel::bounded(1);

    scope.spawn(move || {
        let _guard = TimeoutGuard {
            cancel_flag: cancel_flag_guard,
        };
        let result = operation(cancel_flag_thread);
        let _ = tx.send(result);
    });

    let result = match rx.recv_timeout(effective_timeout) {
        Ok(result) => match result {
            Err(crate::error::CrushError::Plugin(crate::error::PluginError::Cancelled)) => {
                Err(crate::error::CrushError::Cancelled)
            }
            other => other,
        },
        Err(channel::RecvTimeoutError::Timeout) => {
            cancel_flag.store(true, Ordering::Release);
            eprintln!("Warning: Plugin operation timed out after {timeout:?}");
            Err(TimeoutError::Timeout(timeout).into())
        }
        Err(channel::RecvTimeoutError::Disconnected) => {
            eprintln!("Warning: Plugin thread panicked during execution");
            Err(TimeoutError::PluginPanic.into())
        }
    };

    cancel_flag.store(true, Ordering::Release);
    if let Some(handle) = monitor_handle {
        let _ = handle.join();
    }

    result
}

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

    #[test]
    #[allow(clippy::unwrap_used)]
    fn test_operation_completes_within_timeout() {
        let timeout = Duration::from_secs(1);

        let result = run_with_timeout(timeout, |_cancel| {
            // Fast operation
            Ok(42)
        });

        assert!(result.is_ok());
        assert_eq!(result.unwrap(), 42);
    }

    #[test]
    fn test_operation_respects_cancellation() {
        let timeout = Duration::from_millis(50);

        let result = run_with_timeout(timeout, |cancel_flag| {
            // Simulate slow operation that checks cancellation
            for _ in 0..1000 {
                if cancel_flag.load(Ordering::Acquire) {
                    return Err(PluginError::Cancelled.into());
                }
                std::thread::sleep(Duration::from_millis(10));
            }
            Ok(42)
        });

        // Should either timeout or be cancelled
        assert!(result.is_err());
    }

    #[test]
    #[allow(clippy::unwrap_used)]
    fn test_zero_timeout_means_no_timeout() {
        let timeout = Duration::from_secs(0);

        let result = run_with_timeout(timeout, |_cancel| Ok(42));

        // Zero timeout means no timeout - operation should succeed
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), 42);
    }

    #[test]
    fn test_timeout_guard_sets_flag_on_drop() {
        let cancel_flag = Arc::new(AtomicBool::new(false));
        {
            let _guard = TimeoutGuard {
                cancel_flag: Arc::clone(&cancel_flag),
            };
            assert!(!cancel_flag.load(Ordering::Acquire));
        }
        // Flag should be set after guard is dropped
        assert!(cancel_flag.load(Ordering::Acquire));
    }

    #[test]
    #[allow(clippy::unwrap_used)]
    fn test_run_with_timeout_basic_success() {
        let timeout = Duration::from_secs(1);

        let result = run_with_timeout(timeout, |_cancel| Ok(100));

        assert!(result.is_ok());
        assert_eq!(result.unwrap(), 100);
    }

    #[test]
    #[allow(clippy::unwrap_used)]
    fn test_run_with_timeout_operation_error() {
        let timeout = Duration::from_secs(1);

        let result: Result<i32> = run_with_timeout(timeout, |_cancel| {
            Err(PluginError::OperationFailed("test error".to_string()).into())
        });

        assert!(result.is_err());
        let err_msg = result.unwrap_err().to_string();
        assert!(err_msg.contains("test error"));
    }

    #[test]
    fn test_timeout_error_display() {
        let timeout_err = TimeoutError::Timeout(Duration::from_secs(30));
        assert!(timeout_err.to_string().contains("30"));

        let panic_err = TimeoutError::PluginPanic;
        assert!(panic_err.to_string().contains("panicked"));
    }

    #[test]
    #[allow(clippy::unwrap_used)]
    fn test_run_with_timeout_error_propagation() {
        let timeout = Duration::from_secs(1);

        let result: Result<i32> = run_with_timeout(timeout, |_cancel| {
            Err(PluginError::OperationFailed("custom error".to_string()).into())
        });

        assert!(result.is_err());
        let err_msg = result.unwrap_err().to_string();
        assert!(err_msg.contains("custom error"));
    }

    #[test]
    #[allow(clippy::unwrap_used)]
    fn test_effective_timeout_conversion() {
        // Test that 0 timeout becomes Duration::MAX internally
        let timeout = Duration::from_secs(0);

        // This should complete successfully even with "infinite" effective timeout
        let result = run_with_timeout(timeout, |_cancel| {
            std::thread::sleep(Duration::from_millis(10));
            Ok("done".to_string())
        });

        assert!(result.is_ok());
        assert_eq!(result.unwrap(), "done");
    }
}